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 (file)
--- a/lib/cpp/src/thrift/server/TNonblockingServer.cpp
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.cpp
@@ -215,9 +215,6 @@ class TNonblockingServer::TConnection {
    */
   void workSocket();
 
-  /// Close this connection and free or reset its resources.
-  void close();
-
  public:
 
   class Task;
@@ -244,6 +241,9 @@ class TNonblockingServer::TConnection {
     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 @@ void TNonblockingServer::TConnection::checkIdleBufferMemLimit(
 }
 
 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 @@ TNonblockingServer::TConnection* TNonblockingServer::createConnection(
     connectionStack_.pop();
     result->init(socket, ioThread, addr, addrLen);
   }
+  activeConnections_.push_back(result);
   return result;
 }
 
@@ -910,6 +917,8 @@ TNonblockingServer::TConnection* TNonblockingServer::createConnection(
 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 (file)
--- a/lib/cpp/src/thrift/server/TNonblockingServer.h
+++ b/lib/cpp/src/thrift/server/TNonblockingServer.h
@@ -255,6 +255,14 @@ class TNonblockingServer : public TServer {
    */
   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