THRIFT-3814 Fix contention in TNonblockingServerTest
This closes #1005
diff --git a/lib/cpp/src/thrift/server/TNonblockingServer.cpp b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
index 7f44c1f..ccc37a2 100644
--- a/lib/cpp/src/thrift/server/TNonblockingServer.cpp
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
@@ -1130,7 +1130,7 @@
if (listen(s, LISTEN_BACKLOG) == -1) {
::THRIFT_CLOSESOCKET(s);
- throw TException("TNonblockingServer::serve() listen");
+ throw TTransportException(TTransportException::NOT_OPEN, "TNonblockingServer::serve() listen");
}
// Cool, this socket is good to go, set it as the serverSocket_
@@ -1438,7 +1438,7 @@
fd_set wfds, efds;
long ret = -1;
long kSize = sizeof(conn);
- const char* pos = (const char*)const_cast_sockopt(&conn);
+ const char* pos = reinterpret_cast<const char*>(&conn);
while (kSize > 0) {
FD_ZERO(&wfds);
diff --git a/lib/cpp/test/TNonblockingServerTest.cpp b/lib/cpp/test/TNonblockingServerTest.cpp
index 8f4ef6e..48ea913 100644
--- a/lib/cpp/test/TNonblockingServerTest.cpp
+++ b/lib/cpp/test/TNonblockingServerTest.cpp
@@ -47,14 +47,32 @@
class Fixture {
private:
struct Runner : public apache::thrift::concurrency::Runnable {
+ int port;
+ boost::shared_ptr<event_base> userEventBase;
+ boost::shared_ptr<TProcessor> processor;
boost::shared_ptr<server::TNonblockingServer> server;
- bool error;
+
virtual void run() {
- error = false;
+ // When binding to explicit port, allow retrying to workaround bind failures on ports in use
+ int retryCount = port ? 10 : 0;
+ startServer(retryCount);
+ }
+
+ private:
+ void startServer(int retry_count) {
try {
+ server.reset(new server::TNonblockingServer(processor, port));
+ if (userEventBase) {
+ server->registerEvents(userEventBase.get());
+ }
server->serve();
- } catch (const TException&) {
- error = true;
+ } catch (const transport::TTransportException&) {
+ if (retry_count > 0) {
+ ++port;
+ startServer(retry_count - 1);
+ } else {
+ throw;
+ }
}
}
};
@@ -80,38 +98,24 @@
}
int startServer(int port) {
+ boost::shared_ptr<Runner> runner(new Runner);
+ runner->port = port;
+ runner->processor = processor;
+ runner->userEventBase = userEventBase_;
+
boost::scoped_ptr<apache::thrift::concurrency::ThreadFactory> threadFactory(
- new apache::thrift::concurrency::PlatformThreadFactory(
+ new apache::thrift::concurrency::PlatformThreadFactory(
#if !USE_BOOST_THREAD && !USE_STD_THREAD
- concurrency::PlatformThreadFactory::OTHER,
- concurrency::PlatformThreadFactory::NORMAL,
+ concurrency::PlatformThreadFactory::OTHER, concurrency::PlatformThreadFactory::NORMAL,
1,
#endif
- true));
-
- int retry_count = port ? 10 : 0;
- for (int p = port; p <= port + retry_count; p++) {
- server.reset(new server::TNonblockingServer(processor, p));
- if (userEventBase_) {
- try {
- server->registerEvents(userEventBase_.get());
- } catch (const TException&) {
- // retry with next port
- continue;
- }
- }
- boost::shared_ptr<Runner> runner(new Runner);
- runner->server = server;
- thread = threadFactory->newThread(runner);
- thread->start();
- // wait 50ms for the server to begin listening
- THRIFT_SLEEP_USEC(50000);
- if (!runner->error) {
- return p;
- }
- }
- throw transport::TTransportException(transport::TTransportException::NOT_OPEN,
- "Failed to start server.");
+ false));
+ thread = threadFactory->newThread(runner);
+ thread->start();
+ // wait 100 ms for the server to begin listening
+ THRIFT_SLEEP_USEC(100000);
+ server = runner->server;
+ return runner->port;
}
bool canCommunicate(int serverPort) {
@@ -128,10 +132,11 @@
private:
boost::shared_ptr<event_base> userEventBase_;
boost::shared_ptr<test::ParentServiceProcessor> processor;
- boost::shared_ptr<apache::thrift::concurrency::Thread> thread;
-
protected:
boost::shared_ptr<server::TNonblockingServer> server;
+private:
+ boost::shared_ptr<apache::thrift::concurrency::Thread> thread;
+
};
BOOST_AUTO_TEST_SUITE(TNonblockingServerTest)