QPID-1879 Don't use a thread for every new client Connection

- By default the max number of threads now used for network io
  is the number of cpus available.
- This can be overridden with the QPID_MAX_IOTHREADS environment
  variable or the config file
- The client threads are initialised (via a singleton) when first
  used in a Connection::open()

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

1 files changed, 16 insertions, 56 deletions

diff --git a/cpp/src/qpid/client/RdmaConnector.cpp b/cpp/src/qpid/client/RdmaConnector.cpp
index ea3566dacb..e51ee95c61 100644
--- a/cpp/src/qpid/client/RdmaConnector.cpp
+++ b/cpp/src/qpid/client/RdmaConnector.cpp
@@ -26,6 +26,7 @@
 #include "qpid/log/Statement.h"
 #include "qpid/sys/Time.h"
 #include "qpid/framing/AMQFrame.h"
+#include "qpid/framing/InitiationHandler.h"
 #include "qpid/sys/rdma/RdmaIO.h"
 #include "qpid/sys/Dispatcher.h"
 #include "qpid/sys/Poller.h"
@@ -48,7 +49,7 @@ using namespace qpid::framing;
 using boost::format;
 using boost::str;
 
-  class RdmaConnector : public Connector, public sys::Codec, private sys::Runnable
+  class RdmaConnector : public Connector, public sys::Codec
 {
     struct Buff;
 
@@ -60,30 +61,25 @@ using boost::str;
     Frames frames;
     size_t lastEof; // Position after last EOF in frames
     uint64_t currentSize;
-    Bounds* bounds;        
-    
-    
+    Bounds* bounds;
+
     framing::ProtocolVersion version;
     bool initiated;
 
-    sys::Mutex pollingLock;    
+    sys::Mutex pollingLock;
     bool polling;
-    bool joined;
 
     sys::ShutdownHandler* shutdownHandler;
     framing::InputHandler* input;
     framing::InitiationHandler* initialiser;
     framing::OutputHandler* output;
 
-    sys::Thread receiver;
-
     Rdma::AsynchIO* aio;
     sys::Poller::shared_ptr poller;
     std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
 
     ~RdmaConnector();
 
-    void run();
     void handleClosed();
     bool closeInternal();
 
@@ -101,7 +97,7 @@ using boost::str;
     std::string identifier;
 
     ConnectionImpl* impl;
-    
+
     void connect(const std::string& host, int port);
     void close();
     void send(framing::AMQFrame& frame);
@@ -120,15 +116,16 @@ using boost::str;
     bool canEncode();
 
 public:
-    RdmaConnector(framing::ProtocolVersion pVersion,
+    RdmaConnector(Poller::shared_ptr,
+              framing::ProtocolVersion pVersion,
               const ConnectionSettings&, 
               ConnectionImpl*);
 };
 
 // Static constructor which registers connector here
 namespace {
-    Connector* create(framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) {
-        return new RdmaConnector(v, s, c);
+    Connector* create(Poller::shared_ptr p, framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) {
+        return new RdmaConnector(p, v, s, c);
     }
 
     struct StaticInit {
@@ -140,7 +137,8 @@ namespace {
 }
 
 
-RdmaConnector::RdmaConnector(ProtocolVersion ver,
+RdmaConnector::RdmaConnector(Poller::shared_ptr p,
+                     ProtocolVersion ver,
                      const ConnectionSettings& settings,
                      ConnectionImpl* cimpl)
     : maxFrameSize(settings.maxFrameSize),
@@ -150,9 +148,9 @@ RdmaConnector::RdmaConnector(ProtocolVersion ver,
       version(ver), 
       initiated(false),
       polling(false),
-      joined(true),
       shutdownHandler(0),
       aio(0),
+      poller(p),
       impl(cimpl)
 {
     QPID_LOG(debug, "RdmaConnector created for " << version);
@@ -165,8 +163,6 @@ RdmaConnector::~RdmaConnector() {
 void RdmaConnector::connect(const std::string& host, int port){
     Mutex::ScopedLock l(pollingLock);
     assert(!polling);
-    assert(joined);
-    poller = Poller::shared_ptr(new Poller);
 
     SocketAddress sa(host, boost::lexical_cast<std::string>(port));
     Rdma::Connector* c = new Rdma::Connector(
@@ -179,8 +175,6 @@ void RdmaConnector::connect(const std::string& host, int port){
     c->start(poller);
 
     polling = true;
-    joined = false;
-    receiver = Thread(this);
 }
 
 // The following only gets run when connected
@@ -215,24 +209,12 @@ void RdmaConnector::rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusiv
 }
 
 bool RdmaConnector::closeInternal() {
-    bool ret;
-    {
     Mutex::ScopedLock l(pollingLock);
-    ret = polling;
-    if (polling) {
-        polling = false;
-        poller->shutdown();
-    }
-    if (joined || receiver.id() == Thread::current().id()) {
-        return ret;
-    }
-    joined = true;
-    }
-
-    receiver.join();
+    bool ret = polling;
+    polling = false;
     return ret;
 }
-        
+
 void RdmaConnector::close() {
     closeInternal();
 }
@@ -356,28 +338,6 @@ void RdmaConnector::eof(Rdma::AsynchIO&) {
     handleClosed();
 }
 
-void RdmaConnector::run(){
-    // Keep the connection impl in memory until run() completes.
-    //GRS: currently the ConnectionImpls destructor is where the Io thread is joined
-    //boost::shared_ptr<ConnectionImpl> protect = impl->shared_from_this();
-    //assert(protect);
-    try {
-        Dispatcher d(poller);
-	
-        //aio->start(poller);
-        d.run();
-        //aio->queueForDeletion();
-    } catch (const std::exception& e) {
-        {
-        // We're no longer polling
-        Mutex::ScopedLock l(pollingLock);
-        polling = false;
-        }
-        QPID_LOG(error, e.what());
-        handleClosed();
-    }
-}
-
 void RdmaConnector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer> sl)
 {
     securityLayer = sl;