diff options
| -rw-r--r-- | qpid/cpp/src/qpid/broker/Connection.cpp | 14 | ||||
| -rw-r--r-- | qpid/cpp/src/qpid/broker/Connection.h | 5 | ||||
| -rw-r--r-- | qpid/cpp/src/qpid/broker/ConnectionState.h | 1 | ||||
| -rw-r--r-- | qpid/cpp/src/qpid/broker/SessionState.cpp | 14 |
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); } } }; |