Use dedicated class hierarchy to ensure the padding is applied

Author: amarziali

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

@@ -2,6 +2,7 @@
 
 import static datadog.trace.util.BitUtils.nextPowerOfTwo;
 
+import datadog.common.queue.padding.PaddedSequence;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.VarHandle;
 import java.util.AbstractQueue;
@@ -16,18 +17,13 @@
  * @param <E> the type of elements held by this queue
  */
 abstract class BaseQueue<E> extends AbstractQueue<E> implements NonBlockingQueue<E> {
-  protected static final VarHandle HEAD_HANDLE;
-  protected static final VarHandle TAIL_HANDLE;
   protected static final VarHandle ARRAY_HANDLE;
 
   static {
     try {
-      final MethodHandles.Lookup lookup = MethodHandles.lookup();
-      HEAD_HANDLE = lookup.findVarHandle(BaseQueue.class, "head", long.class);
-      TAIL_HANDLE = lookup.findVarHandle(BaseQueue.class, "tail", long.class);
       ARRAY_HANDLE = MethodHandles.arrayElementVarHandle(Object[].class);
-    } catch (ReflectiveOperationException e) {
-      throw new ExceptionInInitializerError(e);
+    } catch (Throwable t) {
+      throw new ExceptionInInitializerError(t);
     }
   }
 
@@ -45,14 +41,10 @@ abstract class BaseQueue<E> extends AbstractQueue<E> implements NonBlockingQueue
   private long p0, p1, p2, p3, p4, p5, p6;
 
   /** Next free slot for producer (single-threaded) */
-  protected volatile long tail = 0L;
-
-  // Padding around tail
-  @SuppressWarnings("unused")
-  private long q0, q1, q2, q3, q4, q5, q6;
+  protected final PaddedSequence tail = new PaddedSequence();
 
   /** Next slot to consume (multi-threaded) */
-  protected volatile long head = 0L;
+  protected final PaddedSequence head = new PaddedSequence();
 
   // Padding around head
   @SuppressWarnings("unused")
@@ -124,8 +116,8 @@ public final int capacity() {
 
   @Override
   public final int size() {
-    long currentTail = (long) TAIL_HANDLE.getVolatile(this);
-    long currentHead = (long) HEAD_HANDLE.getVolatile(this);
+    long currentTail = tail.getVolatile();
+    long currentHead = head.getVolatile();
     return (int) (currentTail - currentHead);
   }
 }

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

@@ -47,12 +47,12 @@ public boolean offer(E e) {
     boolean parkOnSpin = (Thread.currentThread().getId() & 1) == 0;
 
     while (true) {
-      long currentTail = (long) TAIL_HANDLE.getVolatile(this);
+      long currentTail = tail.getVolatile();
 
       // Check if producer limit exceeded
       if (currentTail >= localProducerLimit) {
         // Refresh head only when necessary
-        cachedHead = (long) HEAD_HANDLE.getVolatile(this);
+        cachedHead = head.getVolatile();
         localProducerLimit = cachedHead + capacity;
 
         if (currentTail >= localProducerLimit) {
@@ -64,7 +64,7 @@ public boolean offer(E e) {
       }
 
       // Attempt to claim a slot
-      if (TAIL_HANDLE.compareAndSet(this, currentTail, currentTail + 1)) {
+      if (tail.compareAndSet(currentTail, currentTail + 1)) {
         final int index = (int) (currentTail & mask);
 
         // Release-store ensures producer's write is visible to consumer
@@ -91,7 +91,7 @@ public boolean offer(E e) {
   public final E poll() {
     final Object[] localBuffer = this.buffer;
 
-    long currentHead = (long) HEAD_HANDLE.getOpaque(this);
+    long currentHead = head.getOpaque();
     final int index = (int) (currentHead & mask);
 
     // Acquire-load ensures visibility of producer write
@@ -104,15 +104,15 @@ public final E poll() {
     ARRAY_HANDLE.setOpaque(localBuffer, index, null);
 
     // Advance head using opaque write (consumer-only)
-    HEAD_HANDLE.setOpaque(this, currentHead + 1);
+    head.setOpaque(currentHead + 1);
 
     return (E) value;
   }
 
   @Override
   @SuppressWarnings("unchecked")
   public final E peek() {
-    final int index = (int) ((long) HEAD_HANDLE.getOpaque(this) & mask);
+    final int index = (int) (head.getOpaque() & mask);
     return (E) ARRAY_HANDLE.getVolatile(buffer, index);
   }
 }

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

@@ -1,8 +1,6 @@
 package datadog.common.queue;
 
-import java.lang.invoke.MethodHandles;
-import java.lang.invoke.MethodHandles.Lookup;
-import java.lang.invoke.VarHandle;
+import datadog.common.queue.padding.PaddedThread;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
@@ -18,29 +16,9 @@
  */
 class MpscBlockingConsumerArrayQueueVarHandle<E> extends MpscArrayQueueVarHandle<E>
     implements BlockingConsumerNonBlockingQueue<E> {
-  private static final VarHandle CONSUMER_THREAD_HANDLE;
-
-  static {
-    try {
-      final Lookup lookup = MethodHandles.lookup();
-      CONSUMER_THREAD_HANDLE =
-          lookup.findVarHandle(
-              MpscBlockingConsumerArrayQueueVarHandle.class, "consumerThread", Thread.class);
-    } catch (Throwable t) {
-      throw new ExceptionInInitializerError(t);
-    }
-  }
-
-  // Padding to prevent false sharing
-  @SuppressWarnings("unused")
-  private long p0, p1, p2, p3, p4, p5, p6;
 
   /** Reference to the waiting consumer thread (set atomically). */
-  private volatile Thread consumerThread;
-
-  // Padding around consumerThread
-  @SuppressWarnings("unused")
-  private long q0, q1, q2, q3, q4, q5, q6;
+  private volatile PaddedThread consumerThread;
 
   /**
    * Creates a new MPSC queue.
@@ -65,12 +43,12 @@ public final boolean offer(E e) {
     boolean parkOnSpin = (Thread.currentThread().getId() & 1) == 0;
 
     while (true) {
-      long currentTail = (long) TAIL_HANDLE.getVolatile(this);
+      long currentTail = tail.getVolatile();
 
       // Check if producer limit exceeded
       if (currentTail >= localProducerLimit) {
         // Refresh head only when necessary
-        cachedHead = (long) HEAD_HANDLE.getVolatile(this);
+        cachedHead = head.getVolatile();
         localProducerLimit = cachedHead + capacity;
 
         if (currentTail >= localProducerLimit) {
@@ -82,14 +60,14 @@ public final boolean offer(E e) {
       }
 
       // Attempt to claim a slot
-      if (TAIL_HANDLE.compareAndSet(this, currentTail, currentTail + 1)) {
+      if (tail.compareAndSet(currentTail, currentTail + 1)) {
         final int index = (int) (currentTail & mask);
 
         // Release-store ensures producer's write is visible to consumer
         ARRAY_HANDLE.setRelease(localBuffer, index, e);
 
         // Atomically clear and unpark the consumer if waiting
-        Thread c = (Thread) CONSUMER_THREAD_HANDLE.getAndSet(this, null);
+        Thread c = consumerThread.getAndSet(null);
         if (c != null) {
           LockSupport.unpark(c);
         }
@@ -130,7 +108,7 @@ public final E poll(long timeout, @Nonnull TimeUnit unit) throws InterruptedExce
 
   @Override
   public E take() throws InterruptedException {
-    consumerThread = Thread.currentThread();
+    consumerThread.setVolatile(Thread.currentThread());
     E e;
     while ((e = poll()) == null) {
       parkUntilNext(-1);
@@ -151,7 +129,7 @@ public E take() throws InterruptedException {
   private void parkUntilNext(long nanos) throws InterruptedException {
     Thread current = Thread.currentThread();
     // Publish the consumer thread (no ordering required)
-    CONSUMER_THREAD_HANDLE.setOpaque(this, current);
+    consumerThread.setOpaque(current);
     if (nanos <= 0) {
       LockSupport.park(this);
     } else {
@@ -163,6 +141,6 @@ private void parkUntilNext(long nanos) throws InterruptedException {
     }
 
     // Cleanup (no fence needed, single consumer)
-    CONSUMER_THREAD_HANDLE.setOpaque(this, null);
+    consumerThread.setOpaque(null);
   }
 }

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

@@ -35,9 +35,9 @@ public SpmcArrayQueueVarHandle(int requestedCapacity) {
   public boolean offer(E e) {
     Objects.requireNonNull(e);
 
-    long currentTail = tail;
+    long currentTail = tail.getVolatile();
     long wrapPoint = currentTail - capacity;
-    long currentHead = (long) HEAD_HANDLE.getVolatile(this);
+    long currentHead = head.getVolatile();
 
     if (wrapPoint >= currentHead) {
       return false; // queue full
@@ -49,7 +49,7 @@ public boolean offer(E e) {
     ARRAY_HANDLE.setRelease(this.buffer, index, e);
 
     // Single-producer: simple volatile write to advance tail
-    TAIL_HANDLE.setVolatile(this, currentTail + 1);
+    tail.setVolatile(currentTail + 1);
     return true;
   }
 
@@ -62,20 +62,20 @@ public E poll() {
     boolean parkOnSpin = (Thread.currentThread().getId() & 1) == 0;
 
     while (true) {
-      long currentHead = (long) HEAD_HANDLE.getVolatile(this);
+      long currentHead = head.getVolatile();
       long limit = consumerLimit; // cached tail
 
       if (currentHead >= limit) {
         // refresh limit once from tail volatile
-        limit = (long) TAIL_HANDLE.getVolatile(this);
+        limit = tail.getVolatile();
         if (currentHead >= limit) {
           return null; // queue empty
         }
         consumerLimit = limit; // update local cache
       }
 
       // Attempt to claim this slot
-      if (!HEAD_HANDLE.compareAndSet(this, currentHead, currentHead + 1)) {
+      if (!head.compareAndSet(currentHead, currentHead + 1)) {
         // CAS failed. Backoff to reduce contention
         if ((spinCycles & 1) == 0) {
           Thread.onSpinWait();
@@ -107,8 +107,8 @@ public E poll() {
   @Override
   @SuppressWarnings("unchecked")
   public E peek() {
-    long currentHead = (long) HEAD_HANDLE.getVolatile(this);
-    long currentTail = (long) TAIL_HANDLE.getVolatile(this);
+    long currentHead = head.getVolatile();
+    long currentTail = tail.getVolatile();
 
     if (currentHead >= currentTail) return null;
 

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

@@ -25,46 +25,46 @@ public SpscArrayQueueVarHandle(int requestedCapacity) {
   public boolean offer(E e) {
     Objects.requireNonNull(e);
 
-    final long currentTail = (long) TAIL_HANDLE.getOpaque(this);
+    final long currentTail = tail.getOpaque();
     final int index = (int) (currentTail & mask);
 
     if (currentTail - cachedHead >= capacity) {
       // Refresh cached head (read from consumer side)
-      cachedHead = (long) HEAD_HANDLE.getVolatile(this);
+      cachedHead = (long) head.getVolatile();
       if (currentTail - cachedHead >= capacity) {
         return false; // still full
       }
     }
 
     ARRAY_HANDLE.setRelease(buffer, index, e); // publish value
-    TAIL_HANDLE.setOpaque(this, currentTail + 1); // relaxed tail update
+    tail.setOpaque(currentTail + 1); // relaxed tail update
     return true;
   }
 
   @Override
   @SuppressWarnings("unchecked")
   public E poll() {
-    final long currentHead = (long) HEAD_HANDLE.getOpaque(this);
+    final long currentHead = head.getOpaque();
     final int index = (int) (currentHead & mask);
 
     if (currentHead >= cachedTail) {
       // refresh tail cache
-      cachedTail = (long) TAIL_HANDLE.getVolatile(this);
+      cachedTail = tail.getVolatile();
       if (currentHead >= cachedTail) {
         return null; // still empty
       }
     }
 
     Object value = ARRAY_HANDLE.getAcquire(buffer, index);
     ARRAY_HANDLE.setOpaque(buffer, index, null); // clear slot
-    HEAD_HANDLE.setOpaque(this, currentHead + 1); // relaxed head update
+    head.setOpaque(currentHead + 1); // relaxed head update
     return (E) value;
   }
 
   @Override
   @SuppressWarnings("unchecked")
   public E peek() {
-    final int index = (int) ((long) HEAD_HANDLE.getOpaque(this) & mask);
+    final int index = (int) (head.getOpaque() & mask);
     return (E) ARRAY_HANDLE.getVolatile(buffer, index);
   }
 }

utils/queue-utils/src/main/java/datadog/common/queue/padding/LhsPadding.java

@@ -0,0 +1,7 @@
+package datadog.common.queue.padding;
+
+/** Left-hand-side (LHS) padding to prevent false sharing. */
+class LhsPadding {
+  /** 7 bytes padding field to occupy space on the left side of the value. */
+  private long p1, p2, p3, p4, p5, p6, p7;
+}

utils/queue-utils/src/main/java/datadog/common/queue/padding/LongRhsPadding.java

@@ -0,0 +1,7 @@
+package datadog.common.queue.padding;
+
+/** Right-hand-side (RHS) padding to prevent false sharing. */
+class LongRhsPadding extends LongValue {
+  /** 7 bytes fields to occupy space on the right side of the value. */
+  private long p9, p10, p11, p12, p13, p14, p15;
+}

utils/queue-utils/src/main/java/datadog/common/queue/padding/LongValue.java

@@ -0,0 +1,10 @@
+package datadog.common.queue.padding;
+
+/**
+ * Holds the actual sequence value, padded on the left to prevent false sharing with preceding
+ * fields.
+ */
+class LongValue extends LhsPadding {
+  /** The volatile value being protected from false sharing. */
+  protected volatile long value;
+}