QPID-3064, QPID-3157: ensure the node for a given Subscription is unlinked from the SubscriptionList at the point of removal instead of relying on the list head advancing to do so. Add 'marked node' functionality to the SubscriptionList to provide the Subscription cycling functionality to the queue, allowing the list to tidy itself up fully. Correct corner case behaviour for deleted-but-still-linked tail node in SubscriptionListIterator. Add tests.

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

3 files changed, 527 insertions, 99 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index 4dbe94e4d7..a095ef47ea 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -102,9 +102,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
 
     protected final QueueEntryList _entries;
 
-    protected final SubscriptionList _subscriptionList = new SubscriptionList(this);
-
-    private final AtomicReference<SubscriptionList.SubscriptionNode> _lastSubscriptionNode = new AtomicReference<SubscriptionList.SubscriptionNode>(_subscriptionList.getHead());
+    protected final SubscriptionList _subscriptionList = new SubscriptionList();
 
     private volatile Subscription _exclusiveSubscriber;
 
@@ -602,25 +600,25 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             iterate over subscriptions and if any is at the end of the queue and can deliver this message, then deliver the message
 
              */
-            SubscriptionList.SubscriptionNode node = _lastSubscriptionNode.get();
-            SubscriptionList.SubscriptionNode nextNode = node.getNext();
+            SubscriptionList.SubscriptionNode node = _subscriptionList.getMarkedNode();
+            SubscriptionList.SubscriptionNode nextNode = node.findNext();
             if (nextNode == null)
             {
-                nextNode = _subscriptionList.getHead().getNext();
+                nextNode = _subscriptionList.getHead().findNext();
             }
             while (nextNode != null)
             {
-                if (_lastSubscriptionNode.compareAndSet(node, nextNode))
+                if (_subscriptionList.updateMarkedNode(node, nextNode))
                 {
                     break;
                 }
                 else
                 {
-                    node = _lastSubscriptionNode.get();
-                    nextNode = node.getNext();
+                    node = _subscriptionList.getMarkedNode();
+                    nextNode = node.findNext();
                     if (nextNode == null)
                     {
-                        nextNode = _subscriptionList.getHead().getNext();
+                        nextNode = _subscriptionList.getHead().findNext();
                     }
                 }
             }
@@ -642,7 +640,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
                     Subscription sub = nextNode.getSubscription();
                     deliverToSubscription(sub, entry);
                 }
-                nextNode = nextNode.getNext();
+                nextNode = nextNode.findNext();
 
             }
         }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java b/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java
index 9ea81660c6..6dabfd21e2 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java
@@ -20,121 +20,108 @@
 */
 package org.apache.qpid.server.subscription;
 
