try to improve the fairness

Author: amarziali

dd-java-agent/agent-llmobs/build.gradle

@@ -17,7 +17,7 @@ dependencies {
   implementation project(':communication')
   implementation project(':components:json')
   implementation project(':internal-api')
-  api project(':internal-api:internal-api-9')
+  implementation project(':internal-api:internal-api-9')
 
 
   testImplementation project(':dd-java-agent:testing')

internal-api/build.gradle.kts

@@ -281,8 +281,6 @@ dependencies {
   // it contains annotations that are also present in the instrumented application classes
   api("com.datadoghq:dd-javac-plugin-client:0.2.2")
 
-  jmhImplementation(libs.jctools)
-
   testImplementation("org.snakeyaml:snakeyaml-engine:2.9")
   testImplementation(project(":utils:test-utils"))
   testImplementation(libs.bundles.junit5)

internal-api/internal-api-9/build.gradle.kts

@@ -39,7 +39,7 @@ val minimumInstructionCoverage by extra(0.8)
 
 dependencies {
   api(project(":internal-api"))
-  api(libs.jctools) // probably the Queues factory should be moved away from there
+  implementation(libs.jctools) // probably the Queues factory should be moved away from there
 
   testImplementation(project(":dd-java-agent:testing"))
   testImplementation(libs.slf4j)

…stacktrace/queue/SPMCQueueBenchmark.java …/MPSCBlockingConsumerQueueBenchmark.javainternal-api/internal-api-9/src/jmh/java/datadog/trace/util/stacktrace/queue/SPMCQueueBenchmark.java renamed to internal-api/internal-api-9/src/jmh/java/datadog/trace/util/queue/MPSCBlockingConsumerQueueBenchmark.java internal-api/internal-api-9/src/jmh/java/datadog/trace/util/stacktrace/queue/SPMCQueueBenchmark.java renamed to internal-api/internal-api-9/src/jmh/java/datadog/trace/util/queue/MPSCBlockingConsumerQueueBenchmark.java

@@ -1,6 +1,5 @@
-package datadog.trace.util.stacktrace.queue;
+package datadog.trace.util.queue;
 
-import datadog.trace.util.queue.SpmcArrayQueueVarHandle;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -20,58 +19,62 @@
 import org.openjdk.jmh.infra.Blackhole;
 
 /*
-MPSCQueueBenchmark.queueTest                1024  thrpt       145.261           ops/us
-MPSCQueueBenchmark.queueTest:consume        1024  thrpt        84.185           ops/us
-MPSCQueueBenchmark.queueTest:produce        1024  thrpt        61.076           ops/us
-MPSCQueueBenchmark.queueTest               65536  thrpt       187.609           ops/us
-MPSCQueueBenchmark.queueTest:consume       65536  thrpt       117.097           ops/us
-MPSCQueueBenchmark.queueTest:produce       65536  thrpt        70.512           ops/us
- */
+Benchmark                                             (capacity)   Mode  Cnt    Score   Error   Units
+MPSCBlockingConsumerQueueBenchmark.queueTest                1024  thrpt       121.534          ops/us
+MPSCBlockingConsumerQueueBenchmark.queueTest:async          1024  thrpt           NaN             ---
+MPSCBlockingConsumerQueueBenchmark.queueTest:consume        1024  thrpt       110.962          ops/us
+MPSCBlockingConsumerQueueBenchmark.queueTest:produce        1024  thrpt        10.572          ops/us
+MPSCBlockingConsumerQueueBenchmark.queueTest               65536  thrpt       126.856          ops/us
+MPSCBlockingConsumerQueueBenchmark.queueTest:async         65536  thrpt           NaN             ---
+MPSCBlockingConsumerQueueBenchmark.queueTest:consume       65536  thrpt       113.213          ops/us
+MPSCBlockingConsumerQueueBenchmark.queueTest:produce       65536  thrpt        13.644          ops/us
+*/
 @BenchmarkMode(Mode.Throughput)
 @Warmup(iterations = 1, time = 30)
 @Measurement(iterations = 1, time = 30)
 @Fork(1)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @State(Scope.Benchmark)
-public class SPMCQueueBenchmark {
+public class MPSCBlockingConsumerQueueBenchmark {
   @State(Scope.Group)
   public static class QueueState {
-    SpmcArrayQueueVarHandle<Integer> queue;
-    CountDownLatch producerReady;
+    MpscBlockingConsumerArrayQueueVarHandle<Integer> queue;
+    CountDownLatch consumerReady;
 
     @Param({"1024", "65536"})
     int capacity;
 
     @Setup(Level.Iteration)
     public void setup() {
-      queue = new SpmcArrayQueueVarHandle<>(capacity);
-      producerReady = new CountDownLatch(1);
+      queue = new MpscBlockingConsumerArrayQueueVarHandle<>(capacity);
+      consumerReady = new CountDownLatch(1);
     }
   }
 
   @Benchmark
   @Group("queueTest")
   @GroupThreads(4)
-  public void consume(QueueState state, Blackhole bh) {
+  public void produce(QueueState state) {
     try {
-      state.producerReady.await(); // wait until consumer is ready
+      state.consumerReady.await(); // wait until consumer is ready
     } catch (InterruptedException ignored) {
     }
-    Integer v = state.queue.poll();
-    if (v != null) {
-      bh.consume(v);
+
+    // bounded attempt: try once, then yield if full
+    boolean offered = state.queue.offer(0);
+    if (!offered) {
+      Thread.yield();
     }
   }
 
   @Benchmark
   @Group("queueTest")
   @GroupThreads(1)
-  public void produce(QueueState state) {
-    state.producerReady.countDown(); // signal consumers can start
-    // bounded attempt: try once, then yield if full
-    boolean offered = state.queue.offer(0);
-    if (!offered) {
-      Thread.yield();
+  public void consume(QueueState state, Blackhole bh) {
+    state.consumerReady.countDown(); // signal producers can start
+    Integer v = state.queue.poll();
+    if (v != null) {
+      bh.consume(v);
     }
   }
 }

…stacktrace/queue/MPSCQueueBenchmark.java …trace/util/queue/MPSCQueueBenchmark.javainternal-api/internal-api-9/src/jmh/java/datadog/trace/util/stacktrace/queue/MPSCQueueBenchmark.java renamed to internal-api/internal-api-9/src/jmh/java/datadog/trace/util/queue/MPSCQueueBenchmark.java internal-api/internal-api-9/src/jmh/java/datadog/trace/util/stacktrace/queue/MPSCQueueBenchmark.java renamed to internal-api/internal-api-9/src/jmh/java/datadog/trace/util/queue/MPSCQueueBenchmark.java

@@ -1,6 +1,5 @@
-package datadog.trace.util.stacktrace.queue;
+package datadog.trace.util.queue;
 
-import datadog.trace.util.queue.MpscArrayQueueVarHandle;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -21,15 +20,14 @@
 
 /*
 Benchmark                             (capacity)   Mode  Cnt    Score   Error   Units
-MPSCQueueBenchmark.queueTest               65536  thrpt       208.469          ops/us
-MPSCQueueBenchmark.queueTest:async         65536  thrpt           NaN             ---
-MPSCQueueBenchmark.queueTest:consume       65536  thrpt       199.309          ops/us
-MPSCQueueBenchmark.queueTest:produce       65536  thrpt         9.161          ops/us
-MPSCQueueBenchmark.queueTest                1024  thrpt       195.200          ops/us
+MPSCQueueBenchmark.queueTest                1024  thrpt       146.530          ops/us
 MPSCQueueBenchmark.queueTest:async          1024  thrpt           NaN             ---
-MPSCQueueBenchmark.queueTest:consume        1024  thrpt       185.929          ops/us
-MPSCQueueBenchmark.queueTest:produce        1024  thrpt         9.272          ops/us
- */
+MPSCQueueBenchmark.queueTest:consume        1024  thrpt       108.357          ops/us
+MPSCQueueBenchmark.queueTest:produce        1024  thrpt        38.172          ops/us
+MPSCQueueBenchmark.queueTest               65536  thrpt       179.177          ops/us
+MPSCQueueBenchmark.queueTest:async         65536  thrpt           NaN             ---
+MPSCQueueBenchmark.queueTest:consume       65536  thrpt       140.968          ops/us
+MPSCQueueBenchmark.queueTest:produce       65536  thrpt        38.209          ops/us */
 @BenchmarkMode(Mode.Throughput)
 @Warmup(iterations = 1, time = 30)
 @Measurement(iterations = 1, time = 30)

…stacktrace/queue/SPSCQueueBenchmark.java …trace/util/queue/SPSCQueueBenchmark.javainternal-api/internal-api-9/src/jmh/java/datadog/trace/util/stacktrace/queue/SPSCQueueBenchmark.java renamed to internal-api/internal-api-9/src/jmh/java/datadog/trace/util/queue/SPSCQueueBenchmark.java internal-api/internal-api-9/src/jmh/java/datadog/trace/util/stacktrace/queue/SPSCQueueBenchmark.java renamed to internal-api/internal-api-9/src/jmh/java/datadog/trace/util/queue/SPSCQueueBenchmark.java

@@ -1,6 +1,5 @@
-package datadog.trace.util.stacktrace.queue;
+package datadog.trace.util.queue;
 
-import datadog.trace.util.queue.SpscArrayQueueVarHandle;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;

internal-api/internal-api-9/src/main/java/datadog/trace/util/queue/BlockingConsumerNonBlockingQueue.java

@@ -4,11 +4,7 @@
 import javax.annotation.Nonnull;
 
 public interface BlockingConsumerNonBlockingQueue<E> extends NonBlockingQueue<E> {
-  boolean offer(E e, long timeout, @Nonnull TimeUnit unit) throws InterruptedException;
-
   E poll(long timeout, @Nonnull TimeUnit unit) throws InterruptedException;
 
-  void put(E e) throws InterruptedException;
-
   E take() throws InterruptedException;
 }

internal-api/internal-api-9/src/main/java/datadog/trace/util/queue/JctoolsMpscBlockingConsumerWrappedQueue.java

@@ -2,7 +2,6 @@
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.LockSupport;
 import javax.annotation.Nonnull;
 import org.jctools.queues.MpscBlockingConsumerArrayQueue;
 
@@ -17,98 +16,13 @@ public JctoolsMpscBlockingConsumerWrappedQueue(
     this.blockingQueueDelegate = delegate;
   }
 
-  @Override
-  public void put(E e) throws InterruptedException {
-    blockingQueueDelegate.put(e);
-  }
-
   @Override
   public E take() throws InterruptedException {
     return blockingQueueDelegate.take();
   }
 
-  /**
-   * Timed offer with progressive backoff.
-   *
-   * <p>Tries to insert an element into the queue within the given timeout. Uses a spin → yield →
-   * park backoff strategy to reduce CPU usage under contention.
-   *
-   * @param e the element to insert
-   * @param timeout maximum time to wait
-   * @param unit time unit of timeout
-   * @return {@code true} if inserted, {@code false} if timeout expires
-   * @throws InterruptedException if interrupted while waiting
-   */
-  @Override
-  public boolean offer(E e, long timeout, @Nonnull TimeUnit unit) throws InterruptedException {
-    final long deadline = System.nanoTime() + unit.toNanos(timeout);
-    int idleCount = 0;
-
-    while (true) {
-      if (offer(e)) {
-        return true; // successfully inserted
-      }
-
-      long remaining = deadline - System.nanoTime();
-      if (remaining <= 0) {
-        return false; // timeout
-      }
-
-      // Progressive backoff
-      if (idleCount < 100) {
-        // spin (busy-wait)
-      } else if (idleCount < 1_000) {
-        Thread.yield(); // give up CPU to other threads
-      } else {
-        // park for a short duration, up to 1 ms
-        LockSupport.parkNanos(Math.min(remaining, 1_000_000L));
-      }
-
-      idleCount++;
-
-      if (Thread.interrupted()) {
-        throw new InterruptedException();
-      }
-    }
-  }
-
-  /**
-   * Polls with a timeout using progressive backoff.
-   *
-   * @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 {
-    final long deadline = System.nanoTime() + unit.toNanos(timeout);
-    int idleCount = 0;
-
-    while (true) {
-      E value = poll();
-      if (value != null) {
-        return value;
-      }
-
-      long remaining = deadline - System.nanoTime();
-      if (remaining <= 0) {
-        return null;
-      }
-
-      // Progressive backoff
-      if (idleCount < 100) {
-        // spin
-      } else if (idleCount < 1_000) {
-        Thread.yield();
-      } else {
-        LockSupport.parkNanos(Math.min(remaining, 1_000_000L));
-      }
-      idleCount++;
-
-      if (Thread.interrupted()) {
-        throw new InterruptedException();
-      }
-    }
+    return blockingQueueDelegate.poll(timeout, unit);
   }
 }

internal-api/internal-api-9/src/main/java/datadog/trace/util/queue/MpscArrayQueueVarHandle.java

@@ -4,6 +4,7 @@
 import java.lang.invoke.MethodHandles.Lookup;
 import java.lang.invoke.VarHandle;
 import java.util.Objects;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.locks.LockSupport;
 
 /**
@@ -85,23 +86,6 @@ public boolean offer(E e) {
     // jctools does the same local copy to have the jitter optimise the accesses
     final Object[] localBuffer = this.buffer;
 
-    // depending on the thread id, choose a different backoff strategy.
-    // Note: it reduces fairness but also the contention on the cas.
-    boolean s0 = false, s1 = false, s2 = false;
-    switch ((int) (Thread.currentThread().getId() & 3)) {
-      case 0:
-        s0 = true;
-        break;
-      case 1:
-        s1 = true;
-        break;
-      case 2:
-        s2 = true;
-        break;
-      default:
-        break;
-    }
-
     long localProducerLimit = (long) PRODUCER_LIMIT_HANDLE.getVolatile(this);
     long cachedHead = 0L; // Local cache of head to reduce volatile reads
 
@@ -114,7 +98,9 @@ public boolean offer(E e) {
         cachedHead = (long) HEAD_HANDLE.getVolatile(this);
         localProducerLimit = cachedHead + capacity;
 
-        if (currentTail >= localProducerLimit) return false; // queue full
+        if (currentTail >= localProducerLimit) {
+          return false; // queue full
+        }
 
         // Update producerLimit so other producers also benefit
         PRODUCER_LIMIT_HANDLE.setVolatile(this, localProducerLimit);
@@ -130,9 +116,20 @@ public boolean offer(E e) {
       }
 
       // Backoff to reduce contention
-      if (s0) Thread.onSpinWait();
-      else if (s1) Thread.yield();
-      else if (s2) LockSupport.parkNanos(1);
+      switch (ThreadLocalRandom.current().nextInt(0, 4)) {
+        case 0:
+          Thread.yield();
+          break;
+        case 1:
+          LockSupport.parkNanos(1);
+          break;
+        case 2:
+          Thread.onSpinWait();
+          break;
+        default:
+          // busy spin
+          break;
+      }
     }
   }
 
@@ -151,7 +148,9 @@ public E poll() {
 
     // Acquire-load ensures visibility of producer write
     Object value = ARRAY_HANDLE.getAcquire(localBuffer, index);
-    if (value == null) return null;
+    if (value == null) {
+      return null;
+    }
 
     // Clear the slot without additional fence
     ARRAY_HANDLE.setOpaque(localBuffer, index, null);