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

6 files changed, 29 insertions, 4 deletions

diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index b086d59ca5..1e56544920 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -52,6 +52,12 @@ Message::Message(const framing::SequenceNumber& id) :
     expiration(FAR_FUTURE), enqueueCallback(0), dequeueCallback(0),
     inCallback(false), requiredCredit(0) {}
 
+Message::Message(const Message& original) :
+    PersistableMessage(), frames(original.frames), persistenceId(0), redelivered(false), loaded(false),
+    staged(false), forcePersistentPolicy(false), publisher(0), adapter(0), 
+    expiration(FAR_FUTURE), enqueueCallback(0), dequeueCallback(0),
+    inCallback(false), requiredCredit(0) {}
+
 Message::~Message()
 {
     if (expiryPolicy)
diff --git a/cpp/src/qpid/broker/Message.h b/cpp/src/qpid/broker/Message.h
index 4330a03469..ee80657f39 100644
--- a/cpp/src/qpid/broker/Message.h
+++ b/cpp/src/qpid/broker/Message.h
@@ -53,6 +53,7 @@ public:
     typedef boost::function<void (const boost::intrusive_ptr<Message>&)> MessageCallback;
     
     QPID_BROKER_EXTERN Message(const framing::SequenceNumber& id = framing::SequenceNumber());
+    QPID_BROKER_EXTERN Message(const Message&);
     QPID_BROKER_EXTERN ~Message();
         
     uint64_t getPersistenceId() const { return persistenceId; }
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<Message> 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<Message> msg, bool suppressPolicyCheck)
+bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message>& msg, bool suppressPolicyCheck)
 {
     ScopedUse u(barrier);
     if (!u.acquired) return false;
@@ -741,6 +742,11 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> 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<Message> 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<Message> payload = m.payload;
+        enqueue ( 0, payload, true );
     } else {
         QPID_LOG(warning, "Queue informed of enqueued message that has no payload");
     }
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index cdfa8a1a1f..ebef6e4433 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -260,7 +260,7 @@ namespace qpid {
             QPID_BROKER_EXTERN void setLastNodeFailure();
             QPID_BROKER_EXTERN void clearLastNodeFailure();
 
-            bool enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg, bool suppressPolicyCheck = false);
+            bool enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message>& msg, bool suppressPolicyCheck = false);
             void enqueueAborted(boost::intrusive_ptr<Message> msg);
             /**
              * dequeue from store (only done once messages is acknowledged)
diff --git a/cpp/src/qpid/framing/FrameSet.cpp b/cpp/src/qpid/framing/FrameSet.cpp
index c03dd39458..255aaf6e6b 100644
--- a/cpp/src/qpid/framing/FrameSet.cpp
+++ b/cpp/src/qpid/framing/FrameSet.cpp
@@ -29,6 +29,16 @@ using namespace qpid::framing;
 using namespace boost;
 
 FrameSet::FrameSet(const SequenceNumber& _id) : id(_id),contentSize(0),recalculateSize(true) { }
+FrameSet::FrameSet(const FrameSet& original) : id(original.id), contentSize(0), recalculateSize(true)
+{
+    for (Frames::const_iterator i = original.begin(); i != original.end(); ++i) {
+        parts.push_back(AMQFrame(*(i->getBody())));
+        parts.back().setFirstSegment(i->isFirstSegment());
+        parts.back().setLastSegment(i->isLastSegment());
+        parts.back().setFirstFrame(i->isFirstFrame());
+        parts.back().setLastFrame(i->isLastFrame());
+    }
+}
 
 void FrameSet::append(const AMQFrame& part)
 {
diff --git a/cpp/src/qpid/framing/FrameSet.h b/cpp/src/qpid/framing/FrameSet.h
index 398a709353..cae75e5ec8 100644
--- a/cpp/src/qpid/framing/FrameSet.h
+++ b/cpp/src/qpid/framing/FrameSet.h
@@ -46,6 +46,7 @@ public:
     typedef boost::shared_ptr<FrameSet> shared_ptr;
 
     QPID_COMMON_EXTERN FrameSet(const SequenceNumber& id);
+    QPID_COMMON_EXTERN FrameSet(const FrameSet&);
     QPID_COMMON_EXTERN void append(const AMQFrame& part);
     QPID_COMMON_EXTERN bool isComplete() const;