QPID-3191: A fix for the performance problem which replaces the QueueListeners std::vector with a std::deque and adds some other improvements.

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

3 files changed, 29 insertions, 18 deletions

diff --git a/qpid/cpp/src/qpid/broker/Consumer.h b/qpid/cpp/src/qpid/broker/Consumer.h
index b96443fa7c..317338a8ad 100644
--- a/qpid/cpp/src/qpid/broker/Consumer.h
+++ b/qpid/cpp/src/qpid/broker/Consumer.h
@@ -29,15 +29,19 @@ namespace qpid {
 namespace broker {
 
 class Queue;
+class QueueListeners;
 
 class Consumer {
     const bool acquires;
+    // inListeners allows QueueListeners to efficiently track if this instance is registered
+    // for notifications without having to search its containers
+    bool inListeners;
   public:
     typedef boost::shared_ptr<Consumer> shared_ptr;            
     
     framing::SequenceNumber position;
     
-    Consumer(bool preAcquires = true) : acquires(preAcquires) {}
+    Consumer(bool preAcquires = true) : acquires(preAcquires), inListeners(false) {}
     bool preAcquires() const { return acquires; }
     virtual bool deliver(QueuedMessage& msg) = 0;
     virtual void notify() = 0;
@@ -45,6 +49,7 @@ class Consumer {
     virtual bool accept(boost::intrusive_ptr<Message>) { return true; }
     virtual OwnershipToken* getSession() = 0;
     virtual ~Consumer(){}
+    friend class QueueListeners;
 };
 
 }}
diff --git a/qpid/cpp/src/qpid/broker/QueueListeners.cpp b/qpid/cpp/src/qpid/broker/QueueListeners.cpp
index 4d2c57d6b4..591f4443bb 100644
--- a/qpid/cpp/src/qpid/broker/QueueListeners.cpp
+++ b/qpid/cpp/src/qpid/broker/QueueListeners.cpp
@@ -26,19 +26,25 @@ namespace broker {
 
 void QueueListeners::addListener(Consumer::shared_ptr c)
 {
-    if (c->preAcquires()) {
-        add(consumers, c);
-    } else {
-        add(browsers, c);
+    if (!c->inListeners) {
+        if (c->acquires) {
+            add(consumers, c);
+        } else {
+            add(browsers, c);
+        }
+        c->inListeners = true;
     }
 }
 
 void QueueListeners::removeListener(Consumer::shared_ptr c)
 {
-    if (c->preAcquires()) {
-        remove(consumers, c);
-    } else {
-        remove(browsers, c);
+    if (c->inListeners) {
+        if (c->acquires) {
+            remove(consumers, c);
+        } else {
+            remove(browsers, c);
+        }
+        c->inListeners = false;
     }
 }
 
@@ -46,18 +52,20 @@ void QueueListeners::populate(NotificationSet& set)
 {
     if (consumers.size()) {
         set.consumer = consumers.front();
-        consumers.erase(consumers.begin());
+        consumers.pop_front();
+        set.consumer->inListeners = false;
     } else {
-        // Don't swap the vectors, hang on to the memory allocated.
+        // Don't swap the deques, hang on to the memory allocated.
         set.browsers = browsers;
         browsers.clear();
+        for (Listeners::iterator i = set.browsers.begin(); i != set.browsers.end(); i++)
+            (*i)->inListeners = false;
     }
 }
 
 void QueueListeners::add(Listeners& listeners, Consumer::shared_ptr c)
 {
-    Listeners::iterator i = std::find(listeners.begin(), listeners.end(), c);
-    if (i == listeners.end()) listeners.push_back(c);
+    listeners.push_back(c);
 }
 
 void QueueListeners::remove(Listeners& listeners, Consumer::shared_ptr c)
@@ -73,9 +81,7 @@ void QueueListeners::NotificationSet::notify()
 }
 
 bool QueueListeners::contains(Consumer::shared_ptr c) const {
-    return
-        std::find(browsers.begin(), browsers.end(), c) != browsers.end() ||
-        std::find(consumers.begin(), consumers.end(), c) != consumers.end();
+    return c->inListeners;
 }
 
 void QueueListeners::ListenerSet::notifyAll()
diff --git a/qpid/cpp/src/qpid/broker/QueueListeners.h b/qpid/cpp/src/qpid/broker/QueueListeners.h
index 59d1c84ca4..0659499253 100644
--- a/qpid/cpp/src/qpid/broker/QueueListeners.h
+++ b/qpid/cpp/src/qpid/broker/QueueListeners.h
@@ -22,7 +22,7 @@
  *
  */
 #include "qpid/broker/Consumer.h"
-#include <vector>
+#include <deque>
 
 namespace qpid {
 namespace broker {
@@ -40,7 +40,7 @@ namespace broker {
 class QueueListeners
 {
   public:
-    typedef std::vector<Consumer::shared_ptr> Listeners;
+    typedef std::deque<Consumer::shared_ptr> Listeners;
 
     class NotificationSet
     {