Joint checkin from gsim, kpvdr, cctrieloff. See QPID-2102: Exceeding reject queue policy under a transaction causes broker crash

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

14 files changed, 264 insertions, 151 deletions

diff --git a/cpp/src/qpid/broker/DtxAck.cpp b/cpp/src/qpid/broker/DtxAck.cpp
index b189ef4cdb..bca3f90bbe 100644
--- a/cpp/src/qpid/broker/DtxAck.cpp
+++ b/cpp/src/qpid/broker/DtxAck.cpp
@@ -48,12 +48,26 @@ bool DtxAck::prepare(TransactionContext* ctxt) throw()
 
 void DtxAck::commit() throw()
 {
-    for_each(pending.begin(), pending.end(), mem_fun_ref(&DeliveryRecord::committed));
-    pending.clear();
+    try {
+        for_each(pending.begin(), pending.end(), mem_fun_ref(&DeliveryRecord::committed));
+        pending.clear();
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Failed to commit: " << e.what());
+    } catch(...) {
+        QPID_LOG(error, "Failed to commit (unknown error)");
+    }
+
 }
 
 void DtxAck::rollback() throw()
 {
-    for_each(pending.begin(), pending.end(), mem_fun_ref(&DeliveryRecord::requeue));
-    pending.clear();
+    try {
+        for_each(pending.begin(), pending.end(), mem_fun_ref(&DeliveryRecord::requeue));
+        pending.clear();
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Failed to complete rollback: " << e.what());
+    } catch(...) {
+        QPID_LOG(error, "Failed to complete rollback (unknown error)");
+    }
+
 }
diff --git a/cpp/src/qpid/broker/Message.cpp b/cpp/src/qpid/broker/Message.cpp
index 7360010192..ae69a429c8 100644
--- a/cpp/src/qpid/broker/Message.cpp
+++ b/cpp/src/qpid/broker/Message.cpp
@@ -181,11 +181,23 @@ void Message::decodeContent(framing::Buffer& buffer)
     loaded = true;
 }
 
