rpm/thrift
SHA256
1
0
Fork 0
forked from pool/thrift
Code Pull Requests Activity

Accepting request 288551 from home:pluskalm:branches:devel:tools

Browse Source 
- Disable building of static library
- Do not manually set LDFLAGS

- Remove obsolete patches
  * 0001-Add-missing-limits-header.patch  
  * 0002-TNonblockingServer-TLibEventTransport.patch  
  * 0003-TDenseProtocol.patch
- Update library name to match shlib naming policy
- Update to 0.9.2
  * numerous changes, see provide CHANGES for details

OBS-URL: https://build.opensuse.org/request/show/288551
OBS-URL: https://build.opensuse.org/package/show/devel:tools/thrift?expand=0&rev=6
This commit is contained in:
Martin Pluskal 2015-03-03 22:11:19 +00:00 committed by Git OBS Bridge
parent 0dd6172b17
commit 9dc907fd61
9 changed files with 106 additions and 1066 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
0001-Add-missing-limits-header.patch
View File

@ -1,36 +0,0 @@
Index: thrift-0.9.0/lib/cpp/src/thrift/protocol/TDebugProtocol.cpp
===================================================================
--- thrift-0.9.0.orig/lib/cpp/src/thrift/protocol/TDebugProtocol.cpp 2012-12-02 07:53:16.470513572 -0600
+++ thrift-0.9.0/lib/cpp/src/thrift/protocol/TDebugProtocol.cpp 2012-12-02 07:54:47.362509517 -0600
@@ -23,6 +23,7 @@
#include <cctype>
#include <cstdio>
#include <stdexcept>
+#include <limits>
#include <boost/static_assert.hpp>
#include <boost/lexical_cast.hpp>
Index: thrift-0.9.0/lib/cpp/src/thrift/protocol/TDenseProtocol.cpp
===================================================================
--- thrift-0.9.0.orig/lib/cpp/src/thrift/protocol/TDenseProtocol.cpp 2012-12-02 07:52:51.706514677 -0600
+++ thrift-0.9.0/lib/cpp/src/thrift/protocol/TDenseProtocol.cpp 2012-12-02 07:54:06.918511327 -0600
@@ -89,6 +89,7 @@
#define __STDC_LIMIT_MACROS
#include <stdint.h>
+#include <limits>
#include <thrift/protocol/TDenseProtocol.h>
#include <thrift/TReflectionLocal.h>
Index: thrift-0.9.1/lib/cpp/src/thrift/protocol/TJSONProtocol.cpp
===================================================================
--- thrift-0.9.1.orig/lib/cpp/src/thrift/protocol/TJSONProtocol.cpp 2012-12-02 07:53:04.218514119 -0600
+++ thrift-0.9.1/lib/cpp/src/thrift/protocol/TJSONProtocol.cpp 2012-12-02 07:54:24.122510554 -0600
@@ -20,6 +20,7 @@
#include <thrift/protocol/TJSONProtocol.h>
#include <math.h>
+#include <limits>
#include <boost/lexical_cast.hpp>
#include <thrift/protocol/TBase64Utils.h>
#include <thrift/transport/TTransportException.h>

967
0002-TNonblockingServer-TLibEventTransport.patch
View File

