diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/qpid/broker/Queue.cpp | 7 | ||||
| -rw-r--r-- | cpp/src/qpid/management/ManagementObject.h | 4 |
2 files changed, 9 insertions, 2 deletions
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp index a8a7031757..2c9717caa0 100644 --- a/cpp/src/qpid/broker/Queue.cpp +++ b/cpp/src/qpid/broker/Queue.cpp @@ -130,6 +130,7 @@ void Queue::deliver(boost::intrusive_ptr<Message>& msg){ push(msg); msg->enqueueComplete(); if (mgmtObject.get() != 0) { + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgDepth (); @@ -137,6 +138,7 @@ void Queue::deliver(boost::intrusive_ptr<Message>& msg){ } }else { if (mgmtObject.get() != 0) { + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgDepth (); @@ -155,6 +157,7 @@ void Queue::recover(boost::intrusive_ptr<Message>& msg){ push(msg); msg->enqueueComplete(); // mark the message as enqueued if (mgmtObject.get() != 0) { + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgPersistEnqueues (); @@ -173,6 +176,7 @@ void Queue::recover(boost::intrusive_ptr<Message>& msg){ void Queue::process(boost::intrusive_ptr<Message>& msg){ push(msg); if (mgmtObject.get() != 0) { + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->inc_msgTotalEnqueues (); mgmtObject->inc_byteTotalEnqueues (msg->contentSize ()); mgmtObject->inc_msgTxnEnqueues (); @@ -363,6 +367,7 @@ void Queue::consume(Consumer& c, bool requestExclusive){ consumerCount++; if (mgmtObject.get() != 0){ + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->inc_consumers (); } } @@ -373,6 +378,7 @@ void Queue::cancel(Consumer& c){ consumerCount--; if(exclusive) exclusive = 0; if (mgmtObject.get() != 0){ + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->dec_consumers (); } } @@ -403,6 +409,7 @@ void Queue::pop(){ if (policy.get()) policy->dequeued(msg.payload->contentSize()); if (mgmtObject.get() != 0){ + sys::Mutex::ScopedLock mutex(mgmtObject->getLock()); mgmtObject->inc_msgTotalDequeues (); mgmtObject->inc_byteTotalDequeues (msg.payload->contentSize()); mgmtObject->dec_msgDepth (); diff --git a/cpp/src/qpid/management/ManagementObject.h b/cpp/src/qpid/management/ManagementObject.h index 047f8c5754..2661cf2d96 100644 --- a/cpp/src/qpid/management/ManagementObject.h +++ b/cpp/src/qpid/management/ManagementObject.h @@ -44,7 +44,7 @@ class ManagementObject bool instChanged; bool deleted; Manageable* coreObject; - sys::RWlock accessLock; + sys::Mutex accessLock; static const uint8_t TYPE_U8 = 1; static const uint8_t TYPE_U16 = 2; @@ -112,7 +112,7 @@ class ManagementObject deleted = true; } bool isDeleted (void) { return deleted; } - + sys::Mutex& getLock() { return accessLock; } }; typedef std::map<uint64_t,ManagementObject::shared_ptr> ManagementObjectMap; |