svn commit: r720251 - in /incubator/qpid/trunk/qpid/cpp/src: qpid/client/Bounds.cpp qpid/client/Connector.cpp qpid/client/SessionImpl.cpp tests/ClientSessionTest.cpp

[gs...@apache.org]

Author: gsim
Date: Mon Nov 24 10:37:37 2008
New Revision: 720251

URL: http://svn.apache.org/viewvc?rev=720251&view=rev
Log:
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).


Modified:
    incubator/qpid/trunk/qpid/cpp/src/qpid/client/Bounds.cpp
    incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
    incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
    incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Bounds.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Bounds.cpp?rev=720251&r1=720250&r2=720251&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Bounds.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Bounds.cpp Mon Nov 24 10:37:37 2008
@@ -48,8 +48,7 @@
     if (current == 0) return;
     current -= std::min(size, current);
     if (current < max && lock.hasWaiters()) {
-        assert(lock.hasWaiters() == 1);
-        lock.notify();
+        lock.notifyAll();
     }
 }
 

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp?rev=720251&r1=720250&r2=720251&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connector.cpp Mon Nov 24 10:37:37 2008
@@ -294,7 +294,9 @@
 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();
     }

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp?rev=720251&r1=720250&r2=720251&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp Mon Nov 24 10:37:37 2008
@@ -451,8 +451,8 @@
 {
     boost::shared_ptr<ConnectionImpl> c =  connectionWeak.lock();
     if (c) {
-        c->expand(frame.encodedSize(), canBlock);
         channel.handle(frame);
+        c->expand(frame.encodedSize(), canBlock);
     }
 }
 

Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=720251&r1=720250&r2=720251&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Mon Nov 24 10:37:37 2008
@@ -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 @@
 
 }
 
+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()