qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cctriel...@apache.org
Subject svn commit: r575792 - in /incubator/qpid/trunk/qpid/cpp/src/qpid/broker: BrokerQueue.cpp PersistableMessage.h
Date Fri, 14 Sep 2007 21:09:31 GMT
Author: cctrieloff
Date: Fri Sep 14 14:09:31 2007
New Revision: 575792

URL: http://svn.apache.org/viewvc?rev=575792&view=rev
Log:
convert dequeue to counter for async

Modified:
    incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp
    incubator/qpid/trunk/qpid/cpp/src/qpid/broker/PersistableMessage.h

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp?rev=575792&r1=575791&r2=575792&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp Fri Sep 14 14:09:31 2007
@@ -323,6 +323,7 @@
 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;
     }

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/PersistableMessage.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/PersistableMessage.h?rev=575792&r1=575791&r2=575792&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/PersistableMessage.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/PersistableMessage.h Fri Sep 14 14:09:31
2007
@@ -38,6 +38,7 @@
 class PersistableMessage : public Persistable
 {
     sys::Monitor asyncEnqueueLock;
+    sys::Monitor asyncDequeueLock;
 
     /**
      * Tracks the number of outstanding asynchronous enqueue
@@ -49,12 +50,13 @@
     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 @@
 
     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 @@
         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++; 
+    }
+
     
 };
 



Mime
View raw message