Fix race condition in enqueue/dequeue callbacks.

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

2 files changed, 25 insertions, 14 deletions

diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index 2d42c650ce..e2799b0bff 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -389,28 +389,36 @@ void Message::setReplacementMessage(boost::intrusive_ptr<Message> msg, const Que
 }
 
 void Message::allEnqueuesComplete() {
-    MessageCallback* cb = 0;
-    {
-        sys::Mutex::ScopedLock l(lock);
-        std::swap(cb, enqueueCallback);
-    }
+    sys::Mutex::ScopedLock l(callbackLock);
+    MessageCallback* cb = enqueueCallback;
     if (cb && *cb) (*cb)(intrusive_ptr<Message>(this));
 }
 
 void Message::allDequeuesComplete() {
-    MessageCallback* cb = 0;
-    {
-        sys::Mutex::ScopedLock l(lock);
-        std::swap(cb, dequeueCallback);
-    }
+    sys::Mutex::ScopedLock l(callbackLock);
+    MessageCallback* cb = dequeueCallback;
     if (cb && *cb) (*cb)(intrusive_ptr<Message>(this));
 }
 
-void Message::setEnqueueCompleteCallback(MessageCallback& cb) { enqueueCallback = &cb; }
-void Message::resetEnqueueCompleteCallback() { enqueueCallback = 0; }
+void Message::setEnqueueCompleteCallback(MessageCallback& cb) {
+    sys::Mutex::ScopedLock l(callbackLock);
+    enqueueCallback = &cb;
+}
+
+void Message::resetEnqueueCompleteCallback() {
+    sys::Mutex::ScopedLock l(callbackLock);
+    enqueueCallback = 0;
+}
+
+void Message::setDequeueCompleteCallback(MessageCallback& cb) {
+    sys::Mutex::ScopedLock l(callbackLock);
+    dequeueCallback = &cb;
+}
 
-void Message::setDequeueCompleteCallback(MessageCallback& cb) { dequeueCallback = &cb; }
-void Message::resetDequeueCompleteCallback() { dequeueCallback = 0; }
+void Message::resetDequeueCompleteCallback() {
+    sys::Mutex::ScopedLock l(callbackLock);
+    dequeueCallback = 0;
+}
 
 framing::FieldTable& Message::getOrInsertHeaders()
 {
diff --git a/cpp/src/qpid/broker/Message.h b/cpp/src/qpid/broker/Message.h
index 87ff78b3b0..3894960c95 100644
--- a/cpp/src/qpid/broker/Message.h
+++ b/cpp/src/qpid/broker/Message.h
@@ -184,8 +184,11 @@ public:
 
     mutable Replacement replacement;
     mutable boost::intrusive_ptr<Message> empty;
+
+    sys::Mutex callbackLock;
     MessageCallback* enqueueCallback;
     MessageCallback* dequeueCallback;
+
     uint32_t requiredCredit;
     static std::string updateDestination;
 };