Introduce varhandle based queues

Author: amarziali

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

@@ -16,6 +16,8 @@ dependencies {
   implementation project(':communication')
   implementation project(':components:json')
   implementation project(':internal-api')
+  api project(':internal-api:internal-api-9')
+
 
   testImplementation project(':dd-java-agent:testing')
 }

dd-java-agent/agent-llmobs/src/main/java/datadog/trace/llmobs/EvalProcessingWorker.java

@@ -12,7 +12,8 @@
 import datadog.communication.http.OkHttpUtils;
 import datadog.trace.api.Config;
 import datadog.trace.llmobs.domain.LLMObsEval;
-import datadog.trace.util.queue.MpscArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -34,7 +35,7 @@ public class EvalProcessingWorker implements AutoCloseable {
 
   private static final Logger log = LoggerFactory.getLogger(EvalProcessingWorker.class);
 
-  private final MpscArrayQueue<LLMObsEval> queue;
+  private final BaseQueue<LLMObsEval> queue;
   private final Thread serializerThread;
 
   public EvalProcessingWorker(
@@ -43,7 +44,7 @@ public EvalProcessingWorker(
       final TimeUnit timeUnit,
       final SharedCommunicationObjects sco,
       Config config) {
-    this.queue = new MpscArrayQueue<>(capacity);
+    this.queue = Queues.mpscArrayQueue(capacity);
 
     boolean isAgentless = config.isLlmObsAgentlessEnabled();
     if (isAgentless && (config.getApiKey() == null || config.getApiKey().isEmpty())) {
@@ -98,7 +99,7 @@ public static class EvalSerializingHandler implements Runnable {
     private static final Logger log = LoggerFactory.getLogger(EvalSerializingHandler.class);
     private static final int FLUSH_THRESHOLD = 50;
 
-    private final MpscArrayQueue<LLMObsEval> queue;
+    private final BaseQueue<LLMObsEval> queue;
     private final long ticksRequiredToFlush;
     private long lastTicks;
 
@@ -111,7 +112,7 @@ public static class EvalSerializingHandler implements Runnable {
     private final List<LLMObsEval> buffer = new ArrayList<>();
 
     public EvalSerializingHandler(
-        final MpscArrayQueue<LLMObsEval> queue,
+        final BaseQueue<LLMObsEval> queue,
         final long flushInterval,
         final TimeUnit timeUnit,
         final HttpUrl submissionUrl,

dd-trace-core/build.gradle

@@ -66,6 +66,7 @@ dependencies {
   api project(':dd-trace-api')
   api project(':communication')
   api project(':internal-api')
+  api project(':internal-api:internal-api-9')
   implementation project(':components:json')
   implementation project(':utils:container-utils')
   implementation project(':utils:socket-utils')

dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java

@@ -4,7 +4,7 @@
 
 import datadog.trace.common.metrics.SignalItem.StopSignal;
 import datadog.trace.core.util.LRUCache;
-import datadog.trace.util.queue.MpscArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.Iterator;
 import java.util.Map;
@@ -23,7 +23,7 @@ final class Aggregator implements Runnable {
   private static final Logger log = LoggerFactory.getLogger(Aggregator.class);
 
   private final Queue<Batch> batchPool;
-  private final MpscArrayQueue<InboxItem> inbox;
+  private final BaseQueue<InboxItem> inbox;
   private final LRUCache<MetricKey, AggregateMetric> aggregates;
   private final ConcurrentMap<MetricKey, Batch> pending;
   private final Set<MetricKey> commonKeys;
@@ -43,7 +43,7 @@ final class Aggregator implements Runnable {
   Aggregator(
       MetricWriter writer,
       Queue<Batch> batchPool,
-      MpscArrayQueue<InboxItem> inbox,
+      BaseQueue<InboxItem> inbox,
       ConcurrentMap<MetricKey, Batch> pending,
       final Set<MetricKey> commonKeys,
       int maxAggregates,
@@ -64,7 +64,7 @@ final class Aggregator implements Runnable {
   Aggregator(
       MetricWriter writer,
       Queue<Batch> batchPool,
-      MpscArrayQueue<InboxItem> inbox,
+      BaseQueue<InboxItem> inbox,
       ConcurrentMap<MetricKey, Batch> pending,
       final Set<MetricKey> commonKeys,
       int maxAggregates,

dd-trace-core/src/main/java/datadog/trace/common/metrics/ConflatingMetricsAggregator.java

@@ -33,8 +33,8 @@
 import datadog.trace.core.DDTraceCoreInfo;
 import datadog.trace.core.monitor.HealthMetrics;
 import datadog.trace.util.AgentTaskScheduler;
-import datadog.trace.util.queue.MpscArrayQueue;
-import datadog.trace.util.queue.SpmcArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -94,7 +94,7 @@ public final class ConflatingMetricsAggregator implements MetricsAggregator, Eve
   private final ConcurrentHashMap<MetricKey, Batch> pending;
   private final ConcurrentHashMap<MetricKey, MetricKey> keys;
   private final Thread thread;
-  private final MpscArrayQueue<InboxItem> inbox;
+  private final BaseQueue<InboxItem> inbox;
   private final Sink sink;
   private final Aggregator aggregator;
   private final long reportingInterval;
@@ -177,8 +177,8 @@ public ConflatingMetricsAggregator(
       long reportingInterval,
       TimeUnit timeUnit) {
     this.ignoredResources = ignoredResources;
-    this.inbox = new MpscArrayQueue<>(queueSize);
-    this.batchPool = new SpmcArrayQueue<>(maxAggregates);
+    this.inbox = Queues.mpscArrayQueue(queueSize);
+    this.batchPool = Queues.spmcArrayQueue(maxAggregates);
     this.pending = new ConcurrentHashMap<>(maxAggregates * 4 / 3);
     this.keys = new ConcurrentHashMap<>();
     this.features = features;

dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java

@@ -10,7 +10,8 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import datadog.trace.util.AgentTaskScheduler;
-import datadog.trace.util.queue.SpscArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Collections;
@@ -36,7 +37,7 @@ public final class OkHttpSink implements Sink, EventListener {
   private final OkHttpClient client;
   private final HttpUrl metricsUrl;
   private final List<EventListener> listeners;
-  private final SpscArrayQueue<Request> enqueuedRequests = new SpscArrayQueue<>(16);
+  private final BaseQueue<Request> enqueuedRequests = Queues.spscArrayQueue(16);
   private final AtomicLong lastRequestTime = new AtomicLong();
   private final AtomicLong asyncRequestCounter = new AtomicLong();
   private final boolean bufferingEnabled;
@@ -157,9 +158,9 @@ private void handleFailure(okhttp3.Response response) throws IOException {
 
   private static final class Sender implements AgentTaskScheduler.Task<OkHttpSink> {
 
-    private final SpscArrayQueue<Request> inbox;
+    private final BaseQueue<Request> inbox;
 
-    private Sender(SpscArrayQueue<Request> inbox) {
+    private Sender(BaseQueue<Request> inbox) {
       this.inbox = inbox;
     }
 

dd-trace-core/src/main/java/datadog/trace/common/writer/SpanSamplingWorker.java

@@ -9,7 +9,8 @@
 import datadog.trace.common.sampling.SingleSpanSampler;
 import datadog.trace.core.DDSpan;
 import datadog.trace.core.monitor.HealthMetrics;
-import datadog.trace.util.queue.MpscBlockingConsumerArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Queue;
@@ -44,7 +45,7 @@ class DefaultSpanSamplingWorker implements SpanSamplingWorker {
 
     private final Thread spanSamplingThread;
     private final SamplingHandler samplingHandler;
-    private final MpscBlockingConsumerArrayQueue<Object> spanSamplingQueue;
+    private final BaseQueue<Object> spanSamplingQueue;
     private final Queue<Object> primaryQueue;
     private final Queue<Object> secondaryQueue;
     private final SingleSpanSampler singleSpanSampler;
@@ -61,7 +62,7 @@ protected DefaultSpanSamplingWorker(
         DroppingPolicy droppingPolicy) {
       this.samplingHandler = new SamplingHandler();
       this.spanSamplingThread = newAgentThread(SPAN_SAMPLING_PROCESSOR, samplingHandler);
-      this.spanSamplingQueue = new MpscBlockingConsumerArrayQueue<>(capacity);
+      this.spanSamplingQueue = Queues.mpscBlockingConsumerArrayQueue(capacity);
       this.primaryQueue = primaryQueue;
       this.secondaryQueue = secondaryQueue;
       this.singleSpanSampler = singleSpanSampler;
@@ -171,7 +172,7 @@ public void onEvent(Object event) {
         }
       }
 
-      private void consumeBatch(MpscBlockingConsumerArrayQueue<Object> queue) {
+      private void consumeBatch(BaseQueue<Object> queue) {
         queue.drain(this::onEvent, queue.size());
       }
     }

dd-trace-core/src/main/java/datadog/trace/common/writer/TraceProcessingWorker.java

@@ -15,7 +15,8 @@
 import datadog.trace.core.CoreSpan;
 import datadog.trace.core.DDSpan;
 import datadog.trace.core.monitor.HealthMetrics;
-import datadog.trace.util.queue.MpscArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -35,8 +36,8 @@ public class TraceProcessingWorker implements AutoCloseable {
   private static final Logger log = LoggerFactory.getLogger(TraceProcessingWorker.class);
 
   private final PrioritizationStrategy prioritizationStrategy;
-  private final MpscArrayQueue<Object> primaryQueue;
-  private final MpscArrayQueue<Object> secondaryQueue;
+  private final BaseQueue<Object> primaryQueue;
+  private final BaseQueue<Object> secondaryQueue;
   private final TraceSerializingHandler serializingHandler;
   private final Thread serializerThread;
   private final int capacity;
@@ -120,23 +121,23 @@ public long getRemainingCapacity() {
     return primaryQueue.remainingCapacity();
   }
 
-  private static MpscArrayQueue<Object> createQueue(int capacity) {
-    return new MpscArrayQueue<>(capacity);
+  private static BaseQueue<Object> createQueue(int capacity) {
+    return Queues.mpscArrayQueue(capacity);
   }
 
   public static class TraceSerializingHandler implements Runnable {
 
-    private final MpscArrayQueue<Object> primaryQueue;
-    private final MpscArrayQueue<Object> secondaryQueue;
+    private final BaseQueue<Object> primaryQueue;
+    private final BaseQueue<Object> secondaryQueue;
     private final HealthMetrics healthMetrics;
     private final long ticksRequiredToFlush;
     private final boolean doTimeFlush;
     private final PayloadDispatcher payloadDispatcher;
     private long lastTicks;
 
     public TraceSerializingHandler(
-        final MpscArrayQueue<Object> primaryQueue,
-        final MpscArrayQueue<Object> secondaryQueue,
+        final BaseQueue<Object> primaryQueue,
+        final BaseQueue<Object> secondaryQueue,
         final HealthMetrics healthMetrics,
         final PayloadDispatcher payloadDispatcher,
         final long flushInterval,
@@ -237,7 +238,7 @@ private boolean shouldFlush() {
       return false;
     }
 
-    private void consumeBatch(MpscArrayQueue<Object> queue) {
+    private void consumeBatch(BaseQueue<Object> queue) {
       queue.drain(this::onEvent, queue.size());
     }
 

dd-trace-core/src/main/java/datadog/trace/core/PendingTraceBuffer.java

@@ -12,7 +12,8 @@
 import datadog.trace.api.time.TimeSource;
 import datadog.trace.common.writer.TraceDumpJsonExporter;
 import datadog.trace.core.monitor.HealthMetrics;
-import datadog.trace.util.queue.MpscBlockingConsumerArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -62,7 +63,7 @@ private static class DelayingPendingTraceBuffer extends PendingTraceBuffer {
     private static final CommandElement DUMP_ELEMENT = new CommandElement();
     private static final CommandElement STAND_IN_ELEMENT = new CommandElement();
 
-    private final MpscBlockingConsumerArrayQueue<Element> queue;
+    private final BaseQueue<Element> queue;
     private final Thread worker;
     private final TimeSource timeSource;
 
@@ -292,7 +293,7 @@ public DelayingPendingTraceBuffer(
         Config config,
         SharedCommunicationObjects sharedCommunicationObjects,
         HealthMetrics healthMetrics) {
-      this.queue = new MpscBlockingConsumerArrayQueue<>(bufferSize);
+      this.queue = Queues.mpscBlockingConsumerArrayQueue(bufferSize);
       this.worker = newAgentThread(TRACE_MONITOR, new Worker());
       this.timeSource = timeSource;
       boolean runningSpansEnabled = config.isLongRunningTraceEnabled();

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java

@@ -29,7 +29,8 @@
 import datadog.trace.core.DDSpan;
 import datadog.trace.core.DDTraceCoreInfo;
 import datadog.trace.util.AgentTaskScheduler;
-import datadog.trace.util.queue.MpscArrayQueue;
+import datadog.trace.util.queue.BaseQueue;
+import datadog.trace.util.queue.Queues;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -53,7 +54,7 @@ public class DefaultDataStreamsMonitoring implements DataStreamsMonitoring, Even
       new StatsPoint(DataStreamsTags.EMPTY, 0, 0, 0, 0, 0, 0, 0, null);
 
   private final Map<Long, Map<String, StatsBucket>> timeToBucket = new HashMap<>();
-  private final MpscArrayQueue<InboxItem> inbox = new MpscArrayQueue<>(1024);
+  private final BaseQueue<InboxItem> inbox = Queues.mpscArrayQueue(1024);
   private final DatastreamsPayloadWriter payloadWriter;
   private final DDAgentFeaturesDiscovery features;
   private final TimeSource timeSource;