From 71f053b9172cfcd5b2487ca8d96356d6250b346a Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Tue, 15 Jun 2010 15:34:10 +0000 Subject: QPID-2670: copy-on-write when tagging message for loop detection git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@954933 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/qpid/broker/Queue.cpp | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) (limited to 'cpp/src/qpid/broker/Queue.cpp') diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp index 8c9e5b8c48..7f7b2bc312 100644 --- a/cpp/src/qpid/broker/Queue.cpp +++ b/cpp/src/qpid/broker/Queue.cpp @@ -216,7 +216,8 @@ void Queue::requeue(const QueuedMessage& msg){ if(inLastNodeFailure && persistLastNode && !msg.payload->isStoredOnQueue(shared_from_this())) { msg.payload->forcePersistent(); if (msg.payload->isForcedPersistent() ){ - enqueue(0, msg.payload); + boost::intrusive_ptr payload = msg.payload; + enqueue(0, payload); } } } @@ -720,7 +721,7 @@ void Queue::setLastNodeFailure() // return true if store exists, -bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr msg, bool suppressPolicyCheck) +bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr& msg, bool suppressPolicyCheck) { ScopedUse u(barrier); if (!u.acquired) return false; @@ -741,6 +742,11 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr msg, } if (traceId.size()) { + //copy on write: take deep copy of message before modifying it + //as the frames may already be available for delivery on other + //threads + boost::intrusive_ptr copy(new Message(*msg)); + msg = copy; msg->addTraceId(traceId); } @@ -1158,7 +1164,8 @@ void Queue::enqueued(const QueuedMessage& m) policy->enqueued(m); } mgntEnqStats(m.payload); - enqueue ( 0, m.payload, true ); + boost::intrusive_ptr payload = m.payload; + enqueue ( 0, payload, true ); } else { QPID_LOG(warning, "Queue informed of enqueued message that has no payload"); } -- cgit v1.2.1