Skip to content

Commit 3d62379

Browse files
amarzialimcculls
amarziali
and
mcculls
authored
Replace JCTools queues with VarHandle-based implementations for Java 25+ (#9896)
* Removes jctools usage for lock-free queues. * Refactor * Introduce varhandle based queues * Try to optimise more java 9+ implementation * Try to optimise varhandle version further * Remove old java 8 implementation and reintroduce jctools for 1.8 * Use a better time wait poll * try to improve the fairness * Simplify and make more fair the spin wait * Fix the wait condition * Use getAndAdd for the offer fast path * Move to queue-utils * Revert "Use getAndAdd for the offer fast path" This reverts commit 14cc59757eb00dda40ec97b56c282ebd77b70b4e. * let it inline * Refine documentation * Final updates * oups * Update bench figures * Use dedicated class hierarchy to ensure the padding is applied * Pad consumer and producer limits * Pad elements in the buffer * Change nextPowOfTwo * Incorporate latest changes - make queues race free * adjust after rebase * Backport lookahead * Update bench * Realign implementation with candidate upstream * Add junit based tests * update bench results * Avoid allocating the double - remove unused methods * Update utils/queue-utils/build.gradle.kts Co-authored-by: Stuart McCulloch <stuart.mcculloch@datadoghq.com> * Update utils/queue-utils/src/main/java/datadog/common/queue/Queues.java Co-authored-by: Stuart McCulloch <stuart.mcculloch@datadoghq.com> * Update utils/queue-utils/src/main/java/datadog/common/queue/BaseQueue.java Co-authored-by: Stuart McCulloch <stuart.mcculloch@datadoghq.com> * fix build * Add javadoc for iterator * remove unneded size * add missing line in jmh * Activate on jdk 25 onward --------- Co-authored-by: Stuart McCulloch <stuart.mcculloch@datadoghq.com>
1 parent 1a55ea4 commit 3d62379

45 files changed

Lines changed: 3536 additions & 49 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ dependencies {
1717
implementation project(':communication')
1818
implementation project(':components:json')
1919
implementation project(':internal-api')
20+
implementation project(':utils:queue-utils')
21+
2022

2123
testImplementation project(':dd-java-agent:testing')
2224
}

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

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@
66

77
import com.squareup.moshi.JsonAdapter;
88
import com.squareup.moshi.Moshi;
9+
import datadog.common.queue.MessagePassingBlockingQueue;
10+
import datadog.common.queue.Queues;
911
import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
1012
import datadog.communication.ddagent.SharedCommunicationObjects;
1113
import datadog.communication.http.HttpRetryPolicy;
@@ -20,7 +22,6 @@
2022
import okhttp3.OkHttpClient;
2123
import okhttp3.Request;
2224
import okhttp3.RequestBody;
23-
import org.jctools.queues.MpscBlockingConsumerArrayQueue;
2425
import org.slf4j.Logger;
2526
import org.slf4j.LoggerFactory;
2627

@@ -34,7 +35,7 @@ public class EvalProcessingWorker implements AutoCloseable {
3435

3536
private static final Logger log = LoggerFactory.getLogger(EvalProcessingWorker.class);
3637

37-
private final MpscBlockingConsumerArrayQueue<LLMObsEval> queue;
38+
private final MessagePassingBlockingQueue<LLMObsEval> queue;
3839
private final Thread serializerThread;
3940

4041
public EvalProcessingWorker(
@@ -43,7 +44,7 @@ public EvalProcessingWorker(
4344
final TimeUnit timeUnit,
4445
final SharedCommunicationObjects sco,
4546
Config config) {
46-
this.queue = new MpscBlockingConsumerArrayQueue<>(capacity);
47+
this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity);
4748

4849
boolean isAgentless = config.isLlmObsAgentlessEnabled();
4950
if (isAgentless && (config.getApiKey() == null || config.getApiKey().isEmpty())) {
@@ -98,7 +99,7 @@ public static class EvalSerializingHandler implements Runnable {
9899
private static final Logger log = LoggerFactory.getLogger(EvalSerializingHandler.class);
99100
private static final int FLUSH_THRESHOLD = 50;
100101

101-
private final MpscBlockingConsumerArrayQueue<LLMObsEval> queue;
102+
private final MessagePassingBlockingQueue<LLMObsEval> queue;
102103
private final long ticksRequiredToFlush;
103104
private long lastTicks;
104105

@@ -111,7 +112,7 @@ public static class EvalSerializingHandler implements Runnable {
111112
private final List<LLMObsEval> buffer = new ArrayList<>();
112113

113114
public EvalSerializingHandler(
114-
final MpscBlockingConsumerArrayQueue<LLMObsEval> queue,
115+
final MessagePassingBlockingQueue<LLMObsEval> queue,
115116
final long flushInterval,
116117
final TimeUnit timeUnit,
117118
final HttpUrl submissionUrl,

‎dd-java-agent/build.gradle‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -389,6 +389,9 @@ dependencies {
389389
sharedShadowInclude project(':utils:socket-utils'), {
390390
transitive = false
391391
}
392+
sharedShadowInclude project(':utils:queue-utils'), {
393+
transitive = false
394+
}
392395
sharedShadowInclude project(':utils:version-utils'), {
393396
transitive = false
394397
}

‎dd-trace-core/build.gradle‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ dependencies {
6969
implementation project(':components:json')
7070
implementation project(':utils:container-utils')
7171
implementation project(':utils:socket-utils')
72+
implementation project(':utils:queue-utils')
7273
// for span exception debugging
7374
compileOnly project(':dd-java-agent:agent-debugger:debugger-bootstrap')
7475

@@ -80,6 +81,8 @@ dependencies {
8081

8182
implementation group: 'com.google.re2j', name: 're2j', version: '1.7'
8283

84+
jmhImplementation(libs.jctools)
85+
8386
// We have autoservices defined in test subtree, looks like we need this to be able to properly rebuild this
8487
testAnnotationProcessor libs.autoservice.processor
8588
testCompileOnly libs.autoservice.annotation

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

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,10 @@
77
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
88
import java.util.Iterator;
99
import java.util.Map;
10-
import java.util.Queue;
1110
import java.util.Set;
1211
import java.util.concurrent.ConcurrentMap;
1312
import java.util.concurrent.TimeUnit;
1413
import org.jctools.queues.MessagePassingQueue;
15-
import org.jctools.queues.MpscCompoundQueue;
1614
import org.slf4j.Logger;
1715
import org.slf4j.LoggerFactory;
1816

@@ -22,8 +20,8 @@ final class Aggregator implements Runnable {
2220

2321
private static final Logger log = LoggerFactory.getLogger(Aggregator.class);
2422

25-
private final Queue<Batch> batchPool;
26-
private final MpscCompoundQueue<InboxItem> inbox;
23+
private final MessagePassingQueue<Batch> batchPool;
24+
private final MessagePassingQueue<InboxItem> inbox;
2725
private final LRUCache<MetricKey, AggregateMetric> aggregates;
2826
private final ConcurrentMap<MetricKey, Batch> pending;
2927
private final Set<MetricKey> commonKeys;
@@ -42,8 +40,8 @@ final class Aggregator implements Runnable {
4240

4341
Aggregator(
4442
MetricWriter writer,
45-
Queue<Batch> batchPool,
46-
MpscCompoundQueue<InboxItem> inbox,
43+
MessagePassingQueue<Batch> batchPool,
44+
MessagePassingQueue<InboxItem> inbox,
4745
ConcurrentMap<MetricKey, Batch> pending,
4846
final Set<MetricKey> commonKeys,
4947
int maxAggregates,
@@ -63,8 +61,8 @@ final class Aggregator implements Runnable {
6361

6462
Aggregator(
6563
MetricWriter writer,
66-
Queue<Batch> batchPool,
67-
MpscCompoundQueue<InboxItem> inbox,
64+
MessagePassingQueue<Batch> batchPool,
65+
MessagePassingQueue<InboxItem> inbox,
6866
ConcurrentMap<MetricKey, Batch> pending,
6967
final Set<MetricKey> commonKeys,
7068
int maxAggregates,

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

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import static java.util.Collections.unmodifiableSet;
2020
import static java.util.concurrent.TimeUnit.SECONDS;
2121

22+
import datadog.common.queue.Queues;
2223
import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
2324
import datadog.communication.ddagent.SharedCommunicationObjects;
2425
import datadog.trace.api.Config;
@@ -39,16 +40,14 @@
3940
import java.util.HashSet;
4041
import java.util.List;
4142
import java.util.Map;
42-
import java.util.Queue;
4343
import java.util.Set;
4444
import java.util.concurrent.CompletableFuture;
4545
import java.util.concurrent.ConcurrentHashMap;
4646
import java.util.concurrent.Future;
4747
import java.util.concurrent.TimeUnit;
4848
import java.util.function.Function;
4949
import javax.annotation.Nonnull;
50-
import org.jctools.queues.MpscCompoundQueue;
51-
import org.jctools.queues.SpmcArrayQueue;
50+
import org.jctools.queues.MessagePassingQueue;
5251
import org.slf4j.Logger;
5352
import org.slf4j.LoggerFactory;
5453

@@ -90,11 +89,11 @@ public final class ConflatingMetricsAggregator implements MetricsAggregator, Eve
9089
new HashSet<>(Arrays.asList(SPAN_KIND_CLIENT, SPAN_KIND_PRODUCER, SPAN_KIND_CONSUMER)));
9190

9291
private final Set<String> ignoredResources;
93-
private final Queue<Batch> batchPool;
92+
private final MessagePassingQueue<Batch> batchPool;
9493
private final ConcurrentHashMap<MetricKey, Batch> pending;
9594
private final ConcurrentHashMap<MetricKey, MetricKey> keys;
9695
private final Thread thread;
97-
private final MpscCompoundQueue<InboxItem> inbox;
96+
private final MessagePassingQueue<InboxItem> inbox;
9897
private final Sink sink;
9998
private final Aggregator aggregator;
10099
private final long reportingInterval;
@@ -177,8 +176,8 @@ public ConflatingMetricsAggregator(
177176
long reportingInterval,
178177
TimeUnit timeUnit) {
179178
this.ignoredResources = ignoredResources;
180-
this.inbox = new MpscCompoundQueue<>(queueSize);
181-
this.batchPool = new SpmcArrayQueue<>(maxAggregates);
179+
this.inbox = Queues.mpscArrayQueue(queueSize);
180+
this.batchPool = Queues.spmcArrayQueue(maxAggregates);
182181
this.pending = new ConcurrentHashMap<>(maxAggregates * 4 / 3);
183182
this.keys = new ConcurrentHashMap<>();
184183
this.features = features;

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

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import static datadog.trace.common.metrics.EventListener.EventType.OK;
1010
import static java.util.concurrent.TimeUnit.SECONDS;
1111

12+
import datadog.common.queue.Queues;
1213
import datadog.trace.util.AgentTaskScheduler;
1314
import java.io.IOException;
1415
import java.nio.ByteBuffer;
@@ -23,7 +24,7 @@
2324
import okhttp3.OkHttpClient;
2425
import okhttp3.Request;
2526
import okhttp3.RequestBody;
26-
import org.jctools.queues.SpscArrayQueue;
27+
import org.jctools.queues.MessagePassingQueue;
2728
import org.slf4j.Logger;
2829
import org.slf4j.LoggerFactory;
2930

@@ -36,7 +37,7 @@ public final class OkHttpSink implements Sink, EventListener {
3637
private final OkHttpClient client;
3738
private final HttpUrl metricsUrl;
3839
private final List<EventListener> listeners;
39-
private final SpscArrayQueue<Request> enqueuedRequests = new SpscArrayQueue<>(10);
40+
private final MessagePassingQueue<Request> enqueuedRequests = Queues.spscArrayQueue(16);
4041
private final AtomicLong lastRequestTime = new AtomicLong();
4142
private final AtomicLong asyncRequestCounter = new AtomicLong();
4243
private final boolean bufferingEnabled;
@@ -157,9 +158,9 @@ private void handleFailure(okhttp3.Response response) throws IOException {
157158

158159
private static final class Sender implements AgentTaskScheduler.Task<OkHttpSink> {
159160

160-
private final SpscArrayQueue<Request> inbox;
161+
private final MessagePassingQueue<Request> inbox;
161162

162-
private Sender(SpscArrayQueue<Request> inbox) {
163+
private Sender(MessagePassingQueue<Request> inbox) {
163164
this.inbox = inbox;
164165
}
165166

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@
55
import static datadog.trace.util.AgentThreadFactory.newAgentThread;
66
import static java.util.concurrent.TimeUnit.MILLISECONDS;
77

8+
import datadog.common.queue.MessagePassingBlockingQueue;
9+
import datadog.common.queue.Queues;
810
import datadog.communication.ddagent.DroppingPolicy;
911
import datadog.trace.common.sampling.SingleSpanSampler;
1012
import datadog.trace.core.DDSpan;
@@ -13,7 +15,6 @@
1315
import java.util.List;
1416
import java.util.Queue;
1517
import org.jctools.queues.MessagePassingQueue;
16-
import org.jctools.queues.MpscBlockingConsumerArrayQueue;
1718
import org.slf4j.Logger;
1819
import org.slf4j.LoggerFactory;
1920

@@ -45,7 +46,7 @@ class DefaultSpanSamplingWorker implements SpanSamplingWorker {
4546

4647
private final Thread spanSamplingThread;
4748
private final SamplingHandler samplingHandler;
48-
private final MpscBlockingConsumerArrayQueue<Object> spanSamplingQueue;
49+
private final MessagePassingBlockingQueue<Object> spanSamplingQueue;
4950
private final Queue<Object> primaryQueue;
5051
private final Queue<Object> secondaryQueue;
5152
private final SingleSpanSampler singleSpanSampler;
@@ -62,7 +63,7 @@ protected DefaultSpanSamplingWorker(
6263
DroppingPolicy droppingPolicy) {
6364
this.samplingHandler = new SamplingHandler();
6465
this.spanSamplingThread = newAgentThread(SPAN_SAMPLING_PROCESSOR, samplingHandler);
65-
this.spanSamplingQueue = new MpscBlockingConsumerArrayQueue<>(capacity);
66+
this.spanSamplingQueue = Queues.mpscBlockingConsumerArrayQueue(capacity);
6667
this.primaryQueue = primaryQueue;
6768
this.secondaryQueue = secondaryQueue;
6869
this.singleSpanSampler = singleSpanSampler;

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

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@
55
import static datadog.trace.util.AgentThreadFactory.newAgentThread;
66
import static java.util.concurrent.TimeUnit.MILLISECONDS;
77

8+
import datadog.common.queue.MessagePassingBlockingQueue;
9+
import datadog.common.queue.Queues;
810
import datadog.communication.ddagent.DroppingPolicy;
911
import datadog.trace.api.Config;
1012
import datadog.trace.bootstrap.instrumentation.api.SpanPostProcessor;
@@ -20,7 +22,6 @@
2022
import java.util.concurrent.TimeUnit;
2123
import java.util.function.BooleanSupplier;
2224
import org.jctools.queues.MessagePassingQueue;
23-
import org.jctools.queues.MpscBlockingConsumerArrayQueue;
2425
import org.slf4j.Logger;
2526
import org.slf4j.LoggerFactory;
2627

@@ -36,8 +37,8 @@ public class TraceProcessingWorker implements AutoCloseable {
3637
private static final Logger log = LoggerFactory.getLogger(TraceProcessingWorker.class);
3738

3839
private final PrioritizationStrategy prioritizationStrategy;
39-
private final MpscBlockingConsumerArrayQueue<Object> primaryQueue;
40-
private final MpscBlockingConsumerArrayQueue<Object> secondaryQueue;
40+
private final MessagePassingBlockingQueue<Object> primaryQueue;
41+
private final MessagePassingBlockingQueue<Object> secondaryQueue;
4142
private final TraceSerializingHandler serializingHandler;
4243
private final Thread serializerThread;
4344
private final int capacity;
@@ -121,23 +122,23 @@ public long getRemainingCapacity() {
121122
return primaryQueue.remainingCapacity();
122123
}
123124

124-
private static MpscBlockingConsumerArrayQueue<Object> createQueue(int capacity) {
125-
return new MpscBlockingConsumerArrayQueue<>(capacity);
125+
private static MessagePassingBlockingQueue<Object> createQueue(int capacity) {
126+
return Queues.mpscBlockingConsumerArrayQueue(capacity);
126127
}
127128

128129
public static class TraceSerializingHandler implements Runnable {
129130

130-
private final MpscBlockingConsumerArrayQueue<Object> primaryQueue;
131-
private final MpscBlockingConsumerArrayQueue<Object> secondaryQueue;
131+
private final MessagePassingBlockingQueue<Object> primaryQueue;
132+
private final MessagePassingBlockingQueue<Object> secondaryQueue;
132133
private final HealthMetrics healthMetrics;
133134
private final long ticksRequiredToFlush;
134135
private final boolean doTimeFlush;
135136
private final PayloadDispatcher payloadDispatcher;
136137
private long lastTicks;
137138

138139
public TraceSerializingHandler(
139-
final MpscBlockingConsumerArrayQueue<Object> primaryQueue,
140-
final MpscBlockingConsumerArrayQueue<Object> secondaryQueue,
140+
final MessagePassingBlockingQueue<Object> primaryQueue,
141+
final MessagePassingBlockingQueue<Object> secondaryQueue,
141142
final HealthMetrics healthMetrics,
142143
final PayloadDispatcher payloadDispatcher,
143144
final long flushInterval,

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@
66
import static datadog.trace.util.AgentThreadFactory.newAgentThread;
77
import static java.util.Comparator.comparingLong;
88

9+
import datadog.common.queue.MessagePassingBlockingQueue;
10+
import datadog.common.queue.Queues;
911
import datadog.communication.ddagent.SharedCommunicationObjects;
1012
import datadog.trace.api.Config;
1113
import datadog.trace.api.flare.TracerFlare;
@@ -21,7 +23,6 @@
2123
import java.util.function.Predicate;
2224
import java.util.zip.ZipOutputStream;
2325
import org.jctools.queues.MessagePassingQueue;
24-
import org.jctools.queues.MpscBlockingConsumerArrayQueue;
2526
import org.slf4j.Logger;
2627
import org.slf4j.LoggerFactory;
2728

@@ -61,7 +62,7 @@ private static class DelayingPendingTraceBuffer extends PendingTraceBuffer {
6162
private static final CommandElement DUMP_ELEMENT = new CommandElement();
6263
private static final CommandElement STAND_IN_ELEMENT = new CommandElement();
6364

64-
private final MpscBlockingConsumerArrayQueue<Element> queue;
65+
private final MessagePassingBlockingQueue<Element> queue;
6566
private final Thread worker;
6667
private final TimeSource timeSource;
6768

@@ -292,7 +293,7 @@ public DelayingPendingTraceBuffer(
292293
Config config,
293294
SharedCommunicationObjects sharedCommunicationObjects,
294295
HealthMetrics healthMetrics) {
295-
this.queue = new MpscBlockingConsumerArrayQueue<>(bufferSize);
296+
this.queue = Queues.mpscBlockingConsumerArrayQueue(bufferSize);
296297
this.worker = newAgentThread(TRACE_MONITOR, new Worker());
297298
this.timeSource = timeSource;
298299
boolean runningSpansEnabled = config.isLongRunningTraceEnabled();

0 commit comments

Comments
 (0)