Changed the producer rate limit timer callback

so that it generates a callback serialised with
the connection


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

4 files changed, 23 insertions, 11 deletions

diff --git a/qpid/cpp/src/qpid/broker/Connection.cpp b/qpid/cpp/src/qpid/broker/Connection.cpp
index b7446a2220..b06e06d353 100644
--- a/qpid/cpp/src/qpid/broker/Connection.cpp
+++ b/qpid/cpp/src/qpid/broker/Connection.cpp
@@ -80,7 +80,8 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
 
 void Connection::requestIOProcessing(boost::function0<void> callback)
 {
-    ioCallback = callback;
+    ScopedLock<Mutex> l(ioCallbackLock);
+    ioCallbacks.push(callback);
     out.activateOutput();
 }
 
@@ -221,10 +222,13 @@ bool Connection::hasOutput() { return outputTasks.hasOutput(); }
 
 bool Connection::doOutput() {    
     try{
-        if (ioCallback)
-            ioCallback(); // Lend the IO thread for management processing
-        ioCallback = 0;
-
+        {
+        ScopedLock<Mutex> l(ioCallbackLock);
+        while (!ioCallbacks.empty()) {
+            ioCallbacks.front()(); // Lend the IO thread for management processing
+            ioCallbacks.pop();
+        }
+        }
         if (mgmtClosing)
             close(connection::CLOSE_CODE_CONNECTION_FORCED, "Closed by Management Request");
         else
diff --git a/qpid/cpp/src/qpid/broker/Connection.h b/qpid/cpp/src/qpid/broker/Connection.h
index b1e1cda973..b659fe6468 100644
--- a/qpid/cpp/src/qpid/broker/Connection.h
+++ b/qpid/cpp/src/qpid/broker/Connection.h
@@ -25,6 +25,7 @@
 #include <memory>
 #include <sstream>
 #include <vector>
+#include <queue>
 
 #include <boost/ptr_container/ptr_map.hpp>
 
@@ -47,6 +48,7 @@
 #include "qpid/sys/ConnectionOutputHandler.h"
 #include "qpid/sys/Socket.h"
 #include "qpid/sys/TimeoutHandler.h"
+#include "qpid/sys/Mutex.h"
 
 #include <boost/ptr_container/ptr_map.hpp>
 #include <boost/bind.hpp>
@@ -119,7 +121,8 @@ class Connection : public sys::ConnectionInputHandler,
     const bool isLink;
     bool mgmtClosing;
     const std::string mgmtId;
-    boost::function0<void> ioCallback;
+    sys::Mutex ioCallbackLock;
+    std::queue<boost::function0<void> > ioCallbacks;
     qmf::org::apache::qpid::broker::Connection* mgmtObject;
     LinkRegistry& links;
     management::ManagementAgent* agent;
diff --git a/qpid/cpp/src/qpid/broker/ConnectionState.h b/qpid/cpp/src/qpid/broker/ConnectionState.h
index 0d7fbc5b3b..b712af11f5 100644
--- a/qpid/cpp/src/qpid/broker/ConnectionState.h
+++ b/qpid/cpp/src/qpid/broker/ConnectionState.h
@@ -100,6 +100,7 @@ class ConnectionState : public ConnectionToken, public management::Manageable
     framing::FrameHandler* getClusterOrderOutput() { return clusterOrderOut; }
     void setClusterOrderOutput(framing::FrameHandler& fh) { clusterOrderOut = &fh; }
 
+    virtual void requestIOProcessing (boost::function0<void>) = 0;
 
   protected:
     framing::ProtocolVersion version;
diff --git a/qpid/cpp/src/qpid/broker/SessionState.cpp b/qpid/cpp/src/qpid/broker/SessionState.cpp
index b64fc20787..7e5f605753 100644
--- a/qpid/cpp/src/qpid/broker/SessionState.cpp
+++ b/qpid/cpp/src/qpid/broker/SessionState.cpp
@@ -212,11 +212,15 @@ struct ScheduledCreditTask : public TimerTask {
     void fire() {
         // This is the best we can currently do to avoid a destruction/fire race
         if (!isCancelled()) {
-            if ( !sessionState.processSendCredit(0) ) {
-                QPID_LOG(warning, sessionState.getId() << ": Reschedule sending credit");
-                reset();
-                timer.add(this);
-            }
+            sessionState.getConnection().requestIOProcessing(boost::bind(&ScheduledCreditTask::sendCredit, this));
+        }
+    }
+    
+    void sendCredit() {
+        if ( !sessionState.processSendCredit(0) ) {
+            QPID_LOG(warning, sessionState.getId() << ": Reschedule sending credit");
+            reset();
+            timer.add(this);
         }
     }
 };