QPID-1040: Patch from Ted Ross: Asynchronous Connector

Code to allow non-blocking connection of new sockets


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

6 files changed, 109 insertions, 10 deletions

diff --git a/cpp/src/qpid/sys/AsynchIO.h b/cpp/src/qpid/sys/AsynchIO.h
index 13bed78e85..847bdc50e4 100644
--- a/cpp/src/qpid/sys/AsynchIO.h
+++ b/cpp/src/qpid/sys/AsynchIO.h
@@ -52,7 +52,34 @@ private:
 };
 
 /*
- * Asycnchronous reader/writer: 
+ * Asynchronous connector: starts the process of initiating a connection and
+ * invokes a callback when completed or failed.
+ */
+class AsynchConnector : private DispatchHandle {
+public:
+    typedef boost::function1<void, const Socket&> ConnectedCallback;
+    typedef boost::function2<void, int, std::string> FailedCallback;
+
+private:
+    ConnectedCallback connCallback;
+    FailedCallback failCallback;
+    const Socket& socket;
+
+public:
+    AsynchConnector(const Socket& socket,
+                    Poller::shared_ptr poller,
+                    std::string hostname,
+                    uint16_t port,
+                    ConnectedCallback connCb,
+                    FailedCallback failCb = 0);
+
+private:
+    void connComplete(DispatchHandle& handle);
+    void failure(int, std::string);
+};
+
+/*
+ * Asychronous reader/writer: 
  * Reader accepts buffers to read into; reads into the provided buffers
  * and then does a callback with the buffer and amount read. Optionally it can callback
  * when there is something to read but no buffer to read it into.
diff --git a/cpp/src/qpid/sys/Dispatcher.cpp b/cpp/src/qpid/sys/Dispatcher.cpp
index c55f808b42..1df88ce0ae 100644
--- a/cpp/src/qpid/sys/Dispatcher.cpp
+++ b/cpp/src/qpid/sys/Dispatcher.cpp
@@ -322,8 +322,8 @@ void DispatchHandle::stopWatch() {
 // is to ensure that the lock is released before
 // we do the delete
 void DispatchHandle::doDelete() {
-	// Ensure that we're no longer watching anything
-	stopWatch();
+    // Ensure that we're no longer watching anything
+    stopWatch();
 
     // If we're in the middle of a callback defer the delete
     {
diff --git a/cpp/src/qpid/sys/Socket.h b/cpp/src/qpid/sys/Socket.h
index 20dc0f1ce3..806d6b5164 100644
--- a/cpp/src/qpid/sys/Socket.h
+++ b/cpp/src/qpid/sys/Socket.h
@@ -93,7 +93,12 @@ public:
     uint16_t getLocalPort() const;
     uint16_t getRemotePort() const;
 
-    
+    /**
+     * Returns the error code stored in the socket.  This may be used
+     * to determine the result of a non-blocking connect.
+     */
+    int getError() const;
+
     /** Accept a connection from a socket that is already listening
      * and has an incoming connection
      */
diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp
index 65ea380b07..045bc56e90 100644
--- a/cpp/src/qpid/sys/TCPIOPlugin.cpp
+++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp
@@ -101,16 +101,20 @@ void AsynchIOProtocolFactory::accept(Poller::shared_ptr poller, ConnectionCodec:
                            boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, fact, false)));
     acceptor->start(poller);
 }
-    
+
 void AsynchIOProtocolFactory::connect(
     Poller::shared_ptr poller,
     const std::string& host, int16_t port,
     ConnectionCodec::Factory* f)
 {
-    Socket* socket = new Socket();//Should be deleted by handle when socket closes
-    socket->connect(host, port);
-
-    established(poller, *socket, f, true);
+    // Note that the following logic does not cause a memory leak.
+    // The allocated Socket is freed either by the AsynchConnector
+    // upon connection failure or by the AsynchIO upon connection
+    // shutdown.  The allocated AsynchConnector frees itself when it
+    // is no longer needed.
+    Socket* socket = new Socket();
+    new AsynchConnector(*socket, poller, host, port,
+                        boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, f, true));
 }
 
 }} // namespace qpid::sys
diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp
index cedad5c011..9dcb841992 100644
--- a/cpp/src/qpid/sys/posix/AsynchIO.cpp
+++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -97,6 +97,57 @@ void AsynchAcceptor::readable(DispatchHandle& h) {
 }
 
 /*
+ * Asynch Connector
+ */
+
+AsynchConnector::AsynchConnector(const Socket& s,
+                                 Poller::shared_ptr poller,
+                                 std::string hostname,
+                                 uint16_t port,
+                                 ConnectedCallback connCb,
+                                 FailedCallback failCb) :
+    DispatchHandle(s,
+                   0,
+                   boost::bind(&AsynchConnector::connComplete, this, _1),
+                   boost::bind(&AsynchConnector::connComplete, this, _1)),
+    connCallback(connCb),
+    failCallback(failCb),
+    socket(s)
+{
+    socket.setNonblocking();
+    try {
+        socket.connect(hostname, port);
+        startWatch(poller);
+    } catch(std::exception& e) {
+        failure(-1, std::string(e.what()));
+    }
+}
+
+void AsynchConnector::connComplete(DispatchHandle& h)
+{
+    int errCode = socket.getError();
+
+    h.stopWatch();
+    if (errCode == 0) {
+        connCallback(socket);
+	DispatchHandle::doDelete();
+    } else {
+        failure(errCode, std::string(strerror(errCode)));
+    }
+}
+
+void AsynchConnector::failure(int errCode, std::string message)
+{
+    if (failCallback)
+        failCallback(errCode, message);
+
+    socket.close();
+    delete &socket;
+
+    DispatchHandle::doDelete();
+}
+
+/*
  * Asynch reader/writer
  */
 AsynchIO::AsynchIO(const Socket& s,
diff --git a/cpp/src/qpid/sys/posix/Socket.cpp b/cpp/src/qpid/sys/posix/Socket.cpp
index 5f10cd84c2..67f6b6db4c 100644
--- a/cpp/src/qpid/sys/posix/Socket.cpp
+++ b/cpp/src/qpid/sys/posix/Socket.cpp
@@ -148,7 +148,8 @@ void Socket::connect(const std::string& host, int port) const
     if (hp == 0)
         throw Exception(QPID_MSG("Cannot resolve " << host << ": " << h_errstr(h_errno)));
     ::memcpy(&name.sin_addr.s_addr, hp->h_addr_list[0], hp->h_length);
-    if (::connect(socket, (struct sockaddr*)(&name), sizeof(name)) < 0)
+    if ((::connect(socket, (struct sockaddr*)(&name), sizeof(name)) < 0) &&
+        (errno != EINPROGRESS))
         throw qpid::Exception(QPID_MSG(strError(errno) << ": " << host << ":" << port));
 }
 
@@ -257,6 +258,17 @@ uint16_t Socket::getRemotePort() const
     return atoi(getService(impl->fd, true).c_str());
 }
 
+int Socket::getError() const
+{
+    int       result;
+    socklen_t rSize = sizeof (result);
+
+    if (::getsockopt(impl->fd, SOL_SOCKET, SO_ERROR, &result, &rSize) < 0)
+        throw QPID_POSIX_ERROR(errno);
+
+    return result;
+}
+
 void Socket::configure(const Configuration& c)
 {
     c.configurePosixTcpSocket(impl->fd);