rx.internal.util.unsafe

Class ConcurrentCircularArrayQueue<E>

  • Type Parameters:
    E -
    All Implemented Interfaces:
    Iterable<E>, Collection<E>, Queue<E>
    Direct Known Subclasses:
    ConcurrentSequencedCircularArrayQueue, SpmcArrayQueue, SpscArrayQueue 

    public abstract class ConcurrentCircularArrayQueue<E>
extends AbstractQueue<E>
    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.

    Offset calculation is separate from access to enable the reuse of a give compute offset.

    Load/Store methods using a buffer parameter are provided to allow the prevention of final field reload after a LoadLoad barrier.

    • Method Summary

      Modifier and Type Method and Description
      protected long calcElementOffset(long index)
       
      boolean isEmpty()
      This method's accuracy is subject to concurrent modifications happening as the observation is carried out.
      Iterator<E> iterator()
       
      protected E lpElement(E[] buffer, long offset)
      A plain load (no ordering/fences) of an element from a given offset.
      protected E lpElement(long offset)
      A plain load (no ordering/fences) of an element from a given offset.
      protected E lvElement(E[] buffer, long offset)
      A volatile load (load + LoadLoad barrier) of an element from a given offset.
      protected E lvElement(long offset)
      A volatile load (load + LoadLoad barrier) of an element from a given offset.
      boolean offer(M message)
      Called from a producer thread subject to the restrictions appropriate to the implementation and according to the Queue.offer(Object) interface (but failure to offer doesn't necessitate queue is full).
      M peek()
      Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue.peek() interface (barring the hard requirement on null returns).
      M poll()
      Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue.poll() interface (barring the hard requirement on null returns).
      int size()
      This method's accuracy is subject to concurrent modifications happening as the size is estimated and as such is a best effort rather than absolute value.
      protected void soElement(E[] buffer, long offset, E e)
      An ordered store(store + StoreStore barrier) of an element to a given offset
      protected void soElement(long offset, E e)
      An ordered store(store + StoreStore barrier) of an element to a given offset
      protected void spElement(E[] buffer, long offset, E e)
      A plain store (no ordering/fences) of an element to a given offset
      protected void spElement(long offset, E e)
      A plain store (no ordering/fences) of an element to a given offset

    • Field Detail

      • SPARSE_SHIFT

        protected static final int SPARSE_SHIFT

      • capacity

        protected final int capacity

      • mask

        protected final long mask

      • buffer

        protected final E[] buffer

    • Constructor Detail

      • ConcurrentCircularArrayQueue

        public ConcurrentCircularArrayQueue(int capacity)

    • Method Detail

      • calcElementOffset

        protected final long calcElementOffset(long index)
        Parameters:
        index - desirable element index
        Returns:
        the offset in bytes within the array for a given index.

      • spElement

        protected final void spElement(long offset,
                               E e)
        A plain store (no ordering/fences) of an element to a given offset
        Parameters:
        offset - computed via calcElementOffset(long)
        e - a kitty

      • spElement

        protected final void spElement(E[] buffer,
                               long offset,
                               E e)
        A plain store (no ordering/fences) of an element to a given offset
        Parameters:
        buffer - this.buffer
        offset - computed via calcElementOffset(long)
        e - an orderly kitty

      • soElement

        protected final void soElement(long offset,
                               E e)
        An ordered store(store + StoreStore barrier) of an element to a given offset
        Parameters:
        offset - computed via calcElementOffset(long)
        e - an orderly kitty

      • soElement

        protected final void soElement(E[] buffer,
                               long offset,
                               E e)
        An ordered store(store + StoreStore barrier) of an element to a given offset
        Parameters:
        buffer - this.buffer
        offset - computed via calcElementOffset(long)
        e - an orderly kitty

      • lpElement

        protected final E lpElement(long offset)
        A plain load (no ordering/fences) of an element from a given offset.
        Parameters:
        offset - computed via calcElementOffset(long)
        Returns:
        the element at the offset

      • lpElement

        protected final E lpElement(E[] buffer,
                            long offset)
        A plain load (no ordering/fences) of an element from a given offset.
        Parameters:
        buffer - this.buffer
        offset - computed via calcElementOffset(long)
        Returns:
        the element at the offset

      • lvElement

        protected final E lvElement(long offset)
        A volatile load (load + LoadLoad barrier) of an element from a given offset.
        Parameters:
        offset - computed via calcElementOffset(long)
        Returns:
        the element at the offset

      • lvElement

        protected final E lvElement(E[] buffer,
                            long offset)
        A volatile load (load + LoadLoad barrier) of an element from a given offset.
        Parameters:
        buffer - this.buffer
        offset - computed via calcElementOffset(long)
        Returns:
        the element at the offset

      • offer

        public boolean offer(M message)
        Called from a producer thread subject to the restrictions appropriate to the implementation and according to the Queue.offer(Object) interface (but failure to offer doesn't necessitate queue is full).
        Parameters:
        message -
        Returns:
        true if element was inserted into the queue, false if cannot enqueue

      • poll

        public M poll()
        Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue.poll() interface (barring the hard requirement on null returns).
        Returns:
        a message from the queue if one is available, null otherwise(not necessarily empty)

      • peek

        public M peek()
        Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue.peek() interface (barring the hard requirement on null returns).
        Returns:
        a message from the queue if one is available, null otherwise(not necessarily empty)

      • size

        public int size()
        This method's accuracy is subject to concurrent modifications happening as the size is estimated and as such is a best effort rather than absolute value. For some implementations this method may be O(n) rather than O(1).
        Returns:
        number of messages in the queue, between 0 and queue capacity or Integer.MAX_VALUE if not bounded

      • isEmpty

        public boolean isEmpty()
        This method's accuracy is subject to concurrent modifications happening as the observation is carried out.
        Returns:
        true if empty, false otherwise