QPID-1936: Fix potential deadlock for durable ring queue

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

4 files changed, 58 insertions, 15 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index c96b1af6f8..64efc93a22 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -551,11 +551,16 @@ 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);
+        if (policy.get()) {
+            policy->tryEnqueue(qm);
+            //depending on policy, may have some dequeues
+            if (!isRecovery) pendingDequeues.swap(dequeues);
+        }
         if (insertSeqNo) msg->getOrInsertHeaders().setInt64(seqNoKey, sequence);
          
         LVQ::iterator i;
@@ -591,6 +596,10 @@ 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()
@@ -1026,4 +1035,10 @@ bool Queue::isEnqueued(const QueuedMessage& msg)
     return !policy.get() || policy->isEnqueued(msg);
 }
 
+void Queue::addPendingDequeue(const QueuedMessage& msg)
+{
+    //assumes lock is held - true at present but rather nasty as this is a public method
+    pendingDequeues.push_back(msg);    
+}
+
 QueueListeners& Queue::getListeners() { return listeners; }
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index 7890e46b03..339fcc8ce1 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -326,6 +326,18 @@ namespace qpid {
              */
             void recoveryComplete();
 
+            /**
+             * This is a hack to avoid deadlocks in durable ring
+             * queues. It is used for dequeueing messages in response
+             * to an enqueue while avoid holding lock over call to
+             * store.
+             * 
+             * Assumes messageLock is held - true for curent use case
+             * (QueuePolicy::tryEnqueue()) but rather nasty as this is a public
+             * method
+             **/
+            void addPendingDequeue(const QueuedMessage &msg);
+
             // For cluster update
             QueueListeners& getListeners();
         };
diff --git a/cpp/src/qpid/broker/QueuePolicy.cpp b/cpp/src/qpid/broker/QueuePolicy.cpp
index 16fb9adf63..368bd6552d 100644
--- a/cpp/src/qpid/broker/QueuePolicy.cpp
+++ b/cpp/src/qpid/broker/QueuePolicy.cpp
@@ -207,13 +207,9 @@ void RingQueuePolicy::dequeued(const QueuedMessage& m)
 {
     qpid::sys::Mutex::ScopedLock l(lock);
     //find and remove m from queue
-    for (Messages::iterator i = queue.begin(); i != queue.end(); i++) {
-        if (i->payload == m.payload) {
-            queue.erase(i);
-            //now update count and size
-            QueuePolicy::dequeued(m);
-            break;
-        }
+    if (find(m, pendingDequeues, true) || find(m, queue, true)) {
+        //now update count and size
+        QueuePolicy::dequeued(m);
     }
 }
 
@@ -223,12 +219,7 @@ bool RingQueuePolicy::isEnqueued(const QueuedMessage& m)
     //for non-strict ring policy, a message can be replaced (and
     //therefore dequeued) before it is accepted or released by
     //subscriber; need to detect this
-    for (Messages::const_iterator i = queue.begin(); i != queue.end(); i++) {
-        if (i->payload == m.payload) {
-            return true;
-        }
-    }
-    return false;
+    return find(m, pendingDequeues, false) || find(m, queue, false); 
 }
 
 bool RingQueuePolicy::checkLimit(const QueuedMessage& m)
@@ -248,7 +239,18 @@ bool RingQueuePolicy::checkLimit(const QueuedMessage& m)
         oldest = queue.front();
     }
     if (oldest.queue->acquire(oldest) || !strict) {
-        oldest.queue->dequeue(0, oldest);
+        {
+            //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);
@@ -264,6 +266,17 @@ bool RingQueuePolicy::checkLimit(const QueuedMessage& m)
     }
 }
 
+bool RingQueuePolicy::find(const QueuedMessage& m, Messages& q, bool remove)
+{
+    for (Messages::iterator i = q.begin(); i != q.end(); i++) {
+        if (i->payload == m.payload) {
+            if (remove) q.erase(i);
+            return true;
+        }
+    }
+    return false;
+}
+
 std::auto_ptr<QueuePolicy> QueuePolicy::createQueuePolicy(const qpid::framing::FieldTable& settings)
 {
     uint32_t maxCount = getInt(settings, maxCountKey, 0);
diff --git a/cpp/src/qpid/broker/QueuePolicy.h b/cpp/src/qpid/broker/QueuePolicy.h
index 45992f87ac..93e0cc73be 100644
--- a/cpp/src/qpid/broker/QueuePolicy.h
+++ b/cpp/src/qpid/broker/QueuePolicy.h
@@ -101,8 +101,11 @@ class RingQueuePolicy : public QueuePolicy
   private:
     typedef std::deque<QueuedMessage> Messages;
     qpid::sys::Mutex lock;
+    Messages pendingDequeues;
     Messages queue;
     const bool strict;
+
+    bool find(const QueuedMessage&, Messages&, bool remove);
 };
 
 }}