-import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.subscription.Subscription;
 
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.nio.ByteBuffer;
 
 public class SubscriptionList
 {
-
     private final SubscriptionNode _head = new SubscriptionNode();
 
-    private AtomicReference<SubscriptionNode> _tail = new AtomicReference<SubscriptionNode>(_head);
-    private AtomicInteger _size = new AtomicInteger();
-
+    private final AtomicReference<SubscriptionNode> _tail = new AtomicReference<SubscriptionNode>(_head);
+    private final AtomicReference<SubscriptionNode> _subNodeMarker = new AtomicReference<SubscriptionNode>(_head);
+    private final AtomicInteger _size = new AtomicInteger();
 
-    public final class SubscriptionNode
+    public static final class SubscriptionNode
     {
         private final AtomicBoolean _deleted = new AtomicBoolean();
         private final AtomicReference<SubscriptionNode> _next = new AtomicReference<SubscriptionNode>();
         private final Subscription _sub;
 
-
         public SubscriptionNode()
         {
-
+            //used for sentinel head and dummy node construction
             _sub = null;
             _deleted.set(true);
         }
 
         public SubscriptionNode(final Subscription sub)
         {
+            //used for regular node construction
             _sub = sub;
         }
 
-
-        public SubscriptionNode getNext()
+        /**
+         * Retrieves the first non-deleted node following the current node.
+         * Any deleted non-tail nodes encountered during the search are unlinked.
+         *
+         * @return the next non-deleted node, or null if none was found.
+         */
+        public SubscriptionNode findNext()
         {
-
             SubscriptionNode next = nextNode();
             while(next != null && next.isDeleted())
             {
-
                 final SubscriptionNode newNext = next.nextNode();
                 if(newNext != null)
                 {
+                    //try to move our _next reference forward to the 'newNext'
+                    //node to unlink the deleted node
                     _next.compareAndSet(next, newNext);
                     next = nextNode();
                 }
                 else
                 {
+                    //'newNext' is null, meaning 'next' is the current tail. Can't unlink
+                    //the tail node for thread safety reasons, just use the null.
                     next = null;
                 }
-
             }
+
             return next;
         }
 
-        private SubscriptionNode nextNode()
+        /**
+         * Gets the immediately next referenced node in the structure.
+         *
+         * @return the immediately next node in the structure, or null if at the tail.
+         */
+        protected SubscriptionNode nextNode()
         {
             return _next.get();
         }
 
+        /**
+         * Used to initialise the 'next' reference. Will only succeed if the reference was not previously set.
+         *
+         * @param node the SubscriptionNode to set as 'next'
+         * @return whether the operation succeeded
+         */
+        private boolean setNext(final SubscriptionNode node)
+        {
+            return _next.compareAndSet(null, node);
+        }
+
         public boolean isDeleted()
         {
             return _deleted.get();
         }
 
-
         public boolean delete()
         {
-            if(_deleted.compareAndSet(false,true))
-            {
-                _size.decrementAndGet();
-                advanceHead();
-                return true;
-            }
-            else
-            {
-                return false;
-            }
+            return _deleted.compareAndSet(false,true);
         }
 
-
         public Subscription getSubscription()
         {
             return _sub;
         }
     }
 
-
-    public SubscriptionList(AMQQueue queue)
-    {
-    }
-
-    private void advanceHead()
-    {
-        SubscriptionNode head = _head.nextNode();
-        while(head._next.get() != null && head.isDeleted())
-        {
-
-            final SubscriptionNode newhead = head.nextNode();
-            if(newhead != null)
-            {
-                _head._next.compareAndSet(head, newhead);
-            }
-            head = _head.nextNode();
-        }
-    }
-
-
-    public SubscriptionNode add(Subscription sub)
+    private void insert(final SubscriptionNode node, final boolean count)
     {
-        SubscriptionNode node = new SubscriptionNode(sub);
         for (;;)
         {
             SubscriptionNode tail = _tail.get();
@@ -143,11 +130,14 @@ public class SubscriptionList
             {
                 if (next == null)
                 {
-                    if (tail._next.compareAndSet(null, node))
+                    if (tail.setNext(node))
                     {
                         _tail.compareAndSet(tail, node);
-                        _size.incrementAndGet();
-                        return node;
+                        if(count)
+                        {
+                            _size.incrementAndGet();
+                        }
+                        return;
                     }
                 }
                 else
@@ -156,27 +146,99 @@ public class SubscriptionList
                 }
             }
         }
+    }
 
+    public void add(final Subscription sub)
+    {
+        SubscriptionNode node = new SubscriptionNode(sub);
+        insert(node, true);
     }
 
-    public boolean remove(Subscription sub)
+    public boolean remove(final Subscription sub)
     {
-        SubscriptionNode node = _head.getNext();
+        SubscriptionNode prevNode = _head;
+        SubscriptionNode node = _head.nextNode();
+
         while(node != null)
         {
-            if(sub.equals(node._sub) && node.delete())
+            if(sub.equals(node.getSubscription()) && node.delete())
             {
+                _size.decrementAndGet();
+
+                SubscriptionNode tail = _tail.get();
+                if(node == tail)
+                {
+                    //we cant remove the last node from the structure for
+                    //correctness reasons, however we have just 'deleted'
+                    //the tail. Inserting an empty dummy node after it will
+                    //let us scavenge the node containing the Subscription.
+                    insert(new SubscriptionNode(), false);
+                }
+
+                //advance the next node reference in the 'prevNode' to scavange
+                //the newly 'deleted' node for the Subscription.
+                prevNode.findNext();
+
+                nodeMarkerCleanup(node);
+
                 return true;
             }
-            node = node.getNext();
+
+            prevNode = node;
+            node = node.findNext();
         }
+
         return false;
     }
 
