Merged revisions 573738-573739,573741-574077,574079-574236,574238-574265,574267-574503,574505-574554,574556-574584,574586-574873,574875-574901,574903-575737,575739-575787,575789-575810,575812-577772,577774-577940,577942-578057,578059-578732,578734,578736-578744,578746-578827,578829-578844,578846-579114,579116-579146,579148-579197,579199-579228,579230-579573,579575-579576,579579-579601,579603-579613,579615-579708,579710-580021,580023-580039,580042-580060,580062-580065,580067-580080,580082-580257,580259-580264,580266-580350,580352-580984,580986-580991,580994-581001,581003-581170,581172-581188,581190-581206,581208-581245,581247-581292,581294-581539,581541-581565,581567-581620,581622-581626,581628-581646,581648-581967,581969-582197,582199-582200,582203-582204,582206-582262,582264,582267-583084,583087,583089-583104,583106-583146,583148-583153,583155-583169,583171-583172,583174-583398,583400-583414,583416-583417,583419-583437,583439-583482,583484-583517,583519-583545,583547,583549-583774,583777-583807,583809-583881,583883-584107,584109-584112,584114-584123,584125-585564,585566-585569,585571-585574,585576-585641,585643-585908 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1

........
  r585906 | ritchiem | 2007-10-18 10:09:38 +0100 (Thu, 18 Oct 2007) | 1 line
  
  QPID-647 : Async Process start/stop is not regulated tightly enough. Added additional synchronisation to ensure that a new subscriber can start the async if required. As currently the start request can be missed.
........


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

1 files changed, 20 insertions, 17 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
index d872b50458..71e50b178c 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
@@ -211,10 +211,7 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
         }
     }
 
-    /**
-     *
-     * @return the state of the async processor.
-     */
+    /** @return the state of the async processor. */
     public boolean isProcessingAsync()
     {
         return _processing.get();
@@ -830,14 +827,15 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
                 {
                     addMessageToQueue(msg, deliverFirst);
 
+                    //release lock now message is on queue.
+                    _lock.unlock();
+
                     //if  we have a non-filtering subscriber but queued messages && we're not Async && we have other Active subs then something is wrong!
                     if ((s != null && hasQueuedMessages()) && !isProcessingAsync() && _subscriptions.hasActiveSubscribers())
                     {
                         _queue.deliverAsync();
                     }
 
-                    //release lock now message is on queue.
-                    _lock.unlock();
 
                     //Pre Deliver to all subscriptions
                     if (debugEnabled)
@@ -984,7 +982,7 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
         return id;
     }
 
-    Runner asyncDelivery = new Runner();
+    final Runner _asyncDelivery = new Runner();
 
     private class Runner implements Runnable
     {
@@ -1000,11 +998,13 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
                 //Check that messages have not been added since we did our last peek();
                 // Synchronize with the thread that adds to the queue.
                 // If the queue is still empty then we can exit
-
-                if (!(hasQueuedMessages() && _subscriptions.hasActiveSubscribers()))
+                synchronized (_asyncDelivery)
                 {
-                    running = false;
-                    _processing.set(false);
+                    if (!(hasQueuedMessages() && _subscriptions.hasActiveSubscribers()))
+                    {
+                        running = false;
+                        _processing.set(false);
+                    }
                 }
             }
             Thread.currentThread().setName(startName);
@@ -1018,16 +1018,19 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
             _log.debug(debugIdentity() + "Processing Async." + currentStatus());
         }
 
-        if (hasQueuedMessages() && _subscriptions.hasActiveSubscribers())
+        synchronized (_asyncDelivery)
         {
-            //are we already running? if so, don't re-run
-            if (_processing.compareAndSet(false, true))
+            if (hasQueuedMessages() && _subscriptions.hasActiveSubscribers())
             {
-                if (_log.isDebugEnabled())
+                //are we already running? if so, don't re-run
+                if (_processing.compareAndSet(false, true))
                 {
-                    _log.debug(debugIdentity() + "Executing Async process.");
+                    if (_log.isDebugEnabled())
+                    {
+                        _log.debug(debugIdentity() + "Executing Async process.");
+                    }
+                    executor.execute(_asyncDelivery);
                 }
-                executor.execute(asyncDelivery);
             }
         }
     }