Update JCTools Code

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/util/unsafe/ConcurrentCircularArrayQueue.java

@@ -21,11 +21,27 @@
 import java.util.AbstractQueue;
 import java.util.Iterator;
 
-abstract class ConcurrentCircularArrayQueueL0Pad<E> extends AbstractQueue<E>{
+abstract class ConcurrentCircularArrayQueueL0Pad<E> extends AbstractQueue<E> implements MessagePassingQueue<E> {
     long p00, p01, p02, p03, p04, p05, p06, p07;
     long p30, p31, p32, p33, p34, p35, p36, p37;
 }
 
+/**
+ * A concurrent access enabling class used by circular array based queues this class exposes an offset computation
+ * method along with differently memory fenced load/store methods into the underlying array. The class is pre-padded and
+ * the array is padded on either side to help with False sharing prvention. It is expected theat subclasses handle post
+ * padding.
+ * <p>
+ * Offset calculation is separate from access to enable the reuse of a give compute offset.
+ * <p>
+ * Load/Store methods using a <i>buffer</i> parameter are provided to allow the prevention of final field reload after a
+ * LoadLoad barrier.
+ * <p>
+ * 
+ * @author nitsanw
+ * 
+ * @param <E>
+ */
 public abstract class ConcurrentCircularArrayQueue<E> extends ConcurrentCircularArrayQueueL0Pad<E> {
     protected static final int SPARSE_SHIFT = Integer.getInteger("sparse.shift", 0);
     protected static final int BUFFER_PAD = 32;
@@ -51,87 +67,108 @@ public abstract class ConcurrentCircularArrayQueue<E> extends ConcurrentCircular
 
     @SuppressWarnings("unchecked")
     public ConcurrentCircularArrayQueue(int capacity) {
-        this.capacity = Pow2.findNextPositivePowerOfTwo(capacity);
+        this.capacity = Pow2.roundToPowerOfTwo(capacity);
         mask = this.capacity - 1;
         // pad data on either end with some empty slots.
         buffer = (E[]) new Object[(this.capacity << SPARSE_SHIFT) + BUFFER_PAD * 2];
     }
 
-    public ConcurrentCircularArrayQueue(ConcurrentCircularArrayQueue<E> c) {
-        this.capacity = c.capacity;
-        this.mask = c.mask;
-        // pad data on either end with some empty slots.
-        this.buffer = c.buffer;
-    }
-
-    protected final long calcOffset(long index) {
+    /**
+     * @param index desirable element index
+     * @return the offset in bytes within the array for a given index.
+     */
+    protected final long calcElementOffset(long index) {
         return REF_ARRAY_BASE + ((index & mask) << REF_ELEMENT_SHIFT);
     }
 
+    /**
+     * A plain store (no ordering/fences) of an element to a given offset
+     * 
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @param e a kitty
+     */
     protected final void spElement(long offset, E e) {
-        UNSAFE.putObject(buffer, offset, e);
-    }
-
-    protected final void soElement(long offset, E e) {
-        UNSAFE.putOrderedObject(buffer, offset, e);
-    }
-
-    protected final void svElement(long offset, E e) {
-        UNSAFE.putObjectVolatile(buffer, offset, e);
-    }
-
-    @SuppressWarnings("unchecked")
-    protected final E lpElement(long offset) {
-        return (E) UNSAFE.getObject(buffer, offset);
-    }
-
-    @SuppressWarnings("unchecked")
-    protected final E lvElement(long offset) {
-        return (E) UNSAFE.getObjectVolatile(buffer, offset);
+        spElement(buffer, offset, e);
     }
 
+    /**
+     * A plain store (no ordering/fences) of an element to a given offset
+     * 
+     * @param buffer this.buffer
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @param e an orderly kitty
+     */
     protected final void spElement(E[] buffer, long offset, E e) {
         UNSAFE.putObject(buffer, offset, e);
     }
 
+    /**
+     * An ordered store(store + StoreStore barrier) of an element to a given offset
+     * 
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @param e an orderly kitty
+     */
+    protected final void soElement(long offset, E e) {
+        soElement(buffer, offset, e);
+    }
+
+    /**
+     * An ordered store(store + StoreStore barrier) of an element to a given offset
+     * 
+     * @param buffer this.buffer
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @param e an orderly kitty
+     */
     protected final void soElement(E[] buffer, long offset, E e) {
         UNSAFE.putOrderedObject(buffer, offset, e);
     }
 
-    protected final void svElement(E[] buffer, long offset, E e) {
-        UNSAFE.putObjectVolatile(buffer, offset, e);
+    /**
+     * A plain load (no ordering/fences) of an element from a given offset.
+     * 
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @return the element at the offset
+     */
+    protected final E lpElement(long offset) {
+        return lpElement(buffer, offset);
     }
 
+    /**
+     * A plain load (no ordering/fences) of an element from a given offset.
+     * 
+     * @param buffer this.buffer
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @return the element at the offset
+     */
     @SuppressWarnings("unchecked")
     protected final E lpElement(E[] buffer, long offset) {
         return (E) UNSAFE.getObject(buffer, offset);
     }
 
+    /**
+     * A volatile load (load + LoadLoad barrier) of an element from a given offset.
+     * 
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @return the element at the offset
+     */
+    protected final E lvElement(long offset) {
+        return lvElement(buffer, offset);
+    }
+
+    /**
+     * A volatile load (load + LoadLoad barrier) of an element from a given offset.
+     * 
+     * @param buffer this.buffer
+     * @param offset computed via {@link ConcurrentCircularArrayQueue#calcElementOffset(long)}
+     * @return the element at the offset
+     */
     @SuppressWarnings("unchecked")
     protected final E lvElement(E[] buffer, long offset) {
         return (E) UNSAFE.getObjectVolatile(buffer, offset);
     }
 
-    @Override
-    public boolean offer(E e) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public E poll() {
-        throw new UnsupportedOperationException();
-    }
-    @Override
-    public E peek() {
-        throw new UnsupportedOperationException();
-    }
     @Override
     public Iterator<E> iterator() {
         throw new UnsupportedOperationException();
     }
-
-    @Override
-    public int size() {
-        throw new UnsupportedOperationException();
-    }
 }

rxjava-core/src/main/java/rx/internal/util/unsafe/ConcurrentSequencedCircularArrayQueue.java

@@ -29,8 +29,7 @@ public abstract class ConcurrentSequencedCircularArrayQueue<E> extends Concurren
             throw new IllegalStateException("Unexpected long[] element size");
         }
         // Including the buffer pad in the array base offset
-        ARRAY_BASE = UnsafeAccess.UNSAFE.arrayBaseOffset(long[].class)
-                + (BUFFER_PAD << (ELEMENT_SHIFT - SPARSE_SHIFT));
+        ARRAY_BASE = UnsafeAccess.UNSAFE.arrayBaseOffset(long[].class) + (BUFFER_PAD << (ELEMENT_SHIFT - SPARSE_SHIFT));
     }
     protected final long[] sequenceBuffer;
 
@@ -39,56 +38,19 @@ public ConcurrentSequencedCircularArrayQueue(int capacity) {
         // pad data on either end with some empty slots.
         sequenceBuffer = new long[(this.capacity << SPARSE_SHIFT) + BUFFER_PAD * 2];
         for (long i = 0; i < this.capacity; i++) {
-            soSequenceElement(calcSequenceOffset(i), i);
+            soSequence(sequenceBuffer, calcSequenceOffset(i), i);
         }
     }
 
-    public ConcurrentSequencedCircularArrayQueue(ConcurrentSequencedCircularArrayQueue<E> c) {
-        super(c);
-        this.sequenceBuffer = c.sequenceBuffer;
-    }
-
     protected final long calcSequenceOffset(long index) {
         return ARRAY_BASE + ((index & mask) << ELEMENT_SHIFT);
     }
 
-    protected final void spSequenceElement(long offset, long e) {
-        UNSAFE.putLong(sequenceBuffer, offset, e);
-    }
-
-    protected final void soSequenceElement(long offset, long e) {
-        UNSAFE.putOrderedLong(sequenceBuffer, offset, e);
-    }
-
-    protected final void svSequenceElement(long offset, long e) {
-        UNSAFE.putLongVolatile(sequenceBuffer, offset, e);
-    }
-
-    protected final long lpSequenceElement(long offset) {
-        return UNSAFE.getLong(sequenceBuffer, offset);
-    }
-
-    protected final long lvSequenceElement(long offset) {
-        return UNSAFE.getLongVolatile(sequenceBuffer, offset);
-    }
-
-    protected final void spSequenceElement(long[] buffer, long offset, long e) {
-        UNSAFE.putLong(buffer, offset, e);
-    }
-
-    protected final void soSequenceElement(long[] buffer, long offset, long e) {
+    protected final void soSequence(long[] buffer, long offset, long e) {
         UNSAFE.putOrderedLong(buffer, offset, e);
     }
 
-    protected final void svSequenceElement(long[] buffer, long offset, long e) {
-        UNSAFE.putLongVolatile(buffer, offset, e);
-    }
-
-    protected final long lpSequenceElement(long[] buffer, long offset) {
-        return UNSAFE.getLong(buffer, offset);
-    }
-
-    protected final long lvSequenceElement(long[] buffer, long offset) {
+    protected final long lvSequence(long[] buffer, long offset) {
         return UNSAFE.getLongVolatile(buffer, offset);
     }