THRIFT-1881 TNonblockingServer does not release open connections or threads on shutdown
Patch: Randy Abernethy
diff --git a/lib/cpp/src/thrift/server/TNonblockingServer.cpp b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
index 21fddad..641f714 100644
--- a/lib/cpp/src/thrift/server/TNonblockingServer.cpp
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
@@ -215,9 +215,6 @@
*/
void workSocket();
- /// Close this connection and free or reset its resources.
- void close();
-
public:
class Task;
@@ -244,6 +241,9 @@
std::free(readBuffer_);
}
+ /// Close this connection and free or reset its resources.
+ void close();
+
/**
* Check buffers against any size limits and shrink it if exceeded.
*
@@ -861,18 +861,24 @@
}
TNonblockingServer::~TNonblockingServer() {
- // TODO: We currently leak any active TConnection objects.
- // Since we're shutting down and destroying the event_base, the TConnection
- // objects will never receive any additional callbacks. (And even if they
- // did, it would be bad, since they keep a pointer around to the server,
- // which is being destroyed.)
-
+ // Close any active connections (moves them to the idle connection stack)
+ while (activeConnections_.size()) {
+ activeConnections_.front()->close();
+ }
// Clean up unused TConnection objects in connectionStack_
while (!connectionStack_.empty()) {
TConnection* connection = connectionStack_.top();
connectionStack_.pop();
delete connection;
}
+ // The TNonblockingIOThread objects have shared_ptrs to the Thread
+ // objects and the Thread objects have shared_ptrs to the TNonblockingIOThread
+ // objects (as runnable) so these objects will never deallocate without help.
+ while (!ioThreads_.empty()) {
+ boost::shared_ptr<TNonblockingIOThread> iot = ioThreads_.back();
+ ioThreads_.pop_back();
+ iot->setThread(boost::shared_ptr<Thread>());
+ }
}
/**
@@ -901,6 +907,7 @@
connectionStack_.pop();
result->init(socket, ioThread, addr, addrLen);
}
+ activeConnections_.push_back(result);
return result;
}
@@ -910,6 +917,8 @@
void TNonblockingServer::returnConnection(TConnection* connection) {
Guard g(connMutex_);
+ activeConnections_.erase(std::remove(activeConnections_.begin(), activeConnections_.end(), connection), activeConnections_.end());
+
if (connectionStackLimit_ &&
(connectionStack_.size() >= connectionStackLimit_)) {
delete connection;
diff --git a/lib/cpp/src/thrift/server/TNonblockingServer.h b/lib/cpp/src/thrift/server/TNonblockingServer.h
index e995424..e7bbdc5 100644
--- a/lib/cpp/src/thrift/server/TNonblockingServer.h
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.h
@@ -256,6 +256,14 @@
std::stack<TConnection*> connectionStack_;
/**
+ * This container holds pointers to all active connections. This container
+ * allows the server to clean up unlcosed connection objects at destruction,
+ * which in turn allows their transports, protocols, processors and handlers
+ * to deallocate and clean up correctly.
+ */
+ std::vector<TConnection*> activeConnections_;
+
+ /**
* Called when server socket had something happen. We accept all waiting
* client connections on listen socket fd and assign TConnection objects
* to handle those requests.