+    private void nodeMarkerCleanup(SubscriptionNode node)
+    {
+        SubscriptionNode markedNode = _subNodeMarker.get();
+        if(node == markedNode)
+        {
+            //if the marked node is the one we are removing, then
+            //replace it with a dummy pointing at the next node.
+            //this is OK as the marked node is only used to index
+            //into the list and find the next node to use.
+            SubscriptionNode dummy = new SubscriptionNode();
+            dummy.setNext(markedNode.findNext());
+
+            //if the CAS fails the marked node has changed, thus
+            //we don't care about the dummy and just forget it
+            _subNodeMarker.compareAndSet(markedNode, dummy);
+        }
+        else if(markedNode != null)
+        {
+            //if the marked node was already deleted then it could
+            //hold subsequently removed nodes after it in the list 
+            //in memory. Scavenge it to ensure their actual removal.
+            if(markedNode != _head && markedNode.isDeleted())
+            {
+                markedNode.findNext();
+            }
+        }
+    }
 
-    public static class SubscriptionNodeIterator
+    public boolean updateMarkedNode(final SubscriptionNode expected, final SubscriptionNode nextNode)
+    {
+        return _subNodeMarker.compareAndSet(expected, nextNode);
+    }
+
+    /**
+     * Get the current marked SubscriptionNode. This should only be used only to index into the list and find the next node
+     * after the mark, since if the previously marked node was subsequently deleted the item returned may be a dummy node
+     * with reference to the next node.
+     *
+     * @return the previously marked node (or a dummy if it was subsequently deleted)
+     */
+    public SubscriptionNode getMarkedNode()
     {
+        return _subNodeMarker.get();
+    }
+
 
+    public static class SubscriptionNodeIterator
+    {
         private SubscriptionNode _lastNode;
 
         SubscriptionNodeIterator(SubscriptionNode startNode)
@@ -184,49 +246,25 @@ public class SubscriptionList
             _lastNode = startNode;
         }
 
-
-        public boolean atTail()
-        {
-            return _lastNode.nextNode() == null;
-        }
-
         public SubscriptionNode getNode()
         {
-
             return _lastNode;
-
         }
 
         public boolean advance()
         {
+            SubscriptionNode nextNode = _lastNode.findNext();
+            _lastNode = nextNode;
 
-            if(!atTail())
-            {
-                SubscriptionNode nextNode = _lastNode.nextNode();
-                while(nextNode.isDeleted() && nextNode.nextNode() != null)
-                {
-                    nextNode = nextNode.nextNode();
-                }
-                _lastNode = nextNode;
-                return true;
-
-            }
-            else
-            {
-                return false;
-            }
-
+            return _lastNode != null;
         }
-
     }
 
-
     public SubscriptionNodeIterator iterator()
     {
         return new SubscriptionNodeIterator(_head);
     }
 
-
     public SubscriptionNode getHead()
     {
         return _head;
@@ -236,9 +274,6 @@ public class SubscriptionList
     {
         return _size.get();
     }
-
-
-
 }
 
 
