Issue-228: Add unorderedSnapshot method to MpscArrayQueue and BaseMps…

jctools-core/src/main/java/org/jctools/queues/BaseMpscLinkedArrayQueue.java

@@ -26,9 +26,13 @@
 import static org.jctools.queues.LinkedArrayQueueUtil.modifiedCalcElementOffset;
 import static org.jctools.util.UnsafeAccess.UNSAFE;
 import static org.jctools.util.UnsafeAccess.fieldOffset;
+import static org.jctools.util.UnsafeRefArrayAccess.calcElementOffset;
 import static org.jctools.util.UnsafeRefArrayAccess.lvElement;
 import static org.jctools.util.UnsafeRefArrayAccess.soElement;
 
+import java.util.ArrayList;
+import java.util.List;
+
 abstract class BaseMpscLinkedArrayQueuePad1<E> extends AbstractQueue<E> implements IndexedQueue
 {
     long p01, p02, p03, p04, p05, p06, p07;
@@ -73,13 +77,22 @@ abstract class BaseMpscLinkedArrayQueueConsumerFields<E> extends BaseMpscLinkedA
     private volatile long consumerIndex;
     protected long consumerMask;
     protected E[] consumerBuffer;
+    private volatile E[] volatileConsumerBuffer;
 
     @Override
     public final long lvConsumerIndex()
     {
         return consumerIndex;
     }
 
+    final E[] lvVolatileConsumerBuffer() {
+        return volatileConsumerBuffer;
+    }
+
+    final void svVolatileConsumerBuffer(E[] newValue) {
+        volatileConsumerBuffer = newValue;
+    }
+
     final long lpConsumerIndex()
     {
         return UNSAFE.getLong(this, C_INDEX_OFFSET);
@@ -157,6 +170,7 @@ public BaseMpscLinkedArrayQueue(final int initialCapacity)
         producerBuffer = buffer;
         producerMask = mask;
         consumerBuffer = buffer;
+        svVolatileConsumerBuffer(buffer);
         consumerMask = mask;
         soProducerLimit(mask); // we know it's all empty to start with
     }
@@ -403,7 +417,6 @@ private E[] getNextBuffer(final E[] buffer, final long mask)
     {
         final long offset = nextArrayOffset(mask);
         final E[] nextBuffer = (E[]) lvElement(buffer, offset);
-        soElement(buffer, offset, null);
         return nextBuffer;
     }
 
@@ -439,6 +452,7 @@ private E newBufferPeek(E[] nextBuffer, long index)
     private long newBufferAndOffset(E[] nextBuffer, long index)
     {
         consumerBuffer = nextBuffer;
+        svVolatileConsumerBuffer(nextBuffer);
         consumerMask = (length(nextBuffer) - 2) << 1;
         return modifiedCalcElementOffset(index, consumerMask);
     }
@@ -640,6 +654,30 @@ public void drain(Consumer<E> c, WaitStrategy w, ExitCondition exit)
             c.accept(e);
         }
     }
+
+    public List<E> unorderedSnapshot() {
+        E[] currentBuffer = lvVolatileConsumerBuffer();
+        List<E> elements = new ArrayList<E>();
+        while (true) {
+            int length = length(currentBuffer);
+            for (int i = 0; i < length - 1; i++) {
+                long offset = calcElementOffset(i);
+                Object element = lvElement(currentBuffer, offset);
+                if (element == JUMP || element == null) {
+                    continue;
+                }
+                elements.add((E) element);
+            }
+            long offset = calcElementOffset((length - 1));
+            Object nextArray = lvElement(currentBuffer, offset);
+            if (nextArray != null) {
+                currentBuffer = (E[]) nextArray;
+            } else {
+                break;
+            }
+        }
+        return elements;
+    }
 
     private void resize(long oldMask, E[] oldBuffer, long pIndex, E e)
     {

jctools-core/src/main/java/org/jctools/queues/MpscArrayQueue.java

@@ -19,6 +19,9 @@
 import static org.jctools.util.UnsafeAccess.fieldOffset;
 import static org.jctools.util.UnsafeRefArrayAccess.*;
 
+import java.util.ArrayList;
+import java.util.List;
+
 abstract class MpscArrayQueueL1Pad<E> extends ConcurrentCircularArrayQueue<E>
 {
     long p00, p01, p02, p03, p04, p05, p06, p07;
@@ -562,4 +565,17 @@ public void fill(Supplier<E> s, WaitStrategy w, ExitCondition exit)
             idleCounter = 0;
         }
     }
