- Hold up tearing down client I/O threads until there are no more

  connections active.
- Don't create more client I/O threads than connections.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@936309 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 8 insertions, 1 deletions

diff --git a/qpid/cpp/src/qpid/client/ConnectionImpl.cpp b/qpid/cpp/src/qpid/client/ConnectionImpl.cpp
index 692afaae00..23e312ad72 100644
--- a/qpid/cpp/src/qpid/client/ConnectionImpl.cpp
+++ b/qpid/cpp/src/qpid/client/ConnectionImpl.cpp
@@ -82,6 +82,7 @@ class IOThread {
     int ioThreads;
     int connections;
     Mutex threadLock;
+    Condition noConnections;
     std::vector<Thread> t;
     Poller::shared_ptr poller_;
 
@@ -91,7 +92,7 @@ public:
         ++connections;
         if (!poller_)
             poller_.reset(new Poller);
-        if (ioThreads < maxIOThreads) {
+        if (ioThreads < connections && ioThreads < maxIOThreads) {
             QPID_LOG(debug, "Created IO thread: " << ioThreads);
             ++ioThreads;
             t.push_back( Thread(poller_.get()) );
@@ -101,6 +102,8 @@ public:
     void sub() {
         ScopedLock<Mutex> l(threadLock);
         --connections;
+        if (connections == 0)
+            noConnections.notifyAll();
     }
 
     Poller::shared_ptr poller() const {
@@ -124,6 +127,10 @@ public:
     // and we can't do that before we're unloaded as we can't
     // restart the Poller after shutting it down
     ~IOThread() {
+        ScopedLock<Mutex> l(threadLock);
+        while (connections > 0) {
+            noConnections.wait(threadLock);
+        }
         if (poller_)
             poller_->shutdown();
         for (int i=0; i<ioThreads; ++i) {