QPID-1478: ensure concurrent publishers work correctly (as well as reported assertion, the test uncovered a potential deadlock due to bounds being expanded before frames were added to queue).

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

4 files changed, 61 insertions, 4 deletions

diff --git a/cpp/src/qpid/client/Bounds.cpp b/cpp/src/qpid/client/Bounds.cpp
index bbc08edc5e..e2ddb5dfec 100644
--- a/cpp/src/qpid/client/Bounds.cpp
+++ b/cpp/src/qpid/client/Bounds.cpp
@@ -48,8 +48,7 @@ void Bounds::reduce(size_t size) {
     if (current == 0) return;
     current -= std::min(size, current);
     if (current < max && lock.hasWaiters()) {
-        assert(lock.hasWaiters() == 1);
-        lock.notify();
+        lock.notifyAll();
     }
 }
 
diff --git a/cpp/src/qpid/client/Connector.cpp b/cpp/src/qpid/client/Connector.cpp
index ea32c4e097..724d464932 100644
--- a/cpp/src/qpid/client/Connector.cpp
+++ b/cpp/src/qpid/client/Connector.cpp
@@ -294,7 +294,9 @@ void TCPConnector::Writer::init(std::string id, sys::AsynchIO* a) {
 void TCPConnector::Writer::handle(framing::AMQFrame& frame) { 
     Mutex::ScopedLock l(lock);
     frames.push_back(frame);
-    if (frame.getEof()) {//or if we already have a buffers worth
+    //only try to write if this is the end of a frameset or if we
+    //already have a buffers worth of data
+    if (frame.getEof() || (bounds && bounds->getCurrentSize() >= maxFrameSize)) {
         lastEof = frames.size();
         aio->notifyPendingWrite();
     }
diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index cf71d4f4a5..ab8c1bddb8 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -451,8 +451,8 @@ void SessionImpl::sendFrame(AMQFrame& frame, bool canBlock)
 {
     boost::shared_ptr<ConnectionImpl> c =  connectionWeak.lock();
     if (c) {
-        c->expand(frame.encodedSize(), canBlock);
         channel.handle(frame);
+        c->expand(frame.encodedSize(), canBlock);
     }
 }
 
diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp
index 43c12ddf5c..ec040cabff 100644
--- a/cpp/src/tests/ClientSessionTest.cpp
+++ b/cpp/src/tests/ClientSessionTest.cpp
@@ -30,6 +30,8 @@
 
 #include <boost/optional.hpp>
 #include <boost/lexical_cast.hpp>
+#include <boost/bind.hpp>
+#include <boost/ptr_container/ptr_vector.hpp>
 
 #include <vector>
 
@@ -380,6 +382,60 @@ QPID_AUTO_TEST_CASE(testCompleteOnAccept) {
 
 }
 
+namespace
+{
+struct Publisher : qpid::sys::Runnable
+{
+    AsyncSession session;
+    Message message;
+    uint count;
+    Thread thread;
+
+    Publisher(Connection& con, Message m, uint c) : session(con.newSession()), message(m), count(c) {}
+
+    void start()
+    {
+        thread = Thread(*this);
+    }
+
+    void join()
+    {
+        thread.join();
+    }
+
+    void run()
+    {
+        for (uint i = 0; i < count; i++) {
+            session.messageTransfer(arg::content=message);
+        }
+        session.sync();
+        session.close();
+    }
+};
+}
+
+QPID_AUTO_TEST_CASE(testConcurrentSenders)
+{
+    //Ensure concurrent publishing sessions on a connection don't
+    //cause assertions, deadlocks or other undesirables:
+    BrokerFixture fix;
+    Connection connection;
+    ConnectionSettings settings;
+    settings.maxFrameSize = 1024;
+    settings.port = fix.broker->getPort(qpid::broker::Broker::TCP_TRANSPORT);
+    connection.open(settings);
+    AsyncSession session = connection.newSession();
+    Message message(string(512, 'X'));
+    
+    boost::ptr_vector<Publisher> publishers;
+    for (size_t i = 0; i < 5; i++) {
+        publishers.push_back(new Publisher(connection, message, 100));
+    }
+    for_each(publishers.begin(), publishers.end(), boost::bind(&Publisher::start, _1));
+    for_each(publishers.begin(), publishers.end(), boost::bind(&Publisher::join, _1));
+    connection.close();
+}
+
 QPID_AUTO_TEST_SUITE_END()