@ -1,967 +0,0 @@
Index: thrift-0.9.1/lib/cpp/Makefile.am
===================================================================
--- thrift-0.9.1.orig/lib/cpp/Makefile.am 2012-10-12 02:58:06.000000000 +0200
+++ thrift-0.9.1/lib/cpp/Makefile.am 2013-08-15 16:53:40.000000000 +0200
@@ -191,7 +191,8 @@
src/thrift/transport/TTransportUtils.h \
src/thrift/transport/TBufferTransports.h \
src/thrift/transport/TShortReadTransport.h \
- src/thrift/transport/TZlibTransport.h
+ src/thrift/transport/TZlibTransport.h \
+ src/thrift/transport/TLibEventTransport.h
include_serverdir = $(include_thriftdir)/server
include_server_HEADERS = \
Index: thrift-0.9.1/lib/cpp/Makefile.in
===================================================================
--- thrift-0.9.1.orig/lib/cpp/Makefile.in 2012-10-12 02:59:49.000000000 +0200
+++ thrift-0.9.1/lib/cpp/Makefile.in 2013-08-15 16:53:40.000000000 +0200
@@ -627,7 +627,8 @@
src/thrift/transport/TTransportUtils.h \
src/thrift/transport/TBufferTransports.h \
src/thrift/transport/TShortReadTransport.h \
- src/thrift/transport/TZlibTransport.h
+ src/thrift/transport/TZlibTransport.h \
+ src/thrift/transport/TLibEventTransport.h
include_serverdir = $(include_thriftdir)/server
include_server_HEADERS = \
Index: thrift-0.9.1/lib/cpp/src/thrift/server/TNonblockingServer.cpp
===================================================================
--- thrift-0.9.1.orig/lib/cpp/src/thrift/server/TNonblockingServer.cpp 2012-10-12 02:58:06.000000000 +0200
+++ thrift-0.9.1/lib/cpp/src/thrift/server/TNonblockingServer.cpp 2013-08-15 17:11:07.000000000 +0200
@@ -23,11 +23,14 @@
#include <thrift/server/TNonblockingServer.h>
#include <thrift/concurrency/Exception.h>
-#include <thrift/transport/TSocket.h>
+#include <thrift/transport/TLibEventTransport.h>
#include <thrift/concurrency/PlatformThreadFactory.h>
#include <thrift/transport/PlatformSocket.h>
+#include <boost/bind.hpp>
+
#include <iostream>
+#include <queue>
#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
@@ -75,13 +78,6 @@
using apache::thrift::transport::TTransportException;
using boost::shared_ptr;
-/// Three states for sockets: recv frame size, recv data, and send mode
-enum TSocketState {
- SOCKET_RECV_FRAMING,
- SOCKET_RECV,
- SOCKET_SEND
-};
-
/**
* Five states for the nonblocking server:
* 1) initialize
@@ -92,7 +88,6 @@
*/
enum TAppState {
APP_INIT,
- APP_READ_FRAME_SIZE,
APP_READ_REQUEST,
APP_WAIT_TASK,
APP_SEND_RESULT,
@@ -115,7 +110,7 @@
boost::shared_ptr<TProcessor> processor_;
/// Object wrapping network socket
- boost::shared_ptr<TSocket> tSocket_;
+ boost::shared_ptr<TLibEventTransport> tSocket_;
/// Libevent object
struct event event_;
@@ -123,8 +118,8 @@
/// Libevent flags
short eventFlags_;
- /// Socket mode
- TSocketState socketState_;
+ /// Tells whether the frame size was read completely.
+ bool readFrameSize_;
/// Application state
TAppState appState_;
@@ -141,15 +136,12 @@
/// Read buffer size
uint32_t readBufferSize_;
- /// Write buffer
- uint8_t* writeBuffer_;
-
- /// Write buffer size
- uint32_t writeBufferSize_;
-
/// How far through writing are we?
uint32_t writeBufferPos_;
+ /// Write position for external events
+ uint externalWriteBufferPos_;
+
/// Largest size of write buffer seen since buffer was constructed
size_t largestWriteBufferSize_;
@@ -168,6 +160,9 @@
/// Transport that processor writes to
boost::shared_ptr<TMemoryBuffer> outputTransport_;
+ /// Transport that processes writes from an external libevent source.
+ boost::shared_ptr<TMemoryBuffer> externalTransport_;
+
/// extra transport generated by transport factory (e.g. BufferedRouterTransport)
boost::shared_ptr<TTransport> factoryInputTransport_;
boost::shared_ptr<TTransport> factoryOutputTransport_;
@@ -184,6 +179,16 @@
/// Thrift call context, if any
void *connectionContext_;
+ /// Node for queuing write requests from internal and external sources.
+ typedef std::pair<bool, uint32_t> WriteRequest;
+
+ /// Queue with write requests from outputTransport_ and externalTransport_.
+ std::queue<WriteRequest> writeRequests_;
+
+ /// Queues a write request for internal or external buffer and notifies
+ /// libevent.
+ void scheduleWrite(bool externalBuffer, uint32_t len);
+
/// Go into read mode
void setRead() {
setFlags(EV_READ | EV_PERSIST);
@@ -194,6 +199,11 @@
setFlags(EV_WRITE | EV_PERSIST);
}
+ /// Resets write mode after all write requests were send.
+ void clearWrite() {
+ clearFlags(EV_WRITE);
+ }
+
/// Set socket idle
void setIdle() {
setFlags(0);
@@ -207,12 +217,37 @@
void setFlags(short eventFlags);
/**
+ * Clear event flags for this connection.
+ *
+ * @param eventFlags flags we pass to libevent for the connection.
+ */
+ void clearFlags(short eventFlags);
+
+ /**
* Libevent handler called (via our static wrapper) when the connection
* socket had something happen. Rather than use the flags libevent passed,
* we use the connection state to determine whether we need to read or
* write the socket.
*/
- void workSocket();
+ void workSocket(short ev_type);
+
+ /**
+ * Reads frame size and message from socket.
+ */
+ bool readSocket();
+
+ /**
+ * Writes as many queued requests as possible
+ */
+ bool writeSocket();
+
+ /**
+ * Sets the frame size for a message that was written into a TMemoryBuffer.
+ *
+ * @param buf buffer including a new message at the end
+ * @param msgSize Size of the message including frame header
+ */
+ void setFrameSize(TMemoryBuffer& buf, uint32_t msgSize);
public:
@@ -232,7 +267,16 @@
inputTransport_.reset(new TMemoryBuffer(readBuffer_, readBufferSize_));
outputTransport_.reset(
new TMemoryBuffer(static_cast<uint32_t>(server_->getWriteBufferDefaultSize())));
- tSocket_.reset(new TSocket());
+ externalTransport_.reset(
+ new TMemoryBuffer(static_cast<uint32_t>(server_->getWriteBufferDefaultSize())));
+
+ boost::shared_ptr<TProtocol> externalProtocol =
+ server_->getOutputProtocolFactory()->getProtocol(externalTransport_);
+
+ tSocket_.reset(new TLibEventTransport(
+ externalProtocol,
+ boost::bind(&TConnection::beginExternalWriteEvent, this),
+ boost::bind(&TConnection::endExternalWriteEvent, this)));
init(socket, ioThread, addr, addrLen);
}
@@ -243,6 +287,18 @@
/// Close this connection and free or reset its resources.
void close();
+ /**
+ * Callback for TLibEventTransport to write 4 bytes for the frame header
+ * into externalTransport_.
+ */
+ void beginExternalWriteEvent();
+
+ /**
+ * Callback for TLibEventTransport to calculate the frame size and notifiy
+ * libevent for writting.
+ */
+ void endExternalWriteEvent();
+
/**
* Check buffers against any size limits and shrink it if exceeded.
*
@@ -266,13 +322,13 @@
* C-callable event handler for connection events. Provides a callback
* that libevent can understand which invokes connection_->workSocket().
*
- * @param fd the descriptor the event occurred on.
- * @param which the flags associated with the event.
- * @param v void* callback arg where we placed TConnection's "this".
+ * @param fd the descriptor the event occurred on.
+ * @param ev_type the flags associated with the event.
+ * @param v void* callback arg where we placed TConnection's "this".
*/
- static void eventHandler(evutil_socket_t fd, short /* which */, void* v) {
+ static void eventHandler(evutil_socket_t fd, short ev_type, void* v) {
assert(fd == ((TConnection*)v)->getTSocket()->getSocketFD());
- ((TConnection*)v)->workSocket();
+ ((TConnection*)v)->workSocket(ev_type);
}
/**
@@ -393,6 +449,7 @@
socklen_t addrLen) {
tSocket_->setSocketFD(socket);
tSocket_->setCachedAddress(addr, addrLen);
+ tSocket_->setEventBase(*(ioThread->getEventBase()));
ioThread_ = ioThread;
server_ = ioThread->getServer();
@@ -402,12 +459,9 @@
readBufferPos_ = 0;
readWant_ = 0;
- writeBuffer_ = NULL;
- writeBufferSize_ = 0;
- writeBufferPos_ = 0;
largestWriteBufferSize_ = 0;
- socketState_ = SOCKET_RECV_FRAMING;
+ readFrameSize_ = true;
callsForResize_ = 0;
// get input/transports
@@ -435,12 +489,29 @@
processor_ = server_->getProcessor(inputProtocol_, outputProtocol_, tSocket_);
}
-void TNonblockingServer::TConnection::workSocket() {
- int got=0, left=0, sent=0;
+void TNonblockingServer::TConnection::workSocket(short ev_type) {
+
+ if (ev_type & EV_READ) {
+ if (!readSocket()) {
+ return;
+ }
+ }
+
+ if (ev_type & EV_WRITE) {
+ if (!writeSocket()) {
+ return;
+ }
+ }
+
+ if (ev_type & ~(EV_READ | EV_WRITE))
+ GlobalOutput.printf("TConnection::workSocket: unexpected event type %d",
+ ev_type & ~(EV_READ | EV_WRITE));
+}
+
+bool TNonblockingServer::TConnection::readSocket() {
uint32_t fetch = 0;
- switch (socketState_) {
- case SOCKET_RECV_FRAMING:
+ if (readFrameSize_) {
union {
uint8_t buf[sizeof(uint32_t)];
uint32_t size;
@@ -456,20 +527,22 @@
if (fetch == 0) {
// Whenever we get here it means a remote disconnect
close();
- return;
+
+ return false;
}
readBufferPos_ += fetch;
- } catch (TTransportException& te) {
- GlobalOutput.printf("TConnection::workSocket(): %s", te.what());
+ } catch(TTransportException& te) {
+ GlobalOutput.printf("TConnection::readSocket(): %s", te.what());
close();
- return;
+ return false;
}
if (readBufferPos_ < sizeof(framing.size)) {
// more needed before frame size is known -- save what we have so far
readWant_ = framing.size;
- return;
+
+ return true;
}
readWant_ = ntohl(framing.size);
@@ -483,84 +556,154 @@
(uint64_t)server_->getMaxFrameSize(),
tSocket_->getSocketInfo().c_str());
close();
- return;
- }
- // size known; now get the rest of the frame
- transition();
- return;
-
- case SOCKET_RECV:
- // It is an error to be in this state if we already have all the data
- assert(readBufferPos_ < readWant_);
- try {
- // Read from the socket
- fetch = readWant_ - readBufferPos_;
- got = tSocket_->read(readBuffer_ + readBufferPos_, fetch);
+ return false;
}
- catch (TTransportException& te) {
- GlobalOutput.printf("TConnection::workSocket(): %s", te.what());
- close();
- return;
- }
+ readFrameSize_ = false;
- if (got > 0) {
- // Move along in the buffer
- readBufferPos_ += got;
-
- // Check that we did not overdo it
- assert(readBufferPos_ <= readWant_);
-
- // We are done reading, move onto the next state
- if (readBufferPos_ == readWant_) {
- transition();
+ // We just read the request length
+ // Double the buffer size until it is big enough
+ if (readWant_ > readBufferSize_) {
+ if (readBufferSize_ == 0) {
+ readBufferSize_ = 1;
}
- return;
+ uint32_t newSize = readBufferSize_;
+ while (readWant_ > newSize) {
+ newSize *= 2;
+ }
+
+ uint8_t* newBuffer = (uint8_t*)std::realloc(readBuffer_, newSize);
+ if (newBuffer == NULL) {
+ // nothing else to be done...
+ throw std::bad_alloc();
+ }
+ readBuffer_ = newBuffer;
+ readBufferSize_ = newSize;
}
- // Whenever we get down here it means a remote disconnect
+ readBufferPos_= 0;
+ }
+
+ // size known; now get the rest of the frame
+
+ // It is an error to be in this state if we already have all the data
+ assert(readBufferPos_ < readWant_);
+
+ uint32_t got=0;
+
+ try {
+ // Read from the socket
+ fetch = readWant_ - readBufferPos_;
+ got = tSocket_->read(readBuffer_ + readBufferPos_, fetch);
+ }
+ catch (TTransportException& te) {
+ GlobalOutput.printf("TConnection::readSocket(): %s", te.what());
close();
- return;
+ return false;
+ }
- case SOCKET_SEND:
- // Should never have position past size
- assert(writeBufferPos_ <= writeBufferSize_);
+ if (got > 0) {
+ // Move along in the buffer
+ readBufferPos_ += got;
+
+ // Check that we did not overdo it
+ assert(readBufferPos_ <= readWant_);
+
+ // We are done reading, move onto the next state
+ if (readBufferPos_ == readWant_) {
+ transition();
+ }
+
+ return true;
+ }
+
+ // Whenever we get down here it means a remote disconnect
+ close();
+
+ return false;
+}
+
+bool TNonblockingServer::TConnection::writeSocket() {
+ while(writeRequests_.empty() == false) {
+ WriteRequest& writeReq = writeRequests_.front();
+
+ TMemoryBuffer* transport = writeReq.first ? externalTransport_.get() :
+ outputTransport_.get();
+
+ uint8_t* writeBuffer;
+ uint32_t writeBufferSize;
+
+ transport->getBuffer(&writeBuffer, &writeBufferSize);
+ transport->borrow(0, &writeBufferSize);
+
+ uint32_t writeSize = writeReq.second;
+ if(writeBufferSize < writeReq.second) {
+ GlobalOutput.printf("WARNING: Write request size %u > %u buffer size.\n",
+ writeReq.second, writeBufferSize);
+ writeSize = writeBufferSize;
+ }
// If there is no data to send, then let us move on
- if (writeBufferPos_ == writeBufferSize_) {
+ if (writeBufferSize == 0) {
GlobalOutput("WARNING: Send state with no data to send\n");
- transition();
- return;
+ writeRequests_.pop();
+ continue;
+ }
+
+ if (writeBufferSize > largestWriteBufferSize_) {
+ largestWriteBufferSize_ = writeBufferSize;
}
+ uint32_t sent = 0;
+
try {
- left = writeBufferSize_ - writeBufferPos_;
- sent = tSocket_->write_partial(writeBuffer_ + writeBufferPos_, left);
+ sent = tSocket_->write_partial(writeBuffer, writeSize);
}
- catch (TTransportException& te) {
- GlobalOutput.printf("TConnection::workSocket(): %s ", te.what());
+ catch(TTransportException& te) {
+ GlobalOutput.printf("TConnection::writeSocket(): %s ", te.what());
close();
- return;
+
+ return false;
}
- writeBufferPos_ += sent;
+ // We are done!
+ if (sent == writeBufferSize) {
+ transport->consume(sent);
+ writeRequests_.pop();
+ }
+ else if (sent > 0) {
+ // Message was not written completely.
+ transport->consume(sent);
- // Did we overdo it?
- assert(writeBufferPos_ <= writeBufferSize_);
+ // Adjust size in write request.
+ writeReq.second -= sent;
- // We are done!
- if (writeBufferPos_ == writeBufferSize_) {
- transition();
+ return true;
}
+ }
- return;
+ // All queued requests were written, clear write flag.
+ clearWrite();
- default:
- GlobalOutput.printf("Unexpected Socket State %d", socketState_);
- assert(0);
+ // Verify that there's no incomplete message in the output buffers.
+ if(outputTransport_->peek() || externalTransport_->peek())
+ return true;
+
+ // it's now safe to perform buffer size housekeeping.
+ if (server_->getResizeBufferEveryN() > 0
+ && ++callsForResize_ >= server_->getResizeBufferEveryN()) {
+ checkIdleBufferMemLimit(server_->getIdleReadBufferLimit(),
+ server_->getIdleWriteBufferLimit());
+ callsForResize_ = 0;
}
+
+ // Reset write buffer position to the beginning of the buffer.
+ outputTransport_->resetBuffer();
+ externalTransport_->resetBuffer();
+
+ return true;
}
/**
@@ -580,9 +723,9 @@
// We are done reading the request, package the read buffer into transport
// and get back some data from the dispatch function
inputTransport_->resetBuffer(readBuffer_, readBufferPos_);
- outputTransport_->resetBuffer();
// Prepend four bytes of blank space to the buffer so we can
// write the frame size there later.
+ writeBufferPos_ = outputTransport_->writeEnd();
outputTransport_->getWritePtr(4);
outputTransport_->wroteBytes(4);
@@ -646,107 +789,36 @@
// the writeBuffer_
case APP_WAIT_TASK:
+ {
// We have now finished processing a task and the result has been written
// into the outputTransport_, so we grab its contents and place them into
// the writeBuffer_ for actual writing by the libevent thread
server_->decrementActiveProcessors();
- // Get the result of the operation
- outputTransport_->getBuffer(&writeBuffer_, &writeBufferSize_);
- // If the function call generated return data, then move into the send
- // state and get going
- // 4 bytes were reserved for frame size
- if (writeBufferSize_ > 4) {
-
- // Move into write state
- writeBufferPos_ = 0;
- socketState_ = SOCKET_SEND;
-
- // Put the frame size into the write buffer
- int32_t frameSize = (int32_t)htonl(writeBufferSize_ - 4);
- memcpy(writeBuffer_, &frameSize, 4);
-
- // Socket into write mode
- appState_ = APP_SEND_RESULT;
- setWrite();
-
- // Try to work the socket immediately
- // workSocket();
-
- return;
- }
-
- // In this case, the request was oneway and we should fall through
- // right back into the read frame header state
- goto LABEL_APP_INIT;
-
- case APP_SEND_RESULT:
- // it's now safe to perform buffer size housekeeping.
- if (writeBufferSize_ > largestWriteBufferSize_) {
- largestWriteBufferSize_ = writeBufferSize_;
- }
- if (server_->getResizeBufferEveryN() > 0
- && ++callsForResize_ >= server_->getResizeBufferEveryN()) {
- checkIdleBufferMemLimit(server_->getIdleReadBufferLimit(),
- server_->getIdleWriteBufferLimit());
- callsForResize_ = 0;
+ // If the processor did not write a response we have to remove again
+ // the 4 bytes from the frame size.
+ uint32_t writePos = outputTransport_->writeEnd();
+ if (writePos == writeBufferPos_ + 4)
+ outputTransport_->revertLastWrite(4);
+ else {
+ uint32_t size = writePos - writeBufferPos_;
+ setFrameSize(*outputTransport_, size);
+ scheduleWrite(false, size);
}
// N.B.: We also intentionally fall through here into the INIT state!
-
- LABEL_APP_INIT:
+ }
case APP_INIT:
- // Clear write buffer variables
- writeBuffer_ = NULL;
- writeBufferPos_ = 0;
- writeBufferSize_ = 0;
-
// Into read4 state we go
- socketState_ = SOCKET_RECV_FRAMING;
- appState_ = APP_READ_FRAME_SIZE;
+ readFrameSize_ = true;
+ appState_ = APP_READ_REQUEST;
readBufferPos_ = 0;
// Register read event
setRead();
-
- // Try to work the socket right away
- // workSocket();
-
- return;
-
- case APP_READ_FRAME_SIZE:
- // We just read the request length
- // Double the buffer size until it is big enough
- if (readWant_ > readBufferSize_) {
- if (readBufferSize_ == 0) {
- readBufferSize_ = 1;
- }
- uint32_t newSize = readBufferSize_;
- while (readWant_ > newSize) {
- newSize *= 2;
- }
-
- uint8_t* newBuffer = (uint8_t*)std::realloc(readBuffer_, newSize);
- if (newBuffer == NULL) {
- // nothing else to be done...
- throw std::bad_alloc();
- }
- readBuffer_ = newBuffer;
- readBufferSize_ = newSize;
- }
-
- readBufferPos_= 0;
-
- // Move into read request state
- socketState_ = SOCKET_RECV;
- appState_ = APP_READ_REQUEST;
-
- // Work the socket right away
- // workSocket();
-
return;
case APP_CLOSE_CONNECTION:
@@ -760,9 +832,59 @@
}
}
+void TNonblockingServer::TConnection::beginExternalWriteEvent() {
+ // Prepend four bytes of blank space to the buffer so we can
+ // write the frame size there later.
+ externalWriteBufferPos_ = externalTransport_->writeEnd();
+ externalTransport_->getWritePtr(4);
+ externalTransport_->wroteBytes(4);
+}
+
+void TNonblockingServer::TConnection::endExternalWriteEvent() {
+ // If the external function did not write anything we have to remove again
+ // the 4 bytes from the frame size.
+ uint32_t writePos = externalTransport_->writeEnd();
+
+ if (writePos == externalWriteBufferPos_ + 4)
+ externalTransport_->revertLastWrite(4);
+ else {
+ uint32_t size = writePos - externalWriteBufferPos_;
+ setFrameSize(*externalTransport_, size);
+
+ scheduleWrite(true, size);
+ }
+}
+
+void TNonblockingServer::TConnection::setFrameSize(
+ TMemoryBuffer& buf, uint32_t size) {
+ uint8_t* begin = buf.getWritePtr(0) - size;
+
+ // Put the frame size into the write buffer (subtract 4 bytes from
+ // for frame size header).
+ int32_t frameSize = (int32_t)htonl(size - 4);
+ memcpy(begin, &frameSize, 4);
+}
+
+void TNonblockingServer::TConnection::scheduleWrite(
+ bool externalBuffer, uint32_t len) {
+ if( writeRequests_.empty() ) {
+ writeRequests_.push(WriteRequest(externalBuffer, len));
+ setWrite();
+ return;
+ }
+
+ WriteRequest& req = writeRequests_.back();
+
+ // We can extend the last write request if the buffer is the same.
+ if(externalBuffer == req.first)
+ req.second += len;
+ else
+ writeRequests_.push(WriteRequest(externalBuffer, len));
+}
+
void TNonblockingServer::TConnection::setFlags(short eventFlags) {
// Catch the do nothing case
- if (eventFlags_ == eventFlags) {
+ if ((eventFlags_ | eventFlags) == eventFlags_) {
return;
}
@@ -775,7 +897,7 @@
}
// Update in memory structure
- eventFlags_ = eventFlags;
+ eventFlags_ |= eventFlags;
// Do not call event_set if there are no flags
if (!eventFlags_) {
@@ -819,10 +941,43 @@
}
}
+void TNonblockingServer::TConnection::clearFlags(short eventFlags) {
+ if ((~eventFlags & eventFlags_) == eventFlags_)
+ return;
+
+ short newFlags = ~eventFlags & eventFlags_;
+
+ eventFlags_ = 0;
+ if (event_del(&event_) == -1) {
+ GlobalOutput("TConnection::setFlags event_del");
+ return;
+ }
+
+ setFlags(newFlags);
+}
+
/**
* Closes a connection
*/
void TNonblockingServer::TConnection::close() {
+ // Close the socket
+ tSocket_->close();
+
+ // close any factory produced transports
+ factoryInputTransport_->close();
+ factoryOutputTransport_->close();
+
+ // Give this object back to the server that owns it
+ processor_.reset();
+
+ // Remove all queued write requests
+ while(writeRequests_.empty() == false)
+ writeRequests_.pop();
+
+ // Reset write buffer position to the beginning of the buffer.
+ outputTransport_->resetBuffer();
+ externalTransport_->resetBuffer();
+
// Delete the registered libevent
if (event_del(&event_) == -1) {
GlobalOutput.perror("TConnection::close() event_del", THRIFT_GET_SOCKET_ERROR);
@@ -831,16 +986,9 @@
if (serverEventHandler_) {
serverEventHandler_->deleteContext(connectionContext_, inputProtocol_, outputProtocol_);
}
- ioThread_ = NULL;
-
- // Close the socket
- tSocket_->close();
- // close any factory produced transports
- factoryInputTransport_->close();
- factoryOutputTransport_->close();
+ ioThread_ = NULL;
- // Give this object back to the server that owns it
server_->returnConnection(this);
}
@@ -856,6 +1004,7 @@
if (writeLimit > 0 && largestWriteBufferSize_ > writeLimit) {
// just start over
outputTransport_->resetBuffer(static_cast<uint32_t>(server_->getWriteBufferDefaultSize()));
+ externalTransport_->resetBuffer(static_cast<uint32_t>(server_->getWriteBufferDefaultSize()));
largestWriteBufferSize_ = 0;
}
}
@@ -1131,7 +1280,7 @@
void TNonblockingServer::setThreadManager(boost::shared_ptr<ThreadManager> threadManager) {
threadManager_ = threadManager;
if (threadManager) {
- threadManager->setExpireCallback(apache::thrift::stdcxx::bind(&TNonblockingServer::expireClose, this, apache::thrift::stdcxx::placeholders::_1));
+ threadManager->setExpireCallback(boost::bind(&TNonblockingServer::expireClose, this, _1));
threadPoolProcessing_ = true;
} else {
threadPoolProcessing_ = false;
Index: thrift-0.9.1/lib/cpp/src/thrift/transport/TBufferTransports.cpp
===================================================================
--- thrift-0.9.1.orig/lib/cpp/src/thrift/transport/TBufferTransports.cpp 2012-10-12 02:58:06.000000000 +0200
+++ thrift-0.9.1/lib/cpp/src/thrift/transport/TBufferTransports.cpp 2013-08-15 16:53:40.000000000 +0200
@@ -378,6 +378,14 @@
wBase_ += len;
}
+void TMemoryBuffer::revertLastWrite(uint32_t len)
+{
+ if (wBase_ - len < rBase_)
+ throw TTransportException("Can't reset write pointer before read pointer");
+
+ wBase_ -= len;
+}
+
const uint8_t* TMemoryBuffer::borrowSlow(uint8_t* buf, uint32_t* len) {
(void) buf;
rBound_ = wBase_;
Index: thrift-0.9.0/lib/cpp/src/thrift/transport/TBufferTransports.h
===================================================================
--- thrift-0.9.0.orig/lib/cpp/src/thrift/transport/TBufferTransports.h 2012-10-12 02:58:06.000000000 +0200
+++ thrift-0.9.0/lib/cpp/src/thrift/transport/TBufferTransports.h 2013-08-15 16:53:40.000000000 +0200
@@ -683,6 +683,9 @@
// that had been provided by getWritePtr().
void wroteBytes(uint32_t len);
+ // Resets the write pointer by len bytes if the were not yet read.
+ void revertLastWrite(uint32_t len);
+
/*
* TVirtualTransport provides a default implementation of readAll().
* We want to use the TBufferBase version instead.
Index: thrift-0.9.0/lib/cpp/src/thrift/transport/TLibEventTransport.h
===================================================================
--- /dev/null 1970-01-01 00:00:00.000000000 +0000
+++ thrift-0.9.0/lib/cpp/src/thrift/transport/TLibEventTransport.h 2013-08-15 16:53:40.000000000 +0200
@@ -0,0 +1,114 @@
+#ifndef _THRIFT_TRANSPORT_TLIB_EVENTTRANSPORT_H_
+#define _THRIFT_TRANSPORT_TLIB_EVENTTRANSPORT_H_ 1
+
+#include "TSocket.h"
+#include "TBufferTransports.h"
+
+#include <boost/function.hpp>
+#include <boost/shared_ptr.hpp>
+
+struct event_base;
+
+namespace apache { namespace thrift {
+namespace protocol {
+ class TProtocol;
+}
+namespace transport {
+
+/**
+ * TLibEventTransport is used by TNonblockingServer to integrate events from
+ * an external libevent source.
+ *
+ * To support asynchronous communication between a thrift client and server the
+ * following pattern is required:
+ * - all request in a Thrift service must be defined as oneway
+ * - an additional "callback" service is required to send asynchronous
+ * responses and updates from server to client. The "callback" service is
+ * implemented by the client. Here as well all request have to be oneway.
+ * Since all request are oneway they can share the same thrift connection.
+ * - In the server the callback client object must be initialized in the
+ * handler factory with the protocol from TConnectionInfo. For
+ * TLibEventTransport one has to use TLibEventTransport::getProtocol().
+ * - In the client one has to create an own thread to handle the callbacks.
+ * One only has to call process of the generated "callback" processor with
+ * the protocols created for the server connection.
+ *
+ * For an external event beginWriteEvent() must be called before using the
+ * the callback object to write 4 bytes for the frame size into the output
+ * transport. After using the callback object endWriteEvent() must be called
+ * to calculate and set the frame size and notify libevent for writting.
+ */
+class TLibEventTransport : public TSocket
+{
+public:
+ typedef boost::function<void ()> WriteEventCb;
+
+ /*!
+ * TLibEventTransport is created by TNonblockingServer when a new client
+ * connection is accepted.
+ *
+ * @param protocol protocol for the callback object
+ * @param beginEventCb callback to start sending an asynchronous event
+ * @param endEventCb callback to finish sending an asynchronous event
+ */
+ TLibEventTransport(boost::shared_ptr<protocol::TProtocol>& protocol,
+ WriteEventCb beginEventCb,
+ WriteEventCb endEventCb) :
+ eventBase_(0),
+ protocol_(protocol),
+ beginWriteEventCb_(beginEventCb),
+ endWriteEventCb_(endEventCb) {}
+
+ /**
+ * @returns event_base used by this transport.
+ */
+ event_base& getEventBase() const {
+ assert(eventBase_);
+ return *eventBase_;
+ }
+
+ /*!
+ * Setting correct event_base from TNonblockingServer.
+ *
+ * @param ev event_base used by TNonblockingServer
+ */
+ void setEventBase(event_base& ev) {
+ eventBase_ = &ev;
+ }
+
+ /**
+ * @returns the protocol for the callback object.
+ */
+ boost::shared_ptr<protocol::TProtocol>& getProtocol() {
+ return protocol_;
+ }
+
+ /**
+ * Writes 4 bytes for the frame header into the output transport.
+ *
+ * Must be called before using a request from the callback object.
+ */
+ void beginWriteEvent() {
+ beginWriteEventCb_();
+ }
+
+ /**
+ * Calculates the frame size and notifies libevent for writting.
+ *
+ * Must be called after using a request from the callback object.
+ */
+ void endWriteEvent() {
+ endWriteEventCb_();
+ }
+
+private:
+ event_base* eventBase_;
+ boost::shared_ptr<protocol::TProtocol> protocol_;
+ WriteEventCb beginWriteEventCb_;
+ WriteEventCb endWriteEventCb_;
+};
+
+}}} // apache::thrift::transport
+
+
+#endif // _THRIFT_TRANSPORT_TLIB_EVENTTRANSPORT_H_

22
0003-TDenseProtocol.patch
View File

@ -1,22 +0,0 @@
Index: thrift-0.9.0/lib/cpp/src/thrift/protocol/TDenseProtocol.cpp
===================================================================
--- thrift-0.9.0.orig/lib/cpp/src/thrift/protocol/TDenseProtocol.cpp
+++ thrift-0.9.0/lib/cpp/src/thrift/protocol/TDenseProtocol.cpp
@@ -304,7 +304,7 @@ uint32_t TDenseProtocol::writeStructBegi
// We need a new field index for this structure.
idx_stack_.push_back(0);
- return 0;
+ return xfer;
}
uint32_t TDenseProtocol::writeStructEnd() {
@@ -538,7 +538,7 @@ uint32_t TDenseProtocol::readStructBegin
// We need a new field index for this structure.
idx_stack_.push_back(0);
- return 0;
+ return xfer;
}
uint32_t TDenseProtocol::readStructEnd() {

3
thrift-0.9.1.tar.gz
View File

@ -1,3 +0,0 @@
version https://git-lfs.github.com/spec/v1
oid sha256:ac175080c8cac567b0331e394f23ac306472c071628396db2850cb00c41b0017
size 3402353

3
thrift-0.9.2.tar.gz Normal file
View File

@ -0,0 +1,3 @@
version https://git-lfs.github.com/spec/v1
oid sha256:cef50d3934c41db5fa7724440cc6f10a732e7a77fe979b98c23ce45725349570
size 8053046

7
thrift-0.9.2.tar.gz.asc Normal file
View File

@ -0,0 +1,7 @@
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1
iF4EABEIAAYFAlRZonEACgkQuwY2j2a3ePljGQD/SwxjjsWUFADajiK2VQxTugfl
0GLPrC0LugahRChg77ABAJVI4mgI1lkKVXajYNLYln29oUhPa+Zq0avZo8oPyxWd
=JvzH
-----END PGP SIGNATURE-----

17
thrift.changes
View File

@ -1,3 +1,20 @@
-------------------------------------------------------------------
Mon Mar 2 18:44:32 UTC 2015 - mpluskal@suse.com
- Disable building of static library
- Do not manually set LDFLAGS
-------------------------------------------------------------------
Mon Mar 2 14:23:16 UTC 2015 - mpluskal@suse.com
- Remove obsolete patches
* 0001-Add-missing-limits-header.patch
* 0002-TNonblockingServer-TLibEventTransport.patch
* 0003-TDenseProtocol.patch
- Update library name to match shlib naming policy
- Update to 0.9.2
* numerous changes, see provide CHANGES for details
-------------------------------------------------------------------
Sat Feb 22 20:05:26 UTC 2014 - opensuse@dstoecker.de

40
thrift.keyring Normal file
View File

@ -0,0 +1,40 @@
-----BEGIN PGP PUBLIC KEY BLOCK-----
Version: GnuPG v2
mQMuBE2mAQ8RCACgm0DOTToYRWzbuOqxAxDpviHEa3OKhX8H7TaOscwicXiaszkC
yEfvap5j3VaD8g3zYsKoBShtmyR3ceN5/kjA+XYJYGtkvR7V4oq5PJRc/2Z4KEf4
uBND9FDBwg+evZWbax0914v21DRCVsSYWKVL4+Fh0Gt7MMjSONr1dvA7p+buMdPK
LdeUY7G1J0fUwL2QoSnlcyQ7adSyaDSfRZpDXCQvuCVF6CotikEsUdS1eoGtJll0
cc4cTpENVguiMAP3IN++756Jt2DZ7ogGzuZQhzz2K8Hpff4X/NkjkpIEexPjG1So
iQLT+xYpIS3IXUqtSWP/ns0viI83S6EERWWnAQC2YIjyLIGxkPSHeQiXjdnY3Fwy
na435DpYvmdMXxcrbQgAhLfqLRkvu19nj41WHfiWbmMTfqzDeQqUusE1rjN5OS3j
MbcuNFWIIQ8C+7JYe4kQiUfYiraTmQmz3YCR8r2iUViVZHhypEagzHM1jV2TE/p9
XE6jDPHsxs8/80uuzweVTtFtF9jmNqkoXTGFQjMCcBFVx8nNvrWnRgTuiTDLJzfn
pjpipFKuGQHZ8237cDMfmgkteddpN/HArgPBknUg43gGiIi1EL2chH2mHFDrFdpe
KtbhOVk9xCHA00JVPpoQ2lryWMmDD08PY8e1/8UC0yRsDsiFXUh26WGLKZU0Mz8Q
g6urPlJxbn+cLAqf/g1wrR9B/nuoB/zbD1dh2C1thwf/RaMJXbnGoCha6LGXh7Nx
B85uVvlCW4vx1R23BbzdRu28c7IHObtUVICZLtgeekXYzupzkctGO8tM5FXH8G8J
b7CeMfUWKitz2qTlf+Kb5VxtVenEn6ZSAZS8oOmD4Hyjoe8atVZpXB1yhbq353vp
9Jx0ovYs9V+KeBU0qu1gYKovK/QUBkjCeTyyfqu7qbZ1sDPVduHNHL+MyGelcDVg
ar8t67kdAlYxZ7DE1Pr8PlIbkWQafM+zEd2kvQGbpIqbIPuJv7MFRRO+jdKn6EqQ
5u/eL4hkgDABf4p+pP8Di3o4pLOiFrsiFZ/QIYf0G95O8k3MdMpX3Spc9Xnx5rsk
B7QiSmFrZSBGYXJyZWxsIDxqZmFycmVsbEBhcGFjaGUub3JnPohGBBARAgAGBQJO
RCETAAoJENUYeCmaeyOYajEAoKmuYqHYkm3m0KvhxjAYVthbFuVnAKCEhtqXhfi2
L85c9SsiQm3DfSOzhoh6BBMRCAAiBQJNpgEPAhsDBgsJCAcDAgYVCAIJCgsEFgID
AQIeAQIXgAAKCRC7BjaPZrd4+bXkAP9++Sa/CXEOQ4BI/KCwTi0UI3vTDDWI6LmU
oGcR+D8MnwEAtLZlu4kwyJDXKFMeo1B9tyUsR9Z9uEEruL9MHZ7Jvmi5Ag0ETaYB
DxAIAKeznjLErDjImOJLFOrde8ycskDsI0VSe8W6ObZ/4QdZOrN4GSSTsml9oRRE
ibL5PvJtCtCKn/anN7dmgCkcshVLXt0kzCDnotRP9T0J3aAgl+O5sZA9IhIDMo/f
vsTImaXx8cUzZZvg4pyUi5cdmfmwpqSBok6es4V1GnkCajVWBZ0QnWxtlIVqn6RW
bYdM32Nud4/WK5EH+N8m7+nfrN2RPYZ/IBeB0327Eq/5ZztBu1Nv2aNjw5nGkj1j
pXSOfWAc1cQAyOS1UIn7q5HZMf0BHU3yuvSKfGuDftblH/6mN4JbxgKkThiYMbr0
Lw3w9cGdv8Wogam1QIpcP/c4qr8AAwcH/R5eRwBdizI41wT0mO6h5oY+fW1XApR2
+kX6LJEsSQy4xLOP0UofPxOFCkHAr7tKIwUVmxXaG3R3rt2D0hfl6XsEn6wpV52o
YYMes9QGHnAZY0XnuzCWzW4m4+GUtUw5iQV0s+MZt7NocOgfia6ticMx9+7EsiOb
GEHYrd0H9bLVQAJz6j0uVSS3ggVE9l6pG2KR1aU3/LO6dUoowiR4dIGppQOGUNmi
NE27Btkm+y0GfjRRRnaOugpmrjEz4Ynomw2yclhv36HkCM9dWD7F8UzwHY1Ok2AH
6PDLJo3bFq1lFx1/4IHMiAVR7udlTmZO+enuW3QDVYDIVJdKcNbp3GuIYQQYEQgA
CQUCTaYBDwIbDAAKCRC7BjaPZrd4+bnAAPwKaGBc2yrIDdxu9KYWuEPDLYTVkFz6
/TFHMtLz7zbyRAEAmlxId/Q7q4vYoVg1GXyWtMaVRXblIJmign/QfUX/XIg=
=SxJl
-----END PGP PUBLIC KEY BLOCK-----

77
thrift.spec
View File

@ -20,8 +20,8 @@
%define _with_c 1
%define _with_perl 1
%define _with_python 1
#%#define _with_java 1
#%#define _with_ruby 1
#%%define _with_java 0
#%%define _with_ruby 0
%define _with_qt 1
%bcond_with perl
@ -32,15 +32,14 @@
%bcond_with qt
Name: thrift
Version: 0.9.1
Version: 0.9.2
Release: 0
%define libversion 0_9_1
%define libversion 0_9_2
%define libname %{version}
Url: http://thrift.apache.org/
Source: http://apache.openmirror.de/thrift/0.9.1/%{name}-%{version}.tar.gz
Patch1: 0001-Add-missing-limits-header.patch
#Patch2: 0002-TNonblockingServer-TLibEventTransport.patch
#Patch3: 0003-TDenseProtocol.patch
Url: https://thrift.apache.org
Source0: https://www.apache.org/dist/thrift/%{version}/%{name}-%{version}.tar.gz
Source1: https://www.apache.org/dist/thrift/%{version}/%{name}-%{version}.tar.gz.asc
Source2: %{name}.keyring
Summary: Framework for scalable cross-language services development
License: Apache-2.0
Group: Development/Libraries/C and C++
@ -63,21 +62,25 @@ BuildRequires: glib2-devel
%if %{with java}
BuildRequires: java-devel
BuildRequires: ant
%endif
%if %{with ruby}
BuildRequires: ruby-devel
BuildRequires: rubygem(bundler)
%endif
%if %{with perl}
BuildRequires: perl
BuildRequires: perl(Bit::Vector)
BuildRequires: perl(Class::Accessor)
%endif
BuildRequires: bison
BuildRequires: flex
BuildRequires: libevent-devel
BuildRequires: openssl-devel
BuildRequires: automake
BuildRequires: pkg-config
%description
@ -87,11 +90,11 @@ engine to build services that work efficiently and seamlessly between C++,
Java, C#, Python, Ruby, Perl, PHP, Objective C/Cocoa, Smalltalk, Erlang,
Objective Caml, and Haskell.
%package -n libthrift%{libversion}
%package -n libthrift-%{libversion}
Summary: Thrift shared library
Group: System/Libraries
%description -n libthrift%{libversion}
%description -n libthrift-%{libversion}
Thrift shared library
Thrift is a software framework for scalable cross-language services
@ -100,12 +103,13 @@ engine to build services that work efficiently and seamlessly between C++,
Java, C#, Python, Ruby, Perl, PHP, Objective C/Cocoa, Smalltalk, Erlang,
Objective Caml, and Haskell.
%post -n libthrift%{libversion} -p /sbin/ldconfig
%postun -n libthrift%{libversion} -p /sbin/ldconfig
%post -n libthrift-%{libversion} -p /sbin/ldconfig
%postun -n libthrift-%{libversion} -p /sbin/ldconfig
%package -n libthrift-devel
Summary: Thrift C++ library development files
Group: Development/Libraries
Requires: libthrift-%{libversion} = %{version}
%description -n libthrift-devel
Thrift C++ library development files
@ -120,6 +124,7 @@ Objective Caml, and Haskell.
%package -n perl-thrift
Summary: Thrift perl library
Group: Development/Libraries/Perl
%{perl_requires}
%description -n perl-thrift
Thrift perl library
@ -149,33 +154,28 @@ Objective Caml, and Haskell.
%prep
%setup
%patch1 -p1
#%patch2 -p1
#%patch3 -p1
%build
export CXXFLAGS="%{optflags} -fPIC"
export LDFLAGS="-Wl,--hash-style=sysv"
%{configure}
# ln -s fixes 0.9.1 build, should not be requires later
ln -s . test/cpp/.libs
make -j1
# tests require static boost library
%configure \
--disable-tests \
--enable-static=no
%install
pushd compiler/cpp
%makeinstall
%make_install
popd
pushd lib/cpp
%makeinstall
%make_install
popd
%if %{with python}
pushd lib/py
#%#makeinstall
python setup.py install --prefix=%{_prefix} --root=%{buildroot}
%fdupes %{buildroot}%{python_sitearch}
popd
%fdupes
%endif
%if %{with perl}
@ -183,43 +183,44 @@ pushd lib/perl
perl Makefile.PL
%perl_make_install
%perl_process_packlist
rm -rf %{buildroot}%{perl_vendorarch}/auto
popd
%endif
%clean
rm %{buildroot}%{_libdir}/*.la
%files
%defattr(-,root,root)
%doc CHANGES LICENSE NOTICE README
%doc CHANGES LICENSE NOTICE
%{_bindir}/thrift
%files -n libthrift%{libversion}
%files -n libthrift-%{libversion}
%defattr(-,root,root)
%{_libdir}/*-%{libname}.so
%{_libdir}/libthrift-*.so
%{_libdir}/libthriftnb-*.so
%{_libdir}/libthriftqt-*.so
%{_libdir}/libthriftz-*.so
%files -n libthrift-devel
%defattr(-,root,root)
%{_includedir}/thrift
%{_libdir}/*.a
%{_libdir}/*.la
%{_libdir}/*.so
%exclude %{_libdir}/*-%{libname}.so
%{_libdir}/libthrift.so
%{_libdir}/libthriftnb.so
%{_libdir}/libthriftqt.so
%{_libdir}/libthriftz.so
%{_libdir}/pkgconfig/*.pc
%if %{with perl}
%files -n perl-thrift
%defattr(-,root,root)
%{perl_vendorlib}/Thrift
%{perl_vendorlib}/Thrift.pm
%{perl_vendorlib}/Thrift
%endif
%if %{with python}
%files -n python-thrift
%defattr(-,root,root)
#%#{py_sitedir}/*
%{python_sitearch}/thrift-%{version}-py%{py_ver}.egg-info
%{python_sitearch}/thrift
%{py_sitedir}/*
%{python_sitearch}/*
%endif
%changelog