QPID-2511: Altered shutdown sequence in connectors; after connect() has returned there will always now be a shutdown callback made and only at that point is it guaranteed that there will be no further io callbacks. ConnectionImpl consequently waits for this before allowing itself to be deleted.

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

1 files changed, 41 insertions, 7 deletions

diff --git a/cpp/src/qpid/client/ConnectionImpl.cpp b/cpp/src/qpid/client/ConnectionImpl.cpp
index b6043518e8..6b352e2e65 100644
--- a/cpp/src/qpid/client/ConnectionImpl.cpp
+++ b/cpp/src/qpid/client/ConnectionImpl.cpp
@@ -166,7 +166,8 @@ ConnectionImpl::ConnectionImpl(framing::ProtocolVersion v, const ConnectionSetti
     : Bounds(settings.maxFrameSize * settings.bounds),
       handler(settings, v),
       version(v),
-      nextChannel(1)
+      nextChannel(1),
+      shutdownComplete(false)
 {
     QPID_LOG(debug, "ConnectionImpl created for " << version.toString());
     handler.in = boost::bind(&ConnectionImpl::incoming, this, _1);
@@ -181,10 +182,12 @@ ConnectionImpl::ConnectionImpl(framing::ProtocolVersion v, const ConnectionSetti
 const uint16_t ConnectionImpl::NEXT_CHANNEL = std::numeric_limits<uint16_t>::max();
 
 ConnectionImpl::~ConnectionImpl() {
-    // Important to close the connector first, to ensure the
-    // connector thread does not call on us while the destructor
-    // is running.
-    if (connector) connector->close();
+    if (connector) {
+        connector->close();
+        //wait until we get the shutdown callback to ensure that the
+        //io threads will not call back on us after deletion
+        waitForShutdownComplete();
+    }
     theIO().sub();
 }
 
@@ -244,7 +247,13 @@ void ConnectionImpl::open()
     connector.reset(Connector::create(protocol, theIO().poller(), version, handler, this));
     connector->setInputHandler(&handler);
     connector->setShutdownHandler(this);
-    connector->connect(host, port);
+    try {
+        connector->connect(host, port);
+    } catch (const std::exception& e) {
+        QPID_LOG(debug, "Failed to connect to " << protocol << ":" << host << ":" << port << " " << e.what());
+        connector.reset();
+        throw;
+    }
     connector->init();
  
     // Enable heartbeat if requested
@@ -330,9 +339,22 @@ void ConnectionImpl::closed(uint16_t code, const std::string& text) {
     closeInternal(boost::bind(&SessionImpl::connectionClosed, _1, code, text));
 }
 
+void ConnectionImpl::shutdown() {
+    //May need to take a temporary reference to ourselves to prevent
+    //our destructor being called until after we have notified of
+    //shutdown completion; the destructor may be called as a result of
+    //the call to failedConnection().
+    boost::shared_ptr<ConnectionImpl> temp;
+    if (!handler.isClosed()) {
+        temp = shared_from_this();
+        failedConnection();
+    }
+    notifyShutdownComplete();
+}
+
 static const std::string CONN_CLOSED("Connection closed");
 
-void ConnectionImpl::shutdown() {
+void ConnectionImpl::failedConnection() {
     if ( failureCallback )
       failureCallback();
 
@@ -375,4 +397,16 @@ boost::shared_ptr<SessionImpl>  ConnectionImpl::newSession(const std::string& na
     return simpl;
 }
 
+void ConnectionImpl::waitForShutdownComplete()
+{
+    Mutex::ScopedLock l(lock);
+    while(!shutdownComplete) lock.wait();
+}
+void ConnectionImpl::notifyShutdownComplete()
+{
+    Mutex::ScopedLock l(lock);
+    shutdownComplete = true;
+    lock.notifyAll();
+}
+
 }} // namespace qpid::client