Conversant Disruptor

The PushPullConcurrentQueue<E> is a highly optimized concurrent queue designed for a single-thread writer and a single-thread reader. It uses a push-pull pattern to achieve extremely low latency, with transfers in the order of 3-5 nanoseconds on optimized environments.

Important Constraints:

• Environment: Optimized and tested on a 64-bit HotSpot JVM in an Intel x86-64 environment. Other environments should be carefully tested before production use.
• Thread Model: Best suited for a single writer and a single reader. It is noted to be potentially broken if preemption occurs after reading the tail pointer.
• Null Safety: Passing null to offer() will throw a NullPointerException.

// Initialize with a desired capacity
PushPullConcurrentQueue<String> queue = new PushPullConcurrentQueue<>(1024);

// Single writer thread
queue.offer("event_data");

// Single reader thread
String data = queue.poll();

The MPMCConcurrentQueue<E> is a bounded, lock-free, multi-producer multi-consumer queue implementation based on Dmitry Vyukov's algorithm. It is designed for high-concurrency scenarios where multiple threads are simultaneously adding and removing elements.

Key Characteristics

• Bounded Capacity: The queue has a fixed capacity. When initializing, the actual capacity will be rounded up to the next power of two.
• Lock-Free: Uses atomic operations (CAS) for high performance under contention.
• Non-Blocking offer: The offer(E e) method returns false immediately if the queue is full, rather than blocking.
• Non-Blocking poll: The poll() method returns null immediately if the queue is empty.

Core API

• MPMCConcurrentQueue(int capacity): Constructor. Sets the capacity to the next power of two $\ge$ capacity.
• boolean offer(E e): Attempts to insert an element. Returns true if successful, false if the queue is full.
• E poll(): Attempts to retrieve and remove the head of the queue. Returns null if empty.
• E peek(): Retrieves, but does not remove, the head of the queue.
• int remove(E[] e): Drains elements from the queue into the provided array e. Returns the number of elements successfully removed.
• int size(): Returns the current number of elements in the queue.
• int capacity(): Returns the total capacity of the queue.
• boolean isEmpty(): Returns true if the queue contains no elements.
• void clear(): Removes all elements from the queue.
• boolean contains(Object o): Returns true if the queue contains the specified object.

// Initialize a queue with a capacity of 1024
MPMCConcurrentQueue<String> queue = new MPMCConcurrentQueue<>(1024);

// Add an element
boolean success = queue.offer("task-1");

// Retrieve an element
String task = queue.poll();

// Bulk removal
String[] batch = new String[10];
int count = queue.remove(batch);

The MultithreadConcurrentQueue<E> implements the ConcurrentQueue<E> interface. Below are the available methods for interacting with the queue:

/**
 * Construct a blocking queue of the given fixed capacity.
 * Note: actual capacity will be the next power of two larger than capacity.
 */
public MultithreadConcurrentQueue(final int capacity)

/**
 * Attempts to insert the specified element into this queue if it is possible to do so immediately without capacity restrictions.
 * @return true if the element was added, false if the queue is full.
 */
boolean offer(E e)

/**
 * Retrieves and removes the head of this queue, or returns null if this queue is empty.
 * @return the head of this queue, or null if empty.
 */
E poll()

/**
 * Retrieves, but does not remove, the head of this queue.
 * @return the head of this queue.
 */
E peek()

/**
 * Drains the whole queue into the provided array in a single batch operation.
 * This is more efficient than calling poll() repeatedly.
 * @param e the array into which the elements of this queue are to be transferred.
 * @return the number of elements transferred.
 */
int remove(final E[] e)

/**
 * Returns the number of elements currently in the queue.
 * @return the current size.
 */
int size()

/**
 * Returns the maximum capacity of the queue.
 * @return the capacity (always a power of two).
 */
int capacity()

/**
 * Returns true if this queue contains no elements.
 * @return true if empty.
 */
boolean isEmpty()

/**
 * Removes all of the elements from this queue.
 */
void clear()

/**
 * Returns true if this queue contains the specified element.
 * @param o element whose presence in this queue is to be verified.
 * @return true if the element is present.
 */
boolean contains(Object o)

The PushPullConcurrentQueue<E> implements the ConcurrentQueue<E> interface. Below are the available methods for interacting with the queue:

/**
 * Adds an element to the tail of the queue.
 * @param e the element to add
 * @return true if successful, false if the queue is full
 * @throws NullPointerException if e is null
 */
boolean offer(E e);

/**
 * Retrieves and removes the head of this queue, or returns null if this queue is empty.
 * @return the head of this queue, or null if empty
 */
E poll();

/**
 * Removes all available elements from the queue and copies them into the provided array.
 * @param e the array into which the elements are to be placed
 * @return the number of elements removed
 */
int remove(E[] e);

/**
 * Removes all elements from the queue.
 */
void clear();

/**
 * Retrieves, but does not remove, the head of this queue.
 * @return the head of this queue, or null if empty
 */
E peek();

/**
 * Returns the current number of elements in the queue.
 * Note: This implementation may return an approximate value if preemption occurs.
 * @return the size of the queue
 */
int size();

/**
 * Returns the total capacity of the queue.
 * @return the capacity
 */
int capacity();

/**
 * Returns true if this queue contains no elements.
 * @return true if empty
 */
boolean isEmpty();

/**
 * Returns true if this queue contains the specified element.
 * @param o element whose presence in this queue is to be tested
 * @return true if this queue contains the element
 */
boolean contains(Object o);