convert dequeue to counter for async

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

2 files changed, 38 insertions, 8 deletions

diff --git a/cpp/src/qpid/broker/BrokerQueue.cpp b/cpp/src/qpid/broker/BrokerQueue.cpp
index 29e2256b56..e41c54b52c 100644
--- a/cpp/src/qpid/broker/BrokerQueue.cpp
+++ b/cpp/src/qpid/broker/BrokerQueue.cpp
@@ -323,6 +323,7 @@ bool Queue::enqueue(TransactionContext* ctxt, Message::shared_ptr& msg)
 bool Queue::dequeue(TransactionContext* ctxt, Message::shared_ptr& msg)
 {
     if (msg->isPersistent() && store) {
+	msg->dequeueAsync(); //increment to async counter -- for message sent to more than one queue
         store->dequeue(ctxt, *msg.get(), *this);
 	return true;
     }
diff --git a/cpp/src/qpid/broker/PersistableMessage.h b/cpp/src/qpid/broker/PersistableMessage.h
index 30d8d37409..e932fafeae 100644
--- a/cpp/src/qpid/broker/PersistableMessage.h
+++ b/cpp/src/qpid/broker/PersistableMessage.h
@@ -38,6 +38,7 @@ namespace broker {
 class PersistableMessage : public Persistable
 {
     sys::Monitor asyncEnqueueLock;
+    sys::Monitor asyncDequeueLock;
 
     /**
      * Tracks the number of outstanding asynchronous enqueue
@@ -49,12 +50,13 @@ class PersistableMessage : public Persistable
     int asyncEnqueueCounter;
 
     /**
-     * Needs to be set false on Message construction, then
-     * set once the dequeueis complete, it gets set
-     * For transient, once dequeued, for durable, once
-     * dequeue record has been stored.
+     * Tracks the number of outstanding asynchronous dequeue
+     * operations. When the message is dequeued asynchronously the
+     * count is incremented; when that dequeue completes it is
+     * decremented. Thus when it is 0, there are no outstanding
+     * dequeues.
      */
-    bool dequeueCompleted;
+    int asyncDequeueCounter;
 
 public:
     typedef boost::shared_ptr<PersistableMessage> shared_ptr;
@@ -66,7 +68,10 @@ public:
 
     virtual ~PersistableMessage() {};
 
-    PersistableMessage(): asyncEnqueueCounter(0), dequeueCompleted(false) {}
+    PersistableMessage(): 
+    	asyncEnqueueCounter(0), 
+    	asyncDequeueCounter(0) 
+	{}
     
     inline void waitForEnqueueComplete() {
         sys::ScopedLock<sys::Monitor> l(asyncEnqueueLock);
@@ -94,8 +99,32 @@ public:
         asyncEnqueueCounter++; 
     }
 
-    inline bool isDequeueComplete() { return dequeueCompleted; }
-    inline void dequeueComplete() { dequeueCompleted = true; }
+    inline bool isDequeueComplete() { 
+        sys::ScopedLock<sys::Monitor> l(asyncDequeueLock);
+        return asyncDequeueCounter == 0;
+    }
+    
+    inline void dequeueComplete() { 
+        sys::ScopedLock<sys::Monitor> l(asyncDequeueLock);
+        if (asyncDequeueCounter > 0) {
+            if (--asyncDequeueCounter == 0) {
+                asyncDequeueLock.notify();
+            }
+        }
+    }
+
+    inline void waitForDequeueComplete() {
+        sys::ScopedLock<sys::Monitor> l(asyncDequeueLock);
+        while (asyncDequeueCounter > 0) {
+            asyncDequeueLock.wait();
+        }
+    }
+
+    inline void dequeueAsync() { 
+        sys::ScopedLock<sys::Monitor> l(asyncDequeueLock);
+        asyncDequeueCounter++; 
+    }
+
     
 };