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

4 files changed, 118 insertions, 72 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
     {
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/RollbackOrderTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/RollbackOrderTest.java
index 2efe93eed8..ff766c907d 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/RollbackOrderTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/RollbackOrderTest.java
@@ -144,6 +144,7 @@ public class RollbackOrderTest extends QpidTestCase
                 }
                 catch (JMSException e)
                 {
+                    System.out.println("Error:" + e.getMessage());
                     exceptions[(int)count.getCount()] = e;
                 }
                 catch (AssertionFailedError cf)
@@ -154,6 +155,7 @@ public class RollbackOrderTest extends QpidTestCase
                         count.countDown();
                     }
 
+                    System.out.println("Error:" + cf.getMessage());
                     System.err.println(cf.getMessage());
                     cf.printStackTrace();
                     failed.set(true);
@@ -175,11 +177,15 @@ public class RollbackOrderTest extends QpidTestCase
             }
         }
 
+//        _consumer.close();
+        _connection.close();
+        
         assertFalse("Exceptions thrown during test run, Check Std.err.", failed.get());
     }
 
     @Override public void tearDown() throws Exception
     {
+
         drainQueue(_queue);
 
         super.tearDown();
diff --git a/qpid/java/test-profiles/08Excludes b/qpid/java/test-profiles/08Excludes
index b277c6d929..6f3898384d 100644
--- a/qpid/java/test-profiles/08Excludes
+++ b/qpid/java/test-profiles/08Excludes
@@ -14,8 +14,6 @@ org.apache.qpid.server.persistent.NoLocalAfterRecoveryTest#*
 // QPID-1823: this takes ages to run
 org.apache.qpid.client.SessionCreateTest#*
 
-org.apache.qpid.test.client.RollbackOrderTest#*
-
 // QPID-2097 exclude it from the InVM test runs until InVM JMX Interface is reliable
 org.apache.qpid.management.jmx.ManagementActorLoggingTest#*
 org.apache.qpid.server.queue.ModelTest#*
diff --git a/qpid/java/test-profiles/08StandaloneExcludes b/qpid/java/test-profiles/08StandaloneExcludes
index de876e06bb..ee781fb80f 100644
--- a/qpid/java/test-profiles/08StandaloneExcludes
+++ b/qpid/java/test-profiles/08StandaloneExcludes
@@ -39,8 +39,6 @@ org.apache.qpid.server.persistent.NoLocalAfterRecoveryTest#*
 // QPID-1823: this takes ages to run
 org.apache.qpid.client.SessionCreateTest#*
 
-org.apache.qpid.test.client.RollbackOrderTest#*
-
 // This test requires the standard configuration file for validation.
 // Excluding here does not reduce test coverage.
 org.apache.qpid.server.configuration.ServerConfigurationFileTest#*