You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2010/06/15 17:34:10 UTC
svn commit: r954933 - in /qpid/trunk/qpid/cpp/src/qpid: broker/Message.cpp
broker/Message.h broker/Queue.cpp broker/Queue.h framing/FrameSet.cpp
framing/FrameSet.h
Author: gsim
Date: Tue Jun 15 15:34:10 2010
New Revision: 954933
URL: http://svn.apache.org/viewvc?rev=954933&view=rev
Log:
QPID-2670: copy-on-write when tagging message for loop detection
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.cpp
qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.h
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp?rev=954933&r1=954932&r2=954933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.cpp Tue Jun 15 15:34:10 2010
@@ -52,6 +52,12 @@ Message::Message(const framing::Sequence
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)
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h?rev=954933&r1=954932&r2=954933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.h Tue Jun 15 15:34:10 2010
@@ -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; }
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=954933&r1=954932&r2=954933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Tue Jun 15 15:34:10 2010
@@ -216,7 +216,8 @@ void Queue::requeue(const QueuedMessage&
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*
}
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
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");
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=954933&r1=954932&r2=954933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Tue Jun 15 15:34:10 2010
@@ -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)
Modified: qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.cpp?rev=954933&r1=954932&r2=954933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.cpp Tue Jun 15 15:34:10 2010
@@ -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)
{
Modified: qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.h?rev=954933&r1=954932&r2=954933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/framing/FrameSet.h Tue Jun 15 15:34:10 2010
@@ -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;
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org