ARTEMIS-2015 PriorityLinkedListImpl::isEmpty is not thread-safe

PriorityLinkedListImpl::size access is changed to be safely
observable by a thread different from the one allowed to write
the list.

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/PriorityLinkedList.java

@@ -18,7 +18,8 @@ package org.apache.activemq.artemis.utils.collections;
 
 /**
  * A type of linked list which maintains items according to a priority
- * and allows adding and removing of elements at both ends, and peeking
+ * and allows adding and removing of elements at both ends, and peeking.<br>
+ * Only {@link #size()} and {@link #isEmpty()} are safe to be called concurrently.
  */
 public interface PriorityLinkedList<T> {
 
@@ -30,9 +31,17 @@ public interface PriorityLinkedList<T> {
 
    void clear();
 
+   /**
+    * Returns the size of this list.<br>
+    * It is safe to be called concurrently.
+    */
    int size();
 
    LinkedListIterator<T> iterator();
 
+   /**
+    * Returns {@code true} if empty, {@code false} otherwise.<br>
+    * It is safe to be called concurrently.
+    */
    boolean isEmpty();
 }

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/PriorityLinkedListImpl.java

@@ -18,6 +18,7 @@ package org.apache.activemq.artemis.utils.collections;
 
 import java.lang.reflect.Array;
 import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 /**
  * A priority linked list implementation
@@ -26,9 +27,11 @@ import java.util.NoSuchElementException;
  */
 public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> {
 
+   private static final AtomicIntegerFieldUpdater<PriorityLinkedListImpl> SIZE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(PriorityLinkedListImpl.class, "size");
+
    protected LinkedListImpl<T>[] levels;
 
-   private int size;
+   private volatile int size;
 
    private int lastReset;
 
@@ -65,7 +68,7 @@ public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> {
 
       levels[priority].addHead(t);
 
-      size++;
+      exclusiveIncrementSize(1);
    }
 
    @Override
@@ -74,7 +77,7 @@ public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> {
 
       levels[priority].addTail(t);
 
-      size++;
+      exclusiveIncrementSize(1);
    }
 
    @Override
@@ -94,7 +97,7 @@ public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> {
             t = ll.poll();
 
             if (t != null) {
-               size--;
+               exclusiveIncrementSize(-1);
 
                if (ll.size() == 0) {
                   if (highestPriority == i) {
@@ -116,7 +119,15 @@ public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> {
          list.clear();
       }
 
-      size = 0;
+      exclusiveSetSize(0);
+   }
+
+   private void exclusiveIncrementSize(int amount) {
+      SIZE_UPDATER.lazySet(this, this.size + amount);
+   }
+
+   private void exclusiveSetSize(int value) {
+      SIZE_UPDATER.lazySet(this, value);
    }
 
    @Override
@@ -242,7 +253,7 @@ public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T> {
             highestPriority = i;
          }
 
-         size--;
+         exclusiveIncrementSize(-1);
       }
    }
 }