+
+    public List<E> unorderedSnapshot() {
+        int length = capacity();
+        List<E> elements = new ArrayList<E>();
+        for (int i = 0; i < length; i++) {
+            long offset = calcElementOffset(i);
+            E element = lvElement(buffer, offset);
+            if (element != null) {
+                elements.add(element);
+            }
+        }
+        return elements;
+    }
 }

jctools-core/src/main/java/org/jctools/queues/atomic/BaseMpscLinkedAtomicArrayQueue.java

@@ -21,6 +21,8 @@
 import java.util.Iterator;
 import static org.jctools.queues.atomic.LinkedAtomicArrayQueueUtil.length;
 import static org.jctools.queues.atomic.LinkedAtomicArrayQueueUtil.modifiedCalcElementOffset;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import org.jctools.queues.MessagePassingQueue;
@@ -92,11 +94,21 @@ abstract class BaseMpscLinkedAtomicArrayQueueConsumerFields<E> extends BaseMpscL
 
     protected AtomicReferenceArray<E> consumerBuffer;
 
+    private volatile AtomicReferenceArray<E> volatileConsumerBuffer;
+
     @Override
     public final long lvConsumerIndex() {
         return consumerIndex;
     }
 
+    final AtomicReferenceArray<E> lvVolatileConsumerBuffer() {
+        return volatileConsumerBuffer;
+    }
+
+    final void svVolatileConsumerBuffer(AtomicReferenceArray<E> newValue) {
+        volatileConsumerBuffer = newValue;
+    }
+
     final long lpConsumerIndex() {
         return consumerIndex;
     }
