Class MpscFFLamportOffHeapFixedSizeRingBuffer


public final class MpscFFLamportOffHeapFixedSizeRingBuffer extends OffHeapFixedMessageSizeRingBuffer
A Multi-Producer-Single-Consumer ring buffer. This implies that any thread may call the write methods, but only a single thread may call reads for correctness to maintained.
This implementation follows patterns for False Sharing protection.
This implementation is using the Fast Flow method for polling from the queue (with minor change to correctly publish the index) and an extension of the Leslie Lamport concurrent queue algorithm (originated by Martin Thompson) on the producer side.
  • Field Details

    • consumerIndexCacheAddress

      private final long consumerIndexCacheAddress
  • Constructor Details

    • MpscFFLamportOffHeapFixedSizeRingBuffer

      public MpscFFLamportOffHeapFixedSizeRingBuffer(int capacity, int primitiveMessageSize, int referenceMessageSize)
    • MpscFFLamportOffHeapFixedSizeRingBuffer

      protected MpscFFLamportOffHeapFixedSizeRingBuffer(ByteBuffer buff, int capacity, boolean isProducer, boolean isConsumer, boolean initialize, int primitiveMessageSize, Object[] references, int referenceMessageSize)
      This is to be used for an IPC queue with the direct buffer used being a memory mapped file.
      Parameters:
      buff -
      capacity -
  • Method Details

    • slowPathWriteAcquire

      private long slowPathWriteAcquire(long wrapPoint)
    • writeAcquire

      protected final long writeAcquire()
      Description copied from class: ProxyChannelRingBuffer
      Acquire an offset to write to. If there's no space available a wait strategy may be used.
      Specified by:
      writeAcquire in class OffHeapFixedMessageSizeRingBuffer
      Returns:
      a base address for a message acquired to be written, or EOF if none is available
    • writeRelease

      protected final void writeRelease(long offset)
      Specified by:
      writeRelease in class OffHeapFixedMessageSizeRingBuffer
      Parameters:
      offset - the base address of a message that we are done writing and can be read now
    • writeRelease

      protected final void writeRelease(long offset, int callTypeId)
      Description copied from class: ProxyChannelRingBuffer
      Ordered store of the callTypeId for the message at offset
      Specified by:
      writeRelease in class ProxyChannelRingBuffer
      Parameters:
      offset - the offset that was released for writing
      callTypeId - A unique ID for the call
    • readAcquire

      protected final long readAcquire()
      Description copied from class: ProxyChannelRingBuffer
      Acquire an offset to read from
      Specified by:
      readAcquire in class OffHeapFixedMessageSizeRingBuffer
      Returns:
      a base address for a message acquired to be read, or EOF if none is available
    • readRelease

      protected final void readRelease(long offset)
      Description copied from class: ProxyChannelRingBuffer
      Release the offset from reading
      Specified by:
      readRelease in class OffHeapFixedMessageSizeRingBuffer
      Parameters:
      offset - the base address of a message that we are done reading and can be overwritten now
    • casProducerIndex

      private boolean casProducerIndex(long expected, long update)
    • lpConsumerIndexCache

      private long lpConsumerIndexCache()
    • spConsumerIndexCache

      private void spConsumerIndexCache(long value)
    • spReadReleaseState

      private static void spReadReleaseState(long offset)