QPID-2120 : Updated SimpleAMQQueue to ensure that the sub.isSuspended() check is done before any attempt to retrieve the getLastSeenNode(). This and the commits for QPID-1871,QPID-2116 are tested by RollbackOrderTest that now can be enabled for the Java profile.

Additional changes were done to SAMQQ to improve readability and add some debug loggging. Performance testing should be done to identify any penality for this additional logging.

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

1 files changed, 112 insertions, 68 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index b14b92b014..9894efed20 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -508,8 +508,11 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             throws AMQException
     {
         _deliveredMessages.incrementAndGet();
+        if (_logger.isDebugEnabled())
+        {
+            _logger.debug(sub + ": deliverMessage: " + entry.debugIdentity());
+        }
         sub.send(entry);
-
     }
 
     private boolean subscriptionReadyAndHasInterest(final Subscription sub, final QueueEntry entry)
@@ -1172,9 +1175,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
 
     public void deliverAsync()
     {
-        _stateChangeCount.incrementAndGet();
-
-        Runner runner = new Runner();
+        Runner runner = new Runner(_stateChangeCount.incrementAndGet());
 
         if (_asynchronousRunner.compareAndSet(null, runner))
         {
@@ -1187,13 +1188,23 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
         _asyncDelivery.execute(new SubFlushRunner(sub));
     }
 
+
     private class Runner implements ReadWriteRunnable
     {
+        String _name;
+        public Runner(long count)
+        {
+            _name = "QueueRunner-" + count + "-" + _logActor;
+        }
+
         public void run()
         {
+            String originalName = Thread.currentThread().getName();
             try
             {
+                Thread.currentThread().setName(_name);
                 CurrentActor.set(_logActor);
+
                 processQueue(this);
             }
             catch (AMQException e)
@@ -1203,9 +1214,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             finally
             {
                 CurrentActor.remove();
+                Thread.currentThread().setName(originalName);
             }
-
-
         }
 
         public boolean isRead()
@@ -1217,6 +1227,11 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
         {
             return true;
         }
+
+        public String toString()
+        {
+            return _name;
+        }
     }
 
     private class SubFlushRunner implements ReadWriteRunnable
@@ -1230,27 +1245,36 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
 
         public void run()
         {
-            boolean complete = false;
-            try
-            {
-                CurrentActor.set(_sub.getLogActor());
-                complete = flushSubscription(_sub, new Long(MAX_ASYNC_DELIVERIES));
 
-            }
-            catch (AMQException e)
-            {
-                _logger.error(e);
+            String originalName = Thread.currentThread().getName();
+            try{
+                Thread.currentThread().setName("SubFlushRunner-"+_sub);
+
+                boolean complete = false;
+                try
+                {
+                    CurrentActor.set(_sub.getLogActor());
+                    complete = flushSubscription(_sub, new Long(MAX_ASYNC_DELIVERIES));
+
+                }
+                catch (AMQException e)
+                {
+                    _logger.error(e);
+                }
+                finally
+                {
+                    CurrentActor.remove();
+                }
+                if (!complete && !_sub.isSuspended())
+                {
+                    _asyncDelivery.execute(this);
+                }
             }
             finally
             {
-                CurrentActor.remove();
-            }
-            if (!complete && !_sub.isSuspended())
-            {
-                _asyncDelivery.execute(this);
+                Thread.currentThread().setName(originalName);
             }
 
-
         }
 
         public boolean isRead()
@@ -1278,7 +1302,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             try
             {
                 sub.getSendLock();
-                atTail =  attemptDelivery(sub);
+                atTail = attemptDelivery(sub);
                 if (atTail && sub.isAutoClose())
                 {
                     unregisterSubscription(sub);
@@ -1308,63 +1332,78 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
         return atTail;
     }
 
+    /**
+     * Attempt delivery for the given subscription.
+     *
+     * Looks up the next node for the subscription and attempts to deliver it.
+     *
+     * @param sub
+     * @return
+     * @throws AMQException
+     */
     private boolean attemptDelivery(Subscription sub) throws AMQException
     {
         boolean atTail = false;
         boolean advanced = false;
-        boolean subActive = sub.isActive();
+        boolean subActive = sub.isActive() && !sub.isSuspended();
         if (subActive)
         {
             QueueEntry node = moveSubscriptionToNextNode(sub);
+            _logger.debug(sub + ": attempt delivery: " + node.debugIdentity());
             if (!(node.isAcquired() || node.isDeleted()))
             {
-                if (!sub.isSuspended())
+                if (sub.hasInterest(node))
                 {
-                    if (sub.hasInterest(node))
+                    if (!sub.wouldSuspend(node))
                     {
-                        if (!sub.wouldSuspend(node))
+                        if (!sub.isBrowser() && !node.acquire(sub))
                         {
-                            if (!sub.isBrowser() && !node.acquire(sub))
-                            {
-                                sub.restoreCredit(node);
-                            }
-                            else
+                            sub.restoreCredit(node);
+                        }
+                        else
+                        {
+                            deliverMessage(sub, node);
+
+                            if (sub.isBrowser())
                             {
-                                deliverMessage(sub, node);
+                                QueueEntry newNode = _entries.next(node);
 
-                                if (sub.isBrowser())
+                                if (newNode != null)
                                 {
-                                    QueueEntry newNode = _entries.next(node);
-
-                                    if (newNode != null)
-                                    {
-                                        advanced = true;
-                                        sub.setLastSeenEntry(node, newNode);
-                                        node = sub.getLastSeenEntry();
-                                    }
+                                    advanced = true;
+                                    sub.setLastSeenEntry(node, newNode);
+                                    node = sub.getLastSeenEntry();
                                 }
+                                
                             }
-
-                        }
-                        else // Not enough Credit for message and wouldSuspend
-                        {
-                            //QPID-1187 - Treat the subscription as suspended for this message
-                            // and wait for the message to be removed to continue delivery.
-                            subActive = false;
-                            node.addStateChangeListener(new QueueEntryListener(sub, node));
                         }
+
                     }
-                    else
+                    else // Not enough Credit for message and wouldSuspend
                     {
-                        // this subscription is not interested in this node so we can skip over it
-                        QueueEntry newNode = _entries.next(node);
-                        if (newNode != null)
-                        {
-                            sub.setLastSeenEntry(node, newNode);
-                        }
+                        //QPID-1187 - Treat the subscription as suspended for this message
+                        // and wait for the message to be removed to continue delivery.
+                        
+                        // 2009-09-30 : MR : setting subActive = false only causes, this
+                        // particular delivery attempt to end. This is called from
+                        // flushSubscription and processQueue both of which attempt
+                        // delivery a number of times. Won't a bytes limited
+                        // subscriber with not enough credit for the next message
+                        // create a lot of new QELs? How about a browser that calls
+                        // this method LONG.MAX_LONG times! 
+                        subActive = false;
+                        node.addStateChangeListener(new QueueEntryListener(sub, node));
+                    }
+                }
+                else
+                {
+                    // this subscription is not interested in this node so we can skip over it
+                    QueueEntry newNode = _entries.next(node);
+                    if (newNode != null)
+                    {
+                        sub.setLastSeenEntry(node, newNode);
                     }
                 }
-
             }
             atTail = (_entries.next(node) == null) && !advanced;
         }
@@ -1409,6 +1448,12 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             }
 
         }
+
+        if (_logger.isDebugEnabled())
+        {
+            _logger.debug(sub + ": nextNode: " + (node == null ? "null" : node.debugIdentity()));
+        }
+
         return node;
     }
 
@@ -1423,6 +1468,11 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
 
         _asynchronousRunner.compareAndSet(runner, null);
 
+        // For every message enqueue/requeue the we fire deliveryAsync() which
+        // increases _stateChangeCount. If _sCC changes whilst we are in our loop
+        // (detected by setting previousStateChangeCount to stateChangeCount in the loop body)
+        // then we will continue to run for a maximum of iterations.
+        // So whilst delivery/rejection is going on a processQueue thread will be running
         while (iterations != 0 && ((previousStateChangeCount != (stateChangeCount = _stateChangeCount.get())) || deliveryIncomplete) && _asynchronousRunner.compareAndSet(null, runner))
         {
             // we want to have one extra loop after every subscription has reached the point where it cannot move
@@ -1442,20 +1492,11 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             //iterate over the subscribers and try to advance their pointer
             while (subscriptionIter.advance())
             {
-                boolean closeConsumer = false;
                 Subscription sub = subscriptionIter.getNode().getSubscription();
                 sub.getSendLock();
                 try
                 {
-                    if (sub != null)
-                    {
-
-                        QueueEntry node = moveSubscriptionToNextNode(sub);
-                        if (node != null)
-                        {
-                            done = attemptDelivery(sub);
-                        }
-                    }
+                    done = attemptDelivery(sub);
                     if (done)
                     {
                         if (extraLoops == 0)
@@ -1492,11 +1533,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
         // therefore we should schedule this runner again (unless someone beats us to it :-) ).
         if (iterations == 0 && _asynchronousRunner.compareAndSet(null, runner))
         {
+            if (_logger.isDebugEnabled())
+            {
+                _logger.debug("Rescheduling runner:" + runner);
+            }
             _asyncDelivery.execute(runner);
         }
     }
 
-    @Override
     public void checkMessageStatus() throws AMQException
     {