-void Message::releaseContent(MessageStore* _store)
+void Message::tryReleaseContent()
 {
-    if (!store) {
-        store = _store;
+    if (checkContentReleasable()) {
+        releaseContent();
     }
+}
+
+void Message::releaseContent(MessageStore* s)
+{
+    //deprecated, use setStore(store); releaseContent(); instead
+    if (!store) setStore(s);
+    releaseContent();
+}
+
+void Message::releaseContent()
+{
+    sys::Mutex::ScopedLock l(lock);
     if (store) {
         if (!getPersistenceId()) {
             intrusive_ptr<PersistableMessage> pmsg(this);
@@ -234,6 +246,7 @@ bool Message::getContentFrame(const Queue& queue, AMQFrame& frame, uint16_t maxC
 
 void Message::sendContent(const Queue& queue, framing::FrameHandler& out, uint16_t maxFrameSize) const
 {
+    sys::Mutex::ScopedLock l(lock);
     if (isContentReleased() && !frames.isComplete()) {
 
         uint16_t maxContentSize = maxFrameSize - AMQFrame::frameOverhead();
diff --git a/cpp/src/qpid/broker/Message.h b/cpp/src/qpid/broker/Message.h
index e4d09b1042..3845146955 100644
--- a/cpp/src/qpid/broker/Message.h
+++ b/cpp/src/qpid/broker/Message.h
@@ -129,12 +129,9 @@ public:
     QPID_BROKER_EXTERN void decodeHeader(framing::Buffer& buffer);
     QPID_BROKER_EXTERN void decodeContent(framing::Buffer& buffer);
             
-    /**
-     * Releases the in-memory content data held by this
-     * message. Must pass in a store from which the data can
-     * be reloaded.
-     */
-    void releaseContent(MessageStore* store);
+    void tryReleaseContent();
+    void releaseContent();
+    void releaseContent(MessageStore* s);//deprecated, use 'setStore(store); releaseContent();' instead
     void destroy();
 
     bool getContentFrame(const Queue& queue, framing::AMQFrame& frame, uint16_t maxContentSize, uint64_t offset) const;
diff --git a/cpp/src/qpid/broker/MessageBuilder.cpp b/cpp/src/qpid/broker/MessageBuilder.cpp
index 14b233fd6c..b1a2b77b05 100644
--- a/cpp/src/qpid/broker/MessageBuilder.cpp
+++ b/cpp/src/qpid/broker/MessageBuilder.cpp
@@ -80,7 +80,7 @@ void MessageBuilder::handle(AMQFrame& frame)
             && !NullMessageStore::isNullStore(store)
             && message->getExchangeName() != QPID_MANAGEMENT /* don't stage mgnt messages */)
         {
-            message->releaseContent(store);
+            message->releaseContent();
             staging = true;
         }
     }
@@ -96,6 +96,7 @@ void MessageBuilder::end()
 void MessageBuilder::start(const SequenceNumber& id)
 {
     message = intrusive_ptr<Message>(new Message(id));
+    message->setStore(store);
     state = METHOD;
     staging = false;
 }
diff --git a/cpp/src/qpid/broker/PersistableMessage.cpp b/cpp/src/qpid/broker/PersistableMessage.cpp
index 2ef223aa81..303a0501f4 100644
--- a/cpp/src/qpid/broker/PersistableMessage.cpp
+++ b/cpp/src/qpid/broker/PersistableMessage.cpp
@@ -36,7 +36,6 @@ PersistableMessage::~PersistableMessage() {}
 PersistableMessage::PersistableMessage() :
     asyncEnqueueCounter(0), 
     asyncDequeueCounter(0),
-    contentReleased(false),
     store(0)
 {}
 
@@ -59,9 +58,15 @@ void PersistableMessage::flush()
     } 
 }
 
-void PersistableMessage::setContentReleased() {contentReleased = true; }
+void PersistableMessage::setContentReleased()
+{
+    contentReleaseState.released = true;
+}
 
-bool PersistableMessage::isContentReleased()const { return contentReleased; }
+bool PersistableMessage::isContentReleased() const
+{ 
+    return contentReleaseState.released;
+}
        
 bool PersistableMessage::isEnqueueComplete() {
     sys::ScopedLock<sys::Mutex> l(asyncEnqueueLock);
@@ -153,6 +158,26 @@ void PersistableMessage::dequeueAsync() {
     asyncDequeueCounter++; 
 }
 
+PersistableMessage::ContentReleaseState::ContentReleaseState() : blocked(false), requested(false), released(false) {}
+
+void PersistableMessage::setStore(MessageStore* s)
+{
+    store = s;
+}
+
+void PersistableMessage::requestContentRelease()
+{
+    contentReleaseState.requested = true;
+}
+void PersistableMessage::blockContentRelease()
+{ 
+    contentReleaseState.blocked = true;
+}
+bool PersistableMessage::checkContentReleasable()
+{ 
+    return contentReleaseState.requested && !contentReleaseState.blocked;
+}
+
 }}
 
 
diff --git a/cpp/src/qpid/broker/PersistableMessage.h b/cpp/src/qpid/broker/PersistableMessage.h
index 0274b41375..a968480249 100644
--- a/cpp/src/qpid/broker/PersistableMessage.h
+++ b/cpp/src/qpid/broker/PersistableMessage.h
@@ -68,8 +68,16 @@ class PersistableMessage : public Persistable
     void enqueueAsync();
     void dequeueAsync();
 
-    bool contentReleased;
     syncList synclist;
+    struct ContentReleaseState
+    {
+        bool blocked;
+        bool requested;
+        bool released;
+        
+        ContentReleaseState();
+    };
+    ContentReleaseState contentReleaseState;
 
   protected:
     /** Called when all enqueues are complete for this message. */
@@ -97,6 +105,11 @@ class PersistableMessage : public Persistable
     void flush();
     
     bool isContentReleased() const;
+
+    void setStore(MessageStore*);
+    void requestContentRelease();
+    void blockContentRelease();
+    bool checkContentReleasable();
        
     QPID_BROKER_EXTERN bool isEnqueueComplete();
 
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index b2a8e223c5..1cc48a949e 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -181,6 +181,8 @@ void Queue::deliver(boost::intrusive_ptr<Message>& msg){
 
 
 void Queue::recover(boost::intrusive_ptr<Message>& msg){
+    if (policy.get()) policy->recoverEnqueued(msg);
+
     push(msg, true);
     if (store){ 
         // setup synclist for recovered messages, so they don't get re-stored on lastNodeFailure
@@ -563,15 +565,12 @@ void Queue::popMsg(QueuedMessage& qmsg)
 }
 
 void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
-    Messages dequeues;
     QueueListeners::NotificationSet copy;
     {
         Mutex::ScopedLock locker(messageLock);   
         QueuedMessage qm(this, msg, ++sequence);
         if (policy.get()) {
-            policy->tryEnqueue(qm);
-            //depending on policy, may have some dequeues
-            if (!isRecovery) pendingDequeues.swap(dequeues);
+            policy->enqueued(qm);
         }
         if (insertSeqNo) msg->getOrInsertHeaders().setInt64(seqNoKey, sequence);
          
@@ -608,10 +607,6 @@ void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
         }
     }
     copy.notify();
-    if (!dequeues.empty()) {
-        //depending on policy, may have some dequeues
-        for_each(dequeues.begin(), dequeues.end(), boost::bind(&Queue::dequeue, this, (TransactionContext*) 0, _1));
-    }
 }
 
 QueuedMessage Queue::getFront()
@@ -697,8 +692,12 @@ void Queue::setLastNodeFailure()
 }
 
 // return true if store exists, 
-bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg)
+bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg, bool suppressPolicyCheck)
 {
+    if (policy.get() && !suppressPolicyCheck) {
+        policy->tryEnqueue(msg);
+    }
+
     if (inLastNodeFailure && persistLastNode){
         msg->forcePersistent();
     }
@@ -713,9 +712,21 @@ bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg)
         store->enqueue(ctxt, pmsg, *this);
         return true;
     }