diff --git a/java/broker/src/test/java/org/apache/qpid/server/subscription/SubscriptionListTest.java b/java/broker/src/test/java/org/apache/qpid/server/subscription/SubscriptionListTest.java
new file mode 100644
index 0000000000..a9acfccd8e
--- /dev/null
+++ b/java/broker/src/test/java/org/apache/qpid/server/subscription/SubscriptionListTest.java
@@ -0,0 +1,395 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.subscription;
+
+import org.apache.qpid.server.subscription.SubscriptionList.SubscriptionNode;
+import org.apache.qpid.server.subscription.SubscriptionList.SubscriptionNodeIterator;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class SubscriptionListTest extends QpidTestCase
+{
+    private SubscriptionList _subList;
+    private MockSubscription _sub1;
+    private MockSubscription _sub2;
+    private MockSubscription _sub3;
+    private SubscriptionNode _node;
+
+    protected void setUp()
+    {
+        _subList = new SubscriptionList();
+
+        _sub1 = new MockSubscription();
+        _sub2 = new MockSubscription();
+        _sub3 = new MockSubscription();
+
+        _subList.add(_sub1);
+        _subList.add(_sub2);
+        _subList.add(_sub3);
+
+        _node = _subList.getHead();
+    }
+
+    /**
+     * Test that if the first (non-head) node in the list is deleted (but is still present),
+     * it is not returned when searching through the list for the next viable node, and the
+     * subsequent viable node is returned instead.
+     */
+    public void testFindNextSkipsFirstDeletedNode()
+    {
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub1).delete());
+
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+        assertEquals("Should have returned node for 2nd subscription", _sub2, _node.getSubscription());
+
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+        assertEquals("Should have returned node for 3rd subscription", _sub3, _node.getSubscription());
+    }
+
+    /**
+     * Test that if a central node in the list is deleted (but is still present),
+     * it is not returned when searching through the list for the next viable node,
+     * and the subsequent viable node is returned instead.
+     */
+    public void testFindNextSkipsCentralDeletedNode()
+    {
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub2).delete());
+
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+        assertEquals("Should have returned node for 3rd subscription", _sub3, _node.getSubscription());
+    }
+
+    /**
+     * Test that if the last node in the list is deleted (but is still present),
+     * it is not returned when searching through the list for the next viable node,
+     * and null is returned instead.
+     */
+    public void testFindNextSkipsLastDeletedNode()
+    {
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+        assertEquals("Should have returned node for 1st subscription", _sub1, _node.getSubscription());
+
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+        assertEquals("Should have returned node for 2nd subscription", _sub2, _node.getSubscription());
+
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub3).delete());
+
+        assertNull("Returned node should be null", _node = _node.findNext());
+    }
+
+    /**
+     * Test that if multiple nodes in the list are deleted (but still present), they
+     * are not returned when searching through the list for the next viable node,
+     * and the subsequent viable node is returned instead.
+     */
+    public void testFindNextSkipsMultipleDeletedNode()
+    {
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub1).delete());
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub2).delete());
+
+        assertNotNull("Returned node should not be null", _node = _node.findNext());
+        assertEquals("Should have returned node for 3rd subscription", _sub3, _node.getSubscription());
+    }
+
+    /**
+     * Test that if a node in the list is marked 'deleted' it is still present in the list
+     * until actually removed. counter-test to verify above testing of getNext() method.
+     */
+    public void testDeletedNodeStillPresent()
+    {
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub1).delete());
+
+        assertNotNull("Node marked deleted should still be present", getNodeForSubscription(_subList, _sub1));
+        assertEquals("All 3 nodes are still expected to be present", 3, countNodes(_subList));
+    }
+
+    /**
+     * Traverses the list nodes in a non-mutating fashion, returning the first node which matches the given
+     * Subscription, or null if none is found.
+     */
+    private SubscriptionNode getNodeForSubscription(final SubscriptionList list, final Subscription sub)
+    {
+        SubscriptionNode node = list.getHead();
+        while (node != null && node.getSubscription() != sub)
+        {
+            node = node.nextNode();
+        }
+
+        return node;
+    }
+
+    /**
+     * Counts the number of (non-head) nodes in the list.
+     */
+    private int countNodes(final SubscriptionList list)
+    {
+        SubscriptionNode node = list.getHead();
+        int count;
+        for(count = -1; node != null; count++)
+        {
+            node = node.nextNode();
+        }
+
+        return count;
+    }
+
+    /**
+     * Tests that the head is returned as expected, and isn't the node for the first subscription.
+     */
+    public void testGetHead()
+    {
+        assertNotNull("List head should be non null", _node);
+        assertNotSame("Head should not be node for first subscription",
+                _node, getNodeForSubscription(_subList, _sub1));
+    }
+
+    /**
+     * Tests that the size is returned correctly in the face of additions and removals.
+     */
+    public void testGetSize()
+    {
+        SubscriptionList subList = new SubscriptionList();
+
+        assertEquals("Unexpected size result", 0, subList.size());
+
+        Subscription sub1 = new MockSubscription();
+        Subscription sub2 = new MockSubscription();
+        Subscription sub3 = new MockSubscription();
+
+        subList.add(sub1);
+        assertEquals("Unexpected size result", 1, subList.size());
+
+        subList.add(sub2);
+        assertEquals("Unexpected size result", 2, subList.size());
+
+        subList.add(sub3);
+        assertEquals("Unexpected size result", 3, subList.size());
+
+        assertTrue("Removing subscription from list should have succeeded", subList.remove(sub1));
+        assertEquals("Unexpected size result", 2, subList.size());
+
+        assertTrue("Removing subscription from list should have succeeded", subList.remove(sub2));
+        assertEquals("Unexpected size result", 1, subList.size());
+
+        assertTrue("Removing subscription from list should have succeeded", subList.remove(sub3));
+        assertEquals("Unexpected size result", 0, subList.size());
+    }
+
+    /**
+     * Test that if the first (non-head) node in the list is removed it is no longer
+     * present in the node structure of the list at all.
+     */
+    public void testRemoveFirstNode()
+    {
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub1));
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub1));
+        assertNull("Should not have been a node present for the removed subscription", getNodeForSubscription(_subList, _sub1));
+        assertEquals("Unexpected number of nodes", 2, countNodes(_subList));
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub2));
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub3));
+    }
+
+    /**
+     * Test that if a central node in the list is removed it is no longer
+     * present in the node structure of the list at all.
+     */
+    public void testRemoveCentralNode()
+    {
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub2));
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub2));
+        assertNull("Should not have been a node present for the removed subscription", getNodeForSubscription(_subList, _sub2));
+        assertEquals("Unexpected number of nodes", 2, countNodes(_subList));
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub1));
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub3));
+    }
+
+    /**
+     * Test that if the subscription contained in the last node of the list is removed
+     * it is no longer present in the node structure of the list at all. However,
+     * as the last node in the structure can't actually be removed a dummy will instead
+     * be present.
+     */
+    public void testRemoveLastNode()
+    {
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub3));
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub3));
+        assertNull("Should not have been a node present for the removed subscription", getNodeForSubscription(_subList, _sub3));
+
+        //We actually expect 3 nodes to remain this time, because the last node cant be removed for thread safety reasons,
+        //however a dummy final node can be used as substitute to allow removal of the subscription node.
+        assertEquals("Unexpected number of nodes", 2 + 1, countNodes(_subList));
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub1));
+        assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub2));
+    }
+
+    /**
+     * Test that if the subscription not contained in the list is requested to be removed
+     * that the removal fails
+     */
+    public void testRemoveNonExistantNode()
+    {
+        Subscription sub4 = new MockSubscription();
+        assertNull("Should not have been a node present for the subscription", getNodeForSubscription(_subList, sub4));
+        assertFalse("Removing subscription node should not have succeeded", _subList.remove(sub4));
+        assertEquals("Unexpected number of nodes", 3, countNodes(_subList));
+    }
+
+    /**
+     * Test that if a subscription node which occurs later in the main list than the marked node is
+     * removed from the list after the marked node is also removed, then the marker node doesn't
+     * serve to retain the subsequent nodes in the list structure (and thus memory) despite their
+     * removal.
+     */
+    public void testDeletedMarkedNodeDoesntLeakSubsequentlyDeletedNodes()
+    {
+        //get the nodes out the list for the 1st and 3rd subscriptions
+        SubscriptionNode sub1Node = getNodeForSubscription(_subList, _sub1);
+        assertNotNull("Should have been a node present for the subscription", sub1Node);
+        SubscriptionNode sub3Node = getNodeForSubscription(_subList, _sub3);
+        assertNotNull("Should have been a node present for the subscription", sub3Node);
+
+        //mark the first subscription node
+        assertTrue("should have succeeded in updating the marked node",
+                _subList.updateMarkedNode(_subList.getMarkedNode(), sub1Node));
+
+        //remove the 1st subscription from the list
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub1));
+        //verify the 1st subscription is no longer the marker node (replaced by a dummy), or in the main list structure
+        assertNotSame("Unexpected marker node", sub1Node, _subList.getMarkedNode());
+        assertNull("Should not have been a node present in the list structure for the marked-but-removed sub1 node",
+                getNodeForSubscription(_subList, _sub1));
+
+        //remove the 2nd subscription from the list
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub2));
+
+        //verify the marker node isn't leaking subsequently removed nodes, by ensuring the very next node
+        //in its list structure is now the 3rd subscription (since the 2nd was removed too)
+        assertEquals("Unexpected next node", sub3Node, _subList.getMarkedNode().nextNode());
+
+        //remove the 3rd and final/tail subscription
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub3));
+
+        //verify the marker node isn't leaking subsequently removed nodes, by ensuring the very next node
+        //in its list structure is now the dummy tail (since the 3rd subscription was removed, and a dummy
+        //tail was inserted) and NOT the 3rd sub node.
+        assertNotSame("Unexpected next node", sub3Node, _subList.getMarkedNode().nextNode());
+        assertTrue("Unexpected next node", _subList.getMarkedNode().nextNode().isDeleted());
+        assertNull("Next non-deleted node from the marker should now be the list end, i.e. null", _subList.getMarkedNode().findNext());
+    }
+
+    /**
+     * Test that setting the marked node to null doesn't cause problems during remove operations
+     */
+    public void testRemoveWithNullMarkedNode()
+    {
+        //set the marker to null
+        assertTrue("should have succeeded in updating the marked node",
+                _subList.updateMarkedNode(_subList.getMarkedNode(), null));
+
+        //remove the 1st subscription from the main list
+        assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub1));
+
+        //verify the 1st subscription is no longer in the main list structure
+        assertNull("Should not have been a node present in the main list structure for sub1",
+                getNodeForSubscription(_subList, _sub1));
+        assertEquals("Unexpected number of nodes", 2, countNodes(_subList));
+    }
+
+    /**
+     * Tests that after the first (non-head) node of the list is marked deleted but has not
+     * yet been removed, the iterator still skips it.
+     */
+    public void testIteratorSkipsFirstDeletedNode()
+    {
+        //'delete' but dont remove the node for the 1st subscription
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub1).delete());
+        assertNotNull("Should still have been a node present for the deleted subscription",
+                getNodeForSubscription(_subList, _sub1));
+
+        SubscriptionNodeIterator iter = _subList.iterator();
+
+        //verify the iterator returns the 2nd subscriptions node
+        assertTrue("Iterator should have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", _sub2, iter.getNode().getSubscription());
+
+        //verify the iterator returns the 3rd subscriptions node and not the 2nd.
+        assertTrue("Iterator should have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", _sub3, iter.getNode().getSubscription());
+    }
+
+    /**
+     * Tests that after a central node of the list is marked deleted but has not yet been removed,
+     * the iterator still skips it.
+     */
+    public void testIteratorSkipsCentralDeletedNode()
+    {
+        //'delete' but dont remove the node for the 2nd subscription
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub2).delete());
+        assertNotNull("Should still have been a node present for the deleted subscription",
+                getNodeForSubscription(_subList, _sub2));
+
+        SubscriptionNodeIterator iter = _subList.iterator();
+
+        //verify the iterator returns the 1st subscriptions node
+        assertTrue("Iterator should have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", _sub1, iter.getNode().getSubscription());
+
+        //verify the iterator returns the 3rd subscriptions node and not the 2nd.
+        assertTrue("Iterator should have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", _sub3, iter.getNode().getSubscription());
+    }
+
+    /**
+     * Tests that after the last node of the list is marked deleted but has not yet been removed,
+     * the iterator still skips it.
+     */
+    public void testIteratorSkipsDeletedFinalNode()
+    {
+        //'delete' but dont remove the node for the 3rd subscription
+        assertTrue("Deleting subscription node should have succeeded",
+                getNodeForSubscription(_subList, _sub3).delete());
+        assertNotNull("Should still have been a node present for the deleted 3rd subscription",
+                getNodeForSubscription(_subList, _sub3));
+
+        SubscriptionNodeIterator iter = _subList.iterator();
+
+        //verify the iterator returns the 1st subscriptions node
+        assertTrue("Iterator should have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", _sub1, iter.getNode().getSubscription());
+
+        //verify the iterator returns the 2nd subscriptions node
+        assertTrue("Iterator should have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", _sub2, iter.getNode().getSubscription());
+
+        //verify the iterator can no longer advance and does not return a subscription node
+        assertFalse("Iterator should not have been able to advance", iter.advance());
+        assertEquals("Iterator returned unexpected SubscriptionNode", null, iter.getNode());
+    }
+}