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

3 files changed, 12 insertions, 9 deletions

diff --git a/cpp/src/qpid/broker/NullMessageStore.cpp b/cpp/src/qpid/broker/NullMessageStore.cpp
index b8e4950287..b3fc7ec9a7 100644
--- a/cpp/src/qpid/broker/NullMessageStore.cpp
+++ b/cpp/src/qpid/broker/NullMessageStore.cpp
@@ -49,10 +49,7 @@ using namespace qpid::broker;
 
 NullMessageStore::NullMessageStore(bool _warn) : warn(_warn){}
 
-void NullMessageStore::init(const std::string& /*dir*/, const bool /*async*/)
-{
-    QPID_LOG(info, "Can't init, store not enabled");
-}
+void NullMessageStore::init(const std::string& /*dir*/, const bool /*async*/) {}
 
 void NullMessageStore::create(PersistableQueue& queue)
 {
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();
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index c855582dbe..082ccce246 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -107,6 +107,7 @@ namespace qpid {
             void serviceBrowser(Consumer::ptr c);
             Consumer::ptr allocate();
             bool seek(QueuedMessage& msg, const framing::SequenceNumber& position);
+            uint32_t getAcquirerCount() const;
  
         protected:
 	   /**