Improved/additional client API tests.

 - Replaced InProcessBroker with a more accurate loopback BrokerFixture.
 - Added asserts for mutex/condition/thread errors in debug build.
 - Added client tests for several exception conditions.
 - Added peer address to log ouput, client/server distinguished by (addr) or [addr]
 - Fixed various deadlocks & races exposed by the new  asserts & tests.


File-by-file:

New BrokerFixture replaces InProcessBroker
D      src/tests/InProcessBroker.h
M      src/tests/BrokerFixture.h
M      src/tests/SocketProxy.h
M      src/tests/Makefile.am

Made it run a bit faster.
M      src/tests/quick_perftest

Redundant
D      src/tests/APRBaseTest.cpp

Updated tests to use BrokerFixture
M      src/tests/ClientChannelTest.cpp
M      src/tests/exception_test.cpp
M      src/tests/ClientSessionTest.cpp

Print thread IDs in decimal, same as GDB.
M      src/qpid/log/Logger.cpp

Assert mutex/condition ops in debug build.
M      src/qpid/sys/posix/check.h
M      src/qpid/sys/posix/Mutex.h
M      src/qpid/sys/posix/Condition.h
M      src/qpid/sys/posix/Thread.h

Added toFd() so SocketProxy can use ::select()
M      src/qpid/sys/Socket.h
M      src/qpid/sys/posix/Socket.cpp

Fixes for races & deadlocks shown up by new tests & asserts.
Mostly shutdown/close issues.
M      src/qpid/client/ConnectionHandler.h
M      src/qpid/client/ConnectionImpl.cpp
M      src/qpid/client/Demux.h
M      src/qpid/client/SessionCore.cpp
M      src/qpid/client/ConnectionHandler.cpp
M      src/qpid/client/Connector.h
M      src/qpid/client/Demux.cpp
M      src/qpid/client/Dispatcher.cpp
M      src/qpid/client/ConnectionImpl.h

Logging peer address.
M      src/qpid/sys/AsynchIOAcceptor.cpp


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

1 files changed, 15 insertions, 7 deletions

diff --git a/cpp/src/qpid/client/Connector.cpp b/cpp/src/qpid/client/Connector.cpp
index 95314dcb40..4fb5aa6b4d 100644
--- a/cpp/src/qpid/client/Connector.cpp
+++ b/cpp/src/qpid/client/Connector.cpp
@@ -27,7 +27,7 @@
 #include "qpid/sys/AsynchIO.h"
 #include "qpid/sys/Dispatcher.h"
 #include "qpid/sys/Poller.h"
-
+#include "qpid/Msg.h"
 #include <boost/bind.hpp>
 
 namespace qpid {
@@ -43,6 +43,7 @@ Connector::Connector(
     send_buffer_size(buffer_size),
     version(ver), 
     closed(true),
+    joined(true),
     timeout(0),
     idleIn(0), idleOut(0), 
     timeoutHandler(0),
@@ -52,11 +53,11 @@ Connector::Connector(
 
 Connector::~Connector() {
     close();
-    if (receiver.id() && receiver.id() != Thread::current().id())
-        receiver.join();
 }
 
 void Connector::connect(const std::string& host, int port){
+    Mutex::ScopedLock l(closedLock);
+    assert(closed);
     socket.connect(host, port);
     closed = false;
     poller = Poller::shared_ptr(new Poller);
@@ -71,20 +72,27 @@ void Connector::connect(const std::string& host, int port){
 }
 
 void Connector::init(){
+    Mutex::ScopedLock l(closedLock);
+    assert(joined);
     ProtocolInitiation init(version);
-
     writeDataBlock(init);
+    joined = false;
     receiver = Thread(this);
 }
 
 bool Connector::closeInternal() {
     Mutex::ScopedLock l(closedLock);
+    bool ret = !closed;
     if (!closed) {
-        poller->shutdown();
         closed = true;
-        return true;
+        poller->shutdown();
+    }
+    if (!joined && receiver.id() != Thread::current().id()) {
+        joined = true;
+        Mutex::ScopedUnlock u(closedLock);
+        receiver.join();
     }
-    return false;
+    return ret;
 }
         
 void Connector::close() {