Fix to allocation algorithm in queue: prevents infinite loop when first consumer for allocation is cancelled after starting to dispatch a particular method.

Removed alarming(!) log statement.



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

1 files changed, 10 insertions, 5 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 1dad0e1864..af248b8fae 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -154,16 +154,15 @@ Consumer::ptr Queue::allocate()
 
 bool Queue::dispatch(QueuedMessage& msg)
 {
+    //additions to the acquirers will result in a separate dispatch
+    //request, so won't result in anyone being missed
+    uint counter = getAcquirerCount();
     Consumer::ptr c = allocate();
-    Consumer::ptr first = c;
-    while(c){
+    while(c && counter--){
         if(c->deliver(msg)) {
             return true;            
         } else {
             c = allocate();
-            if (c == first) { 
-                break;
-            }
         }
     }
     return false;
@@ -194,6 +193,7 @@ void Queue::serviceAllBrowsers()
      Consumers copy;
      {
          RWlock::ScopedRlock locker(consumerLock);
+         if (browsers.empty()) return;//shortcut
          copy = browsers;
      }
      for (Consumers::iterator i = copy.begin(); i != copy.end(); i++) {
@@ -312,6 +312,11 @@ uint32_t Queue::getConsumerCount() const{
     return acquirers.size() + browsers.size();
 }
 
+uint32_t Queue::getAcquirerCount() const{
+    RWlock::ScopedRlock locker(consumerLock);
+    return acquirers.size();
+}
+
 bool Queue::canAutoDelete() const{
     RWlock::ScopedRlock locker(consumerLock);
     return autodelete && acquirers.empty() && browsers.empty();