+    if (!store) {
+        //Messages enqueued on a transient queue should be prevented
+        //from having their content released as it may not be
+        //recoverable by these queue for delivery
+        msg->blockContentRelease();
+    }
     return false;
 }
 
+void Queue::enqueueAborted(boost::intrusive_ptr<Message> msg)
+{
+    Mutex::ScopedLock locker(messageLock);
+    if (policy.get()) policy->enqueueAborted(msg);       
+}
+
 // return true if store exists, 
 bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
 {
@@ -781,7 +792,15 @@ void Queue::create(const FieldTable& _settings)
 
 void Queue::configure(const FieldTable& _settings, bool recovering)
 {
-    setPolicy(QueuePolicy::createQueuePolicy(_settings));
+    if (QueuePolicy::getType(_settings) == QueuePolicy::FLOW_TO_DISK && 
+        (!store || NullMessageStore::isNullStore(store))) {
+        QPID_LOG(warning, "Flow to disk not valid for non-persisted queue");
+        FieldTable copy(_settings);
+        copy.erase(QueuePolicy::typeKey);
+        setPolicy(QueuePolicy::createQueuePolicy(getName(), copy));
+    } else {
+        setPolicy(QueuePolicy::createQueuePolicy(getName(), _settings));
+    }
     //set this regardless of owner to allow use of no-local with exclusive consumers also
     noLocal = _settings.get(qpidNoLocal);
     QPID_LOG(debug, "Configured queue with no-local=" << noLocal);
@@ -975,19 +994,6 @@ void Queue::setExternalQueueStore(ExternalQueueStore* inst) {
     }
 }
 
-bool Queue::releaseMessageContent(const QueuedMessage& m)
-{
-    if (store && !NullMessageStore::isNullStore(store)) {
-        QPID_LOG(debug, "Message " << m.position << " on " << name << " released from memory");
-        m.payload->releaseContent(store);
-        return true;
-    } else {
-        QPID_LOG(warning, "Message " << m.position << " on " << name
-                 << " cannot be released from memory as the queue is not durable");
-        return false;
-    }    
-}
-
 ManagementObject* Queue::GetManagementObject (void) const
 {
     return (ManagementObject*) mgmtObject;
@@ -1044,11 +1050,12 @@ void Queue::insertSequenceNumbers(const std::string& key)
 void Queue::enqueued(const QueuedMessage& m)
 {
     if (m.payload) {
-        if (policy.get()) policy->tryEnqueue(m);
-        mgntEnqStats(m.payload);
-        if (m.payload->isPersistent()) {
-            enqueue ( 0, m.payload );
+        if (policy.get()) {
+            policy->recoverEnqueued(m.payload);
+            policy->enqueued(m);
         }
+        mgntEnqStats(m.payload);
+        enqueue ( 0, m.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 77799fd967..9ac5e3f5e9 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -239,7 +239,8 @@ namespace qpid {
             QPID_BROKER_EXTERN void setLastNodeFailure();
             QPID_BROKER_EXTERN void clearLastNodeFailure();
 
-            bool enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg);
+            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)
              */
@@ -315,8 +316,6 @@ namespace qpid {
                 bindings.eachBinding(f);
             }
 
-            bool releaseMessageContent(const QueuedMessage&);
-
             void popMsg(QueuedMessage& qmsg);
 
             /** Set the position sequence number  for the next message on the queue.
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index 39afe90134..0f1f7f370f 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -28,8 +28,8 @@
 using namespace qpid::broker;
 using namespace qpid::framing;
 
-QueuePolicy::QueuePolicy(uint32_t _maxCount, uint64_t _maxSize, const std::string& _type) : 
-    maxCount(_maxCount), maxSize(_maxSize), type(_type), count(0), size(0), policyExceeded(false) {}
+QueuePolicy::QueuePolicy(const std::string& _name, uint32_t _maxCount, uint64_t _maxSize, const std::string& _type) : 
+    maxCount(_maxCount), maxSize(_maxSize), type(_type), count(0), size(0), policyExceeded(false), name(_name) {}
 
 void QueuePolicy::enqueued(uint64_t _size)
 {
@@ -39,18 +39,15 @@ void QueuePolicy::enqueued(uint64_t _size)
 
 void QueuePolicy::dequeued(uint64_t _size)
 {
-    //Note: underflow detection is not reliable in the face of
-    //concurrent updates (at present locking in Queue.cpp prevents
-    //these anyway); updates are atomic and are safe regardless.
     if (maxCount) {
-        if (count.get() > 0) {
+        if (count > 0) {
             --count;
         } else {
             throw Exception(QPID_MSG("Attempted count underflow on dequeue(" << _size << "): " << *this));
         }
     }
     if (maxSize) {
-        if (_size > size.get()) {
+        if (_size > size) {
             throw Exception(QPID_MSG("Attempted size underflow on dequeue(" << _size << "): " << *this));
         } else {
             size -= _size;
@@ -58,49 +55,53 @@ void QueuePolicy::dequeued(uint64_t _size)
     }
 }
 
-bool QueuePolicy::checkLimit(const QueuedMessage& m)
+bool QueuePolicy::checkLimit(boost::intrusive_ptr<Message> m)
 {
-    bool sizeExceeded = maxSize && (size.get() + m.payload->contentSize()) > maxSize;
-    bool countExceeded = maxCount && (count.get() + 1) > maxCount;
+    bool sizeExceeded = maxSize && (size + m->contentSize()) > maxSize;
+    bool countExceeded = maxCount && (count + 1) > maxCount;
     bool exceeded = sizeExceeded || countExceeded;
     if (exceeded) {
         if (!policyExceeded) {
-            policyExceeded = true;
-            if (m.queue) {
-                if (sizeExceeded) QPID_LOG(info, "Queue cumulative message size exceeded policy for " << m.queue->getName());
-                if (countExceeded) QPID_LOG(info, "Queue message count exceeded policy for " << m.queue->getName());
-            }
+            policyExceeded = true;            
+            if (sizeExceeded) QPID_LOG(info, "Queue cumulative message size exceeded policy for " << name);
+            if (countExceeded) QPID_LOG(info, "Queue message count exceeded policy for " << name);
         }
     } else {
         if (policyExceeded) {
             policyExceeded = false;
-            if (m.queue) {
-                QPID_LOG(info, "Queue cumulative message size and message count within policy for " << m.queue->getName());
-            }
+            QPID_LOG(info, "Queue cumulative message size and message count within policy for " << name);
         }
     }
     return !exceeded;
 }
 
-void QueuePolicy::tryEnqueue(const QueuedMessage& m)
+void QueuePolicy::tryEnqueue(boost::intrusive_ptr<Message> m)
 {
+    qpid::sys::Mutex::ScopedLock l(lock);
     if (checkLimit(m)) {
-        enqueued(m);
+        enqueued(m->contentSize());
     } else {
-        std::string queue = m.queue ? m.queue->getName() : std::string("unknown queue");
-        throw ResourceLimitExceededException(
-            QPID_MSG("Policy exceeded on " << queue << " by message " << m.position 
-                     << " of size " << m.payload->contentSize() << " , policy: " << *this));
+        throw ResourceLimitExceededException(QPID_MSG("Policy exceeded on " << name << ", policy: " << *this));
     }
 }
 
-void QueuePolicy::enqueued(const QueuedMessage& m)
+void QueuePolicy::recoverEnqueued(boost::intrusive_ptr<Message> m)
+{
+    qpid::sys::Mutex::ScopedLock l(lock);
+    enqueued(m->contentSize());
+}
+
+void QueuePolicy::enqueueAborted(boost::intrusive_ptr<Message> m)
 {
-    enqueued(m.payload->contentSize());
+    qpid::sys::Mutex::ScopedLock l(lock);
+    dequeued(m->contentSize());
 }
 
+void QueuePolicy::enqueued(const QueuedMessage&) {}
+
 void QueuePolicy::dequeued(const QueuedMessage& m)
 {
+    qpid::sys::Mutex::ScopedLock l(lock);
     dequeued(m.payload->contentSize());
 }
 
@@ -132,7 +133,7 @@ std::string QueuePolicy::getType(const FieldTable& settings)
         std::transform(t.begin(), t.end(), t.begin(), tolower);        
         if (t == REJECT || t == FLOW_TO_DISK || t == RING || t == RING_STRICT) return t;
     }
-    return FLOW_TO_DISK;
+    return REJECT;
 }
 
 void QueuePolicy::setDefaultMaxSize(uint64_t s)
@@ -148,8 +149,8 @@ void QueuePolicy::encode(Buffer& buffer) const
 {
   buffer.putLong(maxCount);
   buffer.putLongLong(maxSize);
-  buffer.putLong(count.get());
-  buffer.putLongLong(size.get());
+  buffer.putLong(count);
+  buffer.putLongLong(size);
 }
 
 void QueuePolicy::decode ( Buffer& buffer ) 
@@ -179,16 +180,18 @@ const std::string QueuePolicy::RING("ring");
 const std::string QueuePolicy::RING_STRICT("ring_strict");
 uint64_t QueuePolicy::defaultMaxSize(0);
 
-FlowToDiskPolicy::FlowToDiskPolicy(uint32_t _maxCount, uint64_t _maxSize) : 
-    QueuePolicy(_maxCount, _maxSize, FLOW_TO_DISK) {}
+FlowToDiskPolicy::FlowToDiskPolicy(const std::string& _name, uint32_t _maxCount, uint64_t _maxSize) : 
+    QueuePolicy(_name, _maxCount, _maxSize, FLOW_TO_DISK) {}
 
-bool FlowToDiskPolicy::checkLimit(const QueuedMessage& m)
+bool FlowToDiskPolicy::checkLimit(boost::intrusive_ptr<Message> m)
 {
-    return QueuePolicy::checkLimit(m) || m.queue->releaseMessageContent(m);
+    if (!QueuePolicy::checkLimit(m)) m->requestContentRelease(); 
+    return true;
 }
 
-RingQueuePolicy::RingQueuePolicy(uint32_t _maxCount, uint64_t _maxSize, const std::string& _type) : 
-    QueuePolicy(_maxCount, _maxSize, _type), strict(_type == RING_STRICT) {}
+RingQueuePolicy::RingQueuePolicy(const std::string& _name, 
+                                 uint32_t _maxCount, uint64_t _maxSize, const std::string& _type) : 
+    QueuePolicy(_name, _maxCount, _maxSize, _type), strict(_type == RING_STRICT) {}
 
 bool before(const QueuedMessage& a, const QueuedMessage& b)
 {
@@ -197,7 +200,6 @@ bool before(const QueuedMessage& a, const QueuedMessage& b)
 
 void RingQueuePolicy::enqueued(const QueuedMessage& m)
 {
-    QueuePolicy::enqueued(m);
     qpid::sys::Mutex::ScopedLock l(lock);
     //need to insert in correct location based on position
     queue.insert(lower_bound(queue.begin(), queue.end(), m, before), m);
@@ -222,42 +224,28 @@ bool RingQueuePolicy::isEnqueued(const QueuedMessage& m)
     return find(m, pendingDequeues, false) || find(m, queue, false); 
 }
 
-bool RingQueuePolicy::checkLimit(const QueuedMessage& m)
+bool RingQueuePolicy::checkLimit(boost::intrusive_ptr<Message> m)
 {
     if (QueuePolicy::checkLimit(m)) return true;//if haven't hit limit, ok to accept
     
     QueuedMessage oldest;
-    {
-        qpid::sys::Mutex::ScopedLock l(lock);
-        if (queue.empty()) {
-            QPID_LOG(debug, "Message too large for ring queue " 
-                     << (m.queue ? m.queue->getName() : std::string("unknown queue")) 
-                     << " [" << *this  << "] "
-                     << ": message size = " << m.payload->contentSize() << " bytes");
-            return false;
-        }
-        oldest = queue.front();
+    if (queue.empty()) {
+        QPID_LOG(debug, "Message too large for ring queue " << name 
+                 << " [" << *this  << "] "
+                 << ": message size = " << m->contentSize() << " bytes");
+        return false;
     }
+    oldest = queue.front();
     if (oldest.queue->acquire(oldest) || !strict) {
-        {
-            //TODO: fix this!  In the current code, this method is
-            //only ever called with the Queue lock already taken. This
-            //should not be relied upon going forward however and
-            //clearly the locking in this class is insufficient as
-            //there is no guarantee that the message previously atthe
-            //front is still there.
-            qpid::sys::Mutex::ScopedLock l(lock);
-            queue.pop_front();
-            pendingDequeues.push_back(oldest);
-        }
-        oldest.queue->addPendingDequeue(oldest);
-        QPID_LOG(debug, "Ring policy triggered in queue " 
-                 << (m.queue ? m.queue->getName() : std::string("unknown queue"))
-                 << ": removed message " << oldest.position << " to make way for " << m.position);
+        queue.pop_front();
+        pendingDequeues.push_back(oldest);
+        QPID_LOG(debug, "Ring policy triggered in " << name 
+                 << ": removed message " << oldest.position << " to make way for new message");
+        qpid::sys::Mutex::ScopedUnlock u(lock);
+        oldest.queue->dequeue(0, oldest);
         return true;
     } else {
-        QPID_LOG(debug, "Ring policy could not be triggered in queue " 
-                 << (m.queue ? m.queue->getName() : std::string("unknown queue")) 
+        QPID_LOG(debug, "Ring policy could not be triggered in " << name 
                  << ": oldest message (seq-no=" << oldest.position << ") has been delivered but not yet acknowledged or requeued");
         //in strict mode, if oldest message has been delivered (hence
         //cannot be acquired) but not yet acked, it should not be
@@ -277,25 +265,36 @@ bool RingQueuePolicy::find(const QueuedMessage& m, Messages& q, bool remove)
     return false;
 }
 
+std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(uint32_t maxCount, uint64_t maxSize, const std::string& type)
+{
+    return createQueuePolicy("<unspecified>", maxCount, maxSize, type);
+}
+
 std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(const qpid::framing::FieldTable& settings)
 {
+    return createQueuePolicy("<unspecified>", settings);
+}
+
+std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(const std::string& name, const qpid::framing::FieldTable& settings)
+{
     uint32_t maxCount = getInt(settings, maxCountKey, 0);
     uint32_t maxSize = getInt(settings, maxSizeKey, defaultMaxSize);
     if (maxCount || maxSize) {
-        return createQueuePolicy(maxCount, maxSize, getType(settings));
+        return createQueuePolicy(name, maxCount, maxSize, getType(settings));
     } else {
         return std::auto_ptr<QueuePolicy>();
     }
 }
 
-std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(uint32_t maxCount, uint64_t maxSize, const std::string& type)
+std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(const std::string& name, 
+                                                          uint32_t maxCount, uint64_t maxSize, const std::string& type)
 {
     if (type == RING || type == RING_STRICT) {
-        return std::auto_ptr<QueuePolicy>(new RingQueuePolicy(maxCount, maxSize, type));
+        return std::auto_ptr<QueuePolicy>(new RingQueuePolicy(name, maxCount, maxSize, type));
     } else if (type == FLOW_TO_DISK) {
-        return std::auto_ptr<QueuePolicy>(new FlowToDiskPolicy(maxCount, maxSize));
+        return std::auto_ptr<QueuePolicy>(new FlowToDiskPolicy(name, maxCount, maxSize));
     } else {
-        return std::auto_ptr<QueuePolicy>(new QueuePolicy(maxCount, maxSize, type));
+        return std::auto_ptr<QueuePolicy>(new QueuePolicy(name, maxCount, maxSize, type));
     }
 
 }
@@ -305,10 +304,10 @@ namespace qpid {
 
 std::ostream& operator<<(std::ostream& out, const QueuePolicy& p)
 {
-    if (p.maxSize) out << "size: max=" << p.maxSize << ", current=" << p.size.get();
+    if (p.maxSize) out << "size: max=" << p.maxSize << ", current=" << p.size;
     else out << "size: unlimited";
     out << "; ";
-    if (p.maxCount) out << "count: max=" << p.maxCount << ", current=" << p.count.get();
+    if (p.maxCount) out << "count: max=" << p.maxCount << ", current=" << p.count;
     else out << "count: unlimited";    
     out << "; type=" << p.type;
     return out;
diff --git a/cpp/src/qpid/broker/QueuePolicy.h b/cpp/src/qpid/broker/QueuePolicy.h
index 54745876d5..65c52304f2 100644
--- a/cpp/src/qpid/broker/QueuePolicy.h
+++ b/cpp/src/qpid/broker/QueuePolicy.h
@@ -40,12 +40,11 @@ class QueuePolicy
     uint32_t maxCount;
     uint64_t maxSize;
     const std::string type;
-    qpid::sys::AtomicValue<uint32_t> count;
-    qpid::sys::AtomicValue<uint64_t> size;
+    uint32_t count;
+    uint64_t size;
     bool policyExceeded;
             
     static int getInt(const qpid::framing::FieldTable& settings, const std::string& key, int defaultValue);
-    static std::string getType(const qpid::framing::FieldTable& settings);
 
   public:
     static QPID_BROKER_EXTERN const std::string maxCountKey;
@@ -57,10 +56,12 @@ class QueuePolicy
     static QPID_BROKER_EXTERN const std::string RING_STRICT;            
 
     virtual ~QueuePolicy() {}
-    QPID_BROKER_EXTERN void tryEnqueue(const QueuedMessage&);
+    QPID_BROKER_EXTERN void tryEnqueue(boost::intrusive_ptr<Message> msg);
+    QPID_BROKER_EXTERN void recoverEnqueued(boost::intrusive_ptr<Message> msg);
+    QPID_BROKER_EXTERN void enqueueAborted(boost::intrusive_ptr<Message> msg);
+    virtual void enqueued(const QueuedMessage&);
     virtual void dequeued(const QueuedMessage&);
     virtual bool isEnqueued(const QueuedMessage&);
-    virtual bool checkLimit(const QueuedMessage&);
     QPID_BROKER_EXTERN void update(qpid::framing::FieldTable& settings);
     uint32_t getMaxCount() const { return maxCount; }
     uint64_t getMaxSize() const { return maxSize; }           
@@ -69,15 +70,21 @@ class QueuePolicy
     uint32_t encodedSize() const;
 
 
+    static QPID_BROKER_EXTERN std::auto_ptr<QueuePolicy> createQueuePolicy(const std::string& name, const qpid::framing::FieldTable& settings);
+    static QPID_BROKER_EXTERN std::auto_ptr<QueuePolicy> createQueuePolicy(const std::string& name, uint32_t maxCount, uint64_t maxSize, const std::string& type = REJECT);
     static QPID_BROKER_EXTERN std::auto_ptr<QueuePolicy> createQueuePolicy(const qpid::framing::FieldTable& settings);
     static QPID_BROKER_EXTERN std::auto_ptr<QueuePolicy> createQueuePolicy(uint32_t maxCount, uint64_t maxSize, const std::string& type = REJECT);
+    static std::string getType(const qpid::framing::FieldTable& settings);
     static void setDefaultMaxSize(uint64_t);
     friend QPID_BROKER_EXTERN std::ostream& operator<<(std::ostream&,
                                                        const QueuePolicy&);
   protected:
-    QueuePolicy(uint32_t maxCount, uint64_t maxSize, const std::string& type = REJECT);
+    const std::string name;
+    qpid::sys::Mutex lock;
 
-    virtual void enqueued(const QueuedMessage&);
+    QueuePolicy(const std::string& name, uint32_t maxCount, uint64_t maxSize, const std::string& type = REJECT);
+
+    virtual bool checkLimit(boost::intrusive_ptr<Message> msg);
     void enqueued(uint64_t size);
     void dequeued(uint64_t size);
 };
@@ -86,21 +93,20 @@ class QueuePolicy
 class FlowToDiskPolicy : public QueuePolicy
 {
   public:
-    FlowToDiskPolicy(uint32_t maxCount, uint64_t maxSize);
-    bool checkLimit(const QueuedMessage&);
+    FlowToDiskPolicy(const std::string& name, uint32_t maxCount, uint64_t maxSize);
+    bool checkLimit(boost::intrusive_ptr<Message> msg);
 };
 
 class RingQueuePolicy : public QueuePolicy
 {
   public:
-    RingQueuePolicy(uint32_t maxCount, uint64_t maxSize, const std::string& type = RING);
+    RingQueuePolicy(const std::string& name, uint32_t maxCount, uint64_t maxSize, const std::string& type = RING);
     void enqueued(const QueuedMessage&);
     void dequeued(const QueuedMessage&);
     bool isEnqueued(const QueuedMessage&);
-    bool checkLimit(const QueuedMessage&);
+    bool checkLimit(boost::intrusive_ptr<Message> msg);
   private:
     typedef std::deque<QueuedMessage> Messages;
-    qpid::sys::Mutex lock;
     Messages pendingDequeues;
     Messages queue;
     const bool strict;
diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index bdd5f33601..a59d29c3cc 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -356,6 +356,9 @@ void SemanticState::handle(intrusive_ptr<Message> msg) {
     } else {
         DeliverableMessage deliverable(msg);
         route(msg, deliverable);
+        if (msg->checkContentReleasable()) {
+            msg->releaseContent();
+        }
     }
 }
 
diff --git a/cpp/src/qpid/broker/TxAccept.cpp b/cpp/src/qpid/broker/TxAccept.cpp
index e47ac84990..928ac12c10 100644
--- a/cpp/src/qpid/broker/TxAccept.cpp
+++ b/cpp/src/qpid/broker/TxAccept.cpp
@@ -88,7 +88,13 @@ bool TxAccept::prepare(TransactionContext* ctxt) throw()
 
 void TxAccept::commit() throw() 
 {
-    ops.commit();
+    try {
+        ops.commit();
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Failed to commit: " << e.what());
+    } catch(...) {
+        QPID_LOG(error, "Failed to commit (unknown error)");
+    }
 }
 
 void TxAccept::rollback() throw() {}
diff --git a/cpp/src/qpid/broker/TxPublish.cpp b/cpp/src/qpid/broker/TxPublish.cpp
index 17b99fd883..4b083033ea 100644
--- a/cpp/src/qpid/broker/TxPublish.cpp
+++ b/cpp/src/qpid/broker/TxPublish.cpp
@@ -26,9 +26,14 @@ using namespace qpid::broker;
 
 TxPublish::TxPublish(intrusive_ptr<Message> _msg) : msg(_msg) {}
 
-bool TxPublish::prepare(TransactionContext* ctxt) throw(){
+bool TxPublish::prepare(TransactionContext* ctxt) throw()
+{
     try{
-        for_each(queues.begin(), queues.end(), Prepare(ctxt, msg));
+        while (!queues.empty()) {
+            prepare(ctxt, queues.front());
+            prepared.push_back(queues.front());
+            queues.pop_front();
+        }
         return true;
     }catch(const std::exception& e){
         QPID_LOG(error, "Failed to prepare: " << e.what());
@@ -38,11 +43,30 @@ bool TxPublish::prepare(TransactionContext* ctxt) throw(){
     return false;
 }
 
-void TxPublish::commit() throw(){
-    for_each(queues.begin(), queues.end(), Commit(msg));
+void TxPublish::commit() throw()
+{
+    try {
+        for_each(prepared.begin(), prepared.end(), Commit(msg));
+        if (msg->checkContentReleasable()) {
+            msg->releaseContent();
+        }
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Failed to commit: " << e.what());
+    } catch(...) {
+        QPID_LOG(error, "Failed to commit (unknown error)");
+    }
 }
 
-void TxPublish::rollback() throw(){
+void TxPublish::rollback() throw()
+{
+    try {
+        for_each(prepared.begin(), prepared.end(), Rollback(msg));
+    } catch (const std::exception& e) {
+        QPID_LOG(error, "Failed to complete rollback: " << e.what());
+    } catch(...) {
+        QPID_LOG(error, "Failed to complete rollback (unknown error)");
+    }
+
 }
 
 void TxPublish::deliverTo(const boost::shared_ptr<Queue>& queue){
@@ -54,16 +78,14 @@ void TxPublish::deliverTo(const boost::shared_ptr<Queue>& queue){
     }
 }
 
-TxPublish::Prepare::Prepare(TransactionContext* _ctxt, intrusive_ptr<Message>& _msg) 
-    : ctxt(_ctxt), msg(_msg){}
-
-void TxPublish::Prepare::operator()(const boost::shared_ptr<Queue>& queue){
+void TxPublish::prepare(TransactionContext* ctxt, const boost::shared_ptr<Queue> queue)
+{
     if (!queue->enqueue(ctxt, msg)){
         /**
-	* if not store then mark message for ack and deleivery once
-	* commit happens, as async IO will never set it when no store
-	* exists
-	*/
+         * if not store then mark message for ack and deleivery once
+         * commit happens, as async IO will never set it when no store
+         * exists
+         */
 	msg->enqueueComplete();
     }
 }
@@ -74,6 +96,12 @@ void TxPublish::Commit::operator()(const boost::shared_ptr<Queue>& queue){
     queue->process(msg);
 }
 
+TxPublish::Rollback::Rollback(intrusive_ptr<Message>& _msg) : msg(_msg){}
+
+void TxPublish::Rollback::operator()(const boost::shared_ptr<Queue>& queue){
+    queue->enqueueAborted(msg);
+}
+
 uint64_t TxPublish::contentSize ()
 {
     return msg->contentSize ();
diff --git a/cpp/src/qpid/broker/TxPublish.h b/cpp/src/qpid/broker/TxPublish.h
index d5cf5639c4..b6ab9767ab 100644
--- a/cpp/src/qpid/broker/TxPublish.h
+++ b/cpp/src/qpid/broker/TxPublish.h
@@ -47,23 +47,25 @@ namespace qpid {
          * dispatch or to be added to the in-memory queue.
          */
         class TxPublish : public TxOp, public Deliverable{
-            class Prepare{
-                TransactionContext* ctxt;
+
+            class Commit{
                 boost::intrusive_ptr<Message>& msg;
             public:
-                Prepare(TransactionContext* ctxt, boost::intrusive_ptr<Message>& msg);
+                Commit(boost::intrusive_ptr<Message>& msg);
                 void operator()(const boost::shared_ptr<Queue>& queue);            
             };
-
-            class Commit{
+            class Rollback{
                 boost::intrusive_ptr<Message>& msg;
             public:
-                Commit(boost::intrusive_ptr<Message>& msg);
+                Rollback(boost::intrusive_ptr<Message>& msg);
                 void operator()(const boost::shared_ptr<Queue>& queue);            
             };
 
             boost::intrusive_ptr<Message> msg;
             std::list<Queue::shared_ptr> queues;
+            std::list<Queue::shared_ptr> prepared;
+
+            void prepare(TransactionContext* ctxt, boost::shared_ptr<Queue>);
 
         public:
             QPID_BROKER_EXTERN TxPublish(boost::intrusive_ptr<Message> msg);