@@ -181,6 +193,7 @@ public BaseMpscLinkedAtomicArrayQueue(final int initialCapacity) {
         producerBuffer = buffer;
         producerMask = mask;
         consumerBuffer = buffer;
+        svVolatileConsumerBuffer(buffer);
         consumerMask = mask;
         // we know it's all empty to start with
         soProducerLimit(mask);
@@ -373,7 +386,6 @@ private int offerSlowPath(long mask, long pIndex, long producerLimit) {
     private AtomicReferenceArray<E> getNextBuffer(final AtomicReferenceArray<E> buffer, final long mask) {
         final int offset = nextArrayOffset(mask);
         final AtomicReferenceArray<E> nextBuffer = (AtomicReferenceArray<E>) lvElement(buffer, offset);
-        soElement(buffer, offset, null);
         return nextBuffer;
     }
 
@@ -406,6 +418,7 @@ private E newBufferPeek(AtomicReferenceArray<E> nextBuffer, long index) {
 
     private int newBufferAndOffset(AtomicReferenceArray<E> nextBuffer, long index) {
         consumerBuffer = nextBuffer;
+        svVolatileConsumerBuffer(nextBuffer);
         consumerMask = (length(nextBuffer) - 2) << 1;
         return modifiedCalcElementOffset(index, consumerMask);
     }
@@ -571,6 +584,30 @@ public void drain(Consumer<E> c, WaitStrategy w, ExitCondition exit) {
         }
     }
 
+    public List<E> unorderedSnapshot() {
+        AtomicReferenceArray<E> currentBuffer = lvVolatileConsumerBuffer();
+        List<E> elements = new ArrayList<E>();
+        while (true) {
+            int length = length(currentBuffer);
+            for (int i = 0; i < length - 1; i++) {
+                int offset = calcElementOffset(i);
+                Object element = lvElement(currentBuffer, offset);
+                if (element == JUMP || element == null) {
+                    continue;
+                }
+                elements.add((E) element);
+            }
+            int offset = calcElementOffset((length - 1));
+            Object nextArray = lvElement(currentBuffer, offset);
+            if (nextArray != null) {
+                currentBuffer = (AtomicReferenceArray<E>) nextArray;
+            } else {
+                break;
+            }
+        }
+        return elements;
+    }
+
     private void resize(long oldMask, AtomicReferenceArray<E> oldBuffer, long pIndex, E e) {
         int newBufferLength = getNextBufferSize(oldBuffer);
         final AtomicReferenceArray<E> newBuffer = allocate(newBufferLength);

jctools-core/src/main/java/org/jctools/queues/atomic/LinkedAtomicArrayQueueUtil.java

@@ -55,6 +55,11 @@ static int modifiedCalcElementOffset(long index, long mask)
     {
         return (int) (index & mask) >> 1;
     }
+
+    static int calcElementOffset(long index)
+    {
+        return (int) index;
+    }
 
     static int nextArrayOffset(AtomicReferenceArray<?> curr)
     {

jctools-core/src/main/java/org/jctools/queues/atomic/MpscAtomicArrayQueue.java

@@ -14,6 +14,8 @@
 package org.jctools.queues.atomic;
 
 import org.jctools.util.PortableJvmInfo;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 import java.util.concurrent.atomic.AtomicLongArray;
@@ -533,6 +535,19 @@ public void fill(Supplier<E> s, WaitStrategy w, ExitCondition exit) {
         }
     }
 
+    public List<E> unorderedSnapshot() {
+        int length = capacity();
+        List<E> elements = new ArrayList<E>();
+        for (int i = 0; i < length; i++) {
+            int offset = calcElementOffset(i);
+            E element = lvElement(buffer, offset);
+            if (element != null) {
+                elements.add(element);
+            }
+        }
+        return elements;
+    }
+
     /**
      * @deprecated This was renamed to failFastOffer please migrate
      */

jctools-core/src/test/java/org/jctools/queues/MpscArrayQueueSnapshotTest.java

@@ -0,0 +1,52 @@
+/*
+ * 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.jctools.queues;
+
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Before;
+import org.junit.Test;
+
+public class MpscArrayQueueSnapshotTest {
+
+    private MpscArrayQueue<Integer> queue;
+
+    @Before
+    public void setUp() throws Exception {
+        this.queue = new MpscArrayQueue<>(4);
+    }
+
+    @Test
+    public void testSnapshot() {
+        queue.offer(0);
+        assertThat(queue.unorderedSnapshot(), contains(0));
+        for (int i = 1; i < queue.capacity(); i++) {
+            queue.offer(i);
+        }
+        assertThat(queue.unorderedSnapshot(), containsInAnyOrder(0, 1, 2, 3));
+        queue.poll();
+        queue.offer(4);
+        assertThat(queue.unorderedSnapshot(), containsInAnyOrder(1, 2, 3, 4));
+    }
+
+}

jctools-core/src/test/java/org/jctools/queues/MpscUnboundedArrayQueueSnapshotTest.java

@@ -0,0 +1,59 @@
+/*
+ * 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.jctools.queues;
+
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Before;
+import org.junit.Test;
+
+public class MpscUnboundedArrayQueueSnapshotTest {
+
+    private static final int CHUNK_SIZE = 4;
+    private MpscUnboundedArrayQueue<Integer> queue;
+
+    @Before
+    public void setUp() throws Exception {
+        this.queue = new MpscUnboundedArrayQueue<>(CHUNK_SIZE);
+    }
+
+    @Test
+    public void testSnapshot() {
+        queue.offer(0);
+        assertThat(queue.unorderedSnapshot(), contains(0));
+        for (int i = 1; i < CHUNK_SIZE; i++) {
+            queue.offer(i);
+        }
+        assertThat(queue.unorderedSnapshot(), containsInAnyOrder(0, 1, 2, 3));
+        queue.offer(4);
+        queue.offer(5);
+        assertThat(queue.unorderedSnapshot(), containsInAnyOrder(0, 1, 2, 3, 4, 5));
+        queue.poll();
+        assertThat(queue.unorderedSnapshot(), containsInAnyOrder(1, 2, 3, 4, 5));
+        for (int i = 1; i < CHUNK_SIZE; i++) {
+            queue.poll();
+        }
+        assertThat(queue.unorderedSnapshot(), containsInAnyOrder(4, 5));
+    }
+
+}