Refine documentation

Author: amarziali

utils/queue-utils/src/main/java/datadog/common/queue/BaseQueue.java

@@ -8,6 +8,11 @@
 import java.util.function.Supplier;
 import javax.annotation.Nonnull;
 
+/**
+ * Base class for non-blocking queuing operations.
+ *
+ * @param <E> the type of elements held by this queue
+ */
 abstract class BaseQueue<E> extends AbstractQueue<E> implements NonBlockingQueue<E> {
   /** The capacity of the queue (must be a power of two) */
   protected final int capacity;
@@ -20,32 +25,13 @@ public BaseQueue(int capacity) {
     this.mask = this.capacity - 1;
   }
 
-  /**
-   * Drains all available elements from the queue to a consumer.
-   *
-   * <p>This is efficient since it avoids repeated size() checks and returns immediately when empty.
-   *
-   * @param consumer a consumer to accept elements
-   * @return number of elements drained
-   */
   @Override
   public int drain(Consumer<E> consumer) {
     return drain(consumer, Integer.MAX_VALUE);
   }
 
-  /**
-   * Drains up to {@code limit} elements from the queue to a consumer.
-   *
-   * <p>This method is useful for batch processing.
-   *
-   * <p>Each element is removed atomically using poll() and passed to the consumer.
-   *
-   * @param consumer a consumer to accept elements
-   * @param limit maximum number of elements to drain
-   * @return number of elements drained
-   */
   @Override
-  public int drain(Consumer<E> consumer, int limit) {
+  public int drain(@Nonnull Consumer<E> consumer, int limit) {
     int count = 0;
     E e;
     while (count < limit && (e = poll()) != null) {
@@ -55,14 +41,6 @@ public int drain(Consumer<E> consumer, int limit) {
     return count;
   }
 
-  /**
-   * Fills the queue with elements provided by the supplier until either: - the queue is full, or -
-   * the supplier runs out of elements (returns null)
-   *
-   * @param supplier a supplier of elements
-   * @param limit maximum number of elements to attempt to insert
-   * @return number of elements successfully enqueued
-   */
   @Override
   public int fill(@Nonnull Supplier<? extends E> supplier, int limit) {
     if (limit <= 0) {
@@ -95,21 +73,11 @@ public Iterator<E> iterator() {
     throw new UnsupportedOperationException();
   }
 
-  /**
-   * Returns the remaining capacity.
-   *
-   * @return number of additional elements this queue can accept
-   */
   @Override
   public int remainingCapacity() {
     return capacity - size();
   }
 
-  /**
-   * Returns the maximum queue capacity.
-   *
-   * @return number of total elements this queue can accept
-   */
   @Override
   public int capacity() {
     return capacity;

utils/queue-utils/src/main/java/datadog/common/queue/BlockingConsumerNonBlockingQueue.java

@@ -3,8 +3,45 @@
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
 
+/**
+ * A hybrid queue interface combining non-blocking producer semantics with blocking consumer
+ * operations.
+ *
+ * <p>This interface extends {@link NonBlockingQueue} and adds methods that allow consumers to block
+ * while waiting for elements to become available. It is intended for use in scenarios with:
+ *
+ * <ul>
+ *   <li>Multiple or single <b>producers</b> enqueue elements using non-blocking operations (e.g.,
+ *       {@link #offer(Object)}).
+ *   <li>A single <b>consumer</b> that may block until elements are ready (i.e., using {@link
+ *       #take()} or {@link #poll(long, TimeUnit)}).
+ * </ul>
+ *
+ * @param <E> the type of elements held in this queue
+ */
 public interface BlockingConsumerNonBlockingQueue<E> extends NonBlockingQueue<E> {
+
+  /**
+   * Retrieves and removes the head of this queue, waiting up to the specified wait time if
+   * necessary for an element to become available.
+   *
+   * @param timeout how long to wait before giving up, in units of {@code unit}
+   * @param unit the time unit of the {@code timeout} argument; must not be {@code null}
+   * @return the head of this queue, or {@code null} if the specified waiting time elapses before an
+   *     element becomes available
+   * @throws InterruptedException if interrupted while waiting
+   */
   E poll(long timeout, @Nonnull TimeUnit unit) throws InterruptedException;
 
+  /**
+   * Retrieves and removes the head of this queue, waiting if necessary until an element becomes
+   * available.
+   *
+   * <p>This operation blocks the consumer thread if the queue is empty, while producers continue to
+   * operate in a non-blocking manner.
+   *
+   * @return the head of this queue
+   * @throws InterruptedException if interrupted while waiting
+   */
   E take() throws InterruptedException;
 }

utils/queue-utils/src/main/java/datadog/common/queue/JctoolsMpscBlockingConsumerWrappedQueue.java

@@ -5,6 +5,15 @@
 import javax.annotation.Nonnull;
 import org.jctools.queues.MpscBlockingConsumerArrayQueue;
 
+/**
+ * A {@link BlockingConsumerNonBlockingQueue} implementation that wraps a JCTools {@link
+ * MpscBlockingConsumerArrayQueue}.
+ *
+ * <p>All operations delegate directly to the underlying JCTools queue to preserve performance and
+ * memory semantics.
+ *
+ * @param <E> the type of elements held in this queue
+ */
 class JctoolsMpscBlockingConsumerWrappedQueue<E> extends JctoolsWrappedQueue<E>
     implements BlockingConsumerNonBlockingQueue<E> {
 

utils/queue-utils/src/main/java/datadog/common/queue/JctoolsWrappedQueue.java

@@ -7,6 +7,15 @@
 import javax.annotation.Nonnull;
 import org.jctools.queues.MessagePassingQueue;
 
+/**
+ * A {@link NonBlockingQueue} implementation that wraps a {@link MessagePassingQueue} from the
+ * JCTools library to provide a consistent, framework-independent interface.
+ *
+ * <p>This adapter bridges JCTools’ queue APIs with the {@link NonBlockingQueue} abstraction used by
+ * this library. All operations are directly delegated to the underlying {@code MessagePassingQueue}
+ *
+ * @param <E> the type of elements held in this queue
+ */
 class JctoolsWrappedQueue<E> extends AbstractQueue<E> implements NonBlockingQueue<E> {
   private final MessagePassingQueue<E> delegate;
 

utils/queue-utils/src/main/java/datadog/common/queue/MpscArrayQueueVarHandle.java

@@ -72,12 +72,6 @@ public MpscArrayQueueVarHandle(int requestedCapacity) {
     this.producerLimit = capacity;
   }
 
-  /**
-   * Attempts to add an element to the queue.
-   *
-   * @param e the element to add (must be non-null)
-   * @return true if element was enqueued, false if queue is full
-   */
   @Override
   public boolean offer(E e) {
     Objects.requireNonNull(e);
@@ -131,11 +125,6 @@ public boolean offer(E e) {
     }
   }
 
-  /**
-   * Removes and returns the next element, or null if empty.
-   *
-   * @return dequeued element, or null if queue empty
-   */
   @Override
   @SuppressWarnings("unchecked")
   public E poll() {
@@ -159,27 +148,13 @@ public E poll() {
     return (E) value;
   }
 
-  /**
-   * Returns next element without removing it.
-   *
-   * <p>The memory visibility is only correct if the consumer calls it.
-   *
-   * @return next element or null if empty
-   */
   @Override
   @SuppressWarnings("unchecked")
   public E peek() {
     final int index = (int) ((long) HEAD_HANDLE.getOpaque(this) & mask);
     return (E) ARRAY_HANDLE.getVolatile(buffer, index);
   }
 
-  /**
-   * Returns number of elements in queue.
-   *
-   * <p>Volatile reads of tail and head ensure accurate result in multi-threaded context.
-   *
-   * @return current size
-   */
   @Override
   public int size() {
     long currentHead = (long) HEAD_HANDLE.getVolatile(this);

utils/queue-utils/src/main/java/datadog/common/queue/MpscBlockingConsumerArrayQueueVarHandle.java

@@ -85,12 +85,6 @@ public MpscBlockingConsumerArrayQueueVarHandle(int requestedCapacity) {
     this.producerLimit = capacity;
   }
 
-  /**
-   * Attempts to add an element to the queue.
-   *
-   * @param e the element to add (must be non-null)
-   * @return true if element was enqueued, false if queue is full
-   */
   @Override
   public boolean offer(E e) {
     Objects.requireNonNull(e);
@@ -151,11 +145,6 @@ public boolean offer(E e) {
     }
   }
 
-  /**
-   * Removes and returns the next element, or null if empty.
-   *
-   * @return dequeued element, or null if queue empty
-   */
   @Override
   @SuppressWarnings("unchecked")
   public E poll() {
@@ -179,14 +168,6 @@ public E poll() {
     return (E) value;
   }
 
-  /**
-   * Polls with a timeout.
-   *
-   * @param timeout max wait time
-   * @param unit time unit
-   * @return the head element, or null if timed out
-   * @throws InterruptedException if interrupted
-   */
   @Override
   public E poll(long timeout, @Nonnull TimeUnit unit) throws InterruptedException {
     E e = poll();
@@ -204,12 +185,6 @@ public E poll(long timeout, @Nonnull TimeUnit unit) throws InterruptedException
     return poll();
   }
 
-  /**
-   * Retrieves and removes the head element, waiting if necessary until one becomes available.
-   *
-   * @return the next element (never null)
-   * @throws InterruptedException if interrupted while waiting
-   */
   @Override
   public E take() throws InterruptedException {
     consumerThread = Thread.currentThread();
@@ -220,27 +195,13 @@ public E take() throws InterruptedException {
     return e;
   }
 
-  /**
-   * Returns next element without removing it.
-   *
-   * <p>The memory visibility is only correct if the consumer calls it.
-   *
-   * @return next element or null if empty
-   */
   @Override
   @SuppressWarnings("unchecked")
   public E peek() {
     final int index = (int) ((long) HEAD_HANDLE.getOpaque(this) & mask);
     return (E) ARRAY_HANDLE.getVolatile(buffer, index);
   }
 
-  /**
-   * Returns number of elements in queue.
-   *
-   * <p>Volatile reads of tail and head ensure accurate result in multi-threaded context.
-   *
-   * @return current size
-   */
   @Override
   public int size() {
     long currentHead = (long) HEAD_HANDLE.getVolatile(this);
@@ -256,7 +217,7 @@ public int size() {
    *
    * @param nanos max wait time in nanoseconds. If negative, it will park indefinably until waken or
    *     interrupted
-   * @throws InterruptedException if interrupted
+   * @throws InterruptedException if interrupted while waiting
    */
   private void parkUntilNext(long nanos) throws InterruptedException {
     Thread current = Thread.currentThread();

utils/queue-utils/src/main/java/datadog/common/queue/NonBlockingQueue.java

@@ -5,14 +5,68 @@
 import java.util.function.Supplier;
 import javax.annotation.Nonnull;
 
+/**
+ * A non-blocking, concurrent queue supporting high-performance operations for producer-consumer
+ * scenarios. This interface extends {@link Queue} and adds specialized methods for bulk draining
+ * and filling, as well as querying the queue’s fixed capacity.
+ *
+ * <p>Unlike typical {@link java.util.concurrent.BlockingQueue} implementations, this interface does
+ * not provide blocking operations. Instead, producers and consumers are expected to retry or yield
+ * when the queue is full or empty, respectively.
+ *
+ * <p>Implementations are typically array-backed and rely on non-blocking atomic operations (such as
+ * VarHandles or Unsafe-based CAS) to achieve concurrent performance without locks.
+ *
+ * @param <E> the type of elements held in this queue
+ * @see java.util.Queue
+ * @see java.util.concurrent.ConcurrentLinkedQueue
+ */
 public interface NonBlockingQueue<E> extends Queue<E> {
+
+  /**
+   * Drains all available elements from this queue, passing each to the given {@link Consumer}.
+   *
+   * <p>This method will consume as many elements as are currently available, up to the queue’s size
+   * at the time of the call.
+   *
+   * @param consumer the consumer that will process each element; must not be {@code null}
+   * @return the number of elements drained
+   * @throws NullPointerException if {@code consumer} is {@code null}
+   */
   int drain(Consumer<E> consumer);
 
+  /**
+   * Drains up to the specified number of elements from this queue, passing each to the given {@link
+   * Consumer}.
+   *
+   * @param consumer the consumer that will process each element; must not be {@code null}
+   * @param limit the maximum number of elements to drain
+   * @return the actual number of elements drained (maybe less than {@code limit})
+   */
   int drain(Consumer<E> consumer, int limit);
 
+  /**
+   * Fills the queue with elements supplied by the given {@link Supplier}, up to the specified limit
+   * or until the queue becomes full.
+   *
+   * @param supplier the supplier that provides elements to insert; must not be {@code null}
+   * @param limit the maximum number of elements to insert
+   * @return the number of elements successfully added (maybe less than {@code limit})
+   */
   int fill(@Nonnull Supplier<? extends E> supplier, int limit);
 
+  /**
+   * Returns the number of additional elements that can be inserted into this queue without
+   * exceeding its capacity.
+   *
+   * @return the number of remaining slots available for insertion
+   */
   int remainingCapacity();
 
+  /**
+   * Returns the total fixed capacity of this queue.
+   *
+   * @return the maximum number of elements this queue can hold
+   */
   int capacity();
 }