diff --git a/utils/queue-utils/build.gradle.kts b/utils/queue-utils/build.gradle.kts
index 6b72ab9e6a0..2184464b06c 100644
--- a/utils/queue-utils/build.gradle.kts
+++ b/utils/queue-utils/build.gradle.kts
@@ -4,6 +4,7 @@ import org.gradle.jvm.toolchain.JavaLanguageVersion
plugins {
`java-library`
id("dd-trace-java.module.internal-library")
+ id("dd-trace-java.jmh-conventions")
}
dependencies {
diff --git a/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionAlternativesBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionAlternativesBenchmark.java
new file mode 100644
index 00000000000..c8991d767c6
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionAlternativesBenchmark.java
@@ -0,0 +1,286 @@
+package datadog.common.queue;
+
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.jctools.queues.MessagePassingQueue;
+import org.jctools.queues.MpscArrayQueue;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * {@link WorkQueue} against the things a caller would otherwise use, including the two it is
+ * actually replacing. Published because the comparison exists whether or not we run it, and a
+ * reader who has to measure it themselves is entitled to wonder what we found.
+ *
+ *
The decision this table is for. Use {@link MpscArrayQueue} directly when one consumer,
+ * the ring's own bound, and an element you already hold are the whole requirement -- it is the
+ * floor here. Reach for {@link WorkQueue} for its guarantees: an element is built, or a destructive
+ * read made, only once capacity is claimed, through every admission form including batches and
+ * reservations; refusal is reported rather than swallowed; the queue has a lifecycle; and consumer
+ * failures can be retried or handled. The numbers below are what those guarantees cost. They are
+ * not a claim that the API is faster.
+ *
+ *
The baselines are not inventions. {@code arrayBlocking} is {@code WafMetricCollector},
+ * which offers into an {@code ArrayBlockingQueue(1024)}; {@code linkedBlocking} is {@code
+ * RumInjectorMetrics}, which offers into a {@code LinkedBlockingQueue(1024)} and drops the return
+ * value. Both build their element first and find out afterwards whether there was room. {@code
+ * clqWithCounter} is the other thing people write: a {@link ConcurrentLinkedQueue} with an {@link
+ * AtomicInteger} in front of it, guarded by a read and then incremented -- check-then-act, so two
+ * threads at the boundary can both pass, and the bound is a suggestion. It is here because it is
+ * common, not because it is correct; that it is racy is part of what is being compared.
+ *
+ *
Two halves, because the answer differs. The {@code steady} arms admit and drain at one
+ * thread: the per-operation cost with nothing else happening, which is where the alternatives look
+ * their best. The {@code refused} arms sit on a full queue at four threads: the boundary, where a
+ * bounded queue spends its time under load, and where building an element before asking is a wasted
+ * allocation on every call. Read both. A caller whose queue is never full lives in the first table
+ * and should weigh the API for what it buys, not for its speed.
+ *
+ *
Admitting. The raw ring is the floor at about 23ns, with or without {@code fill}.
+ * {@link WorkQueue} costs about 13ns more for the permit counter and the producer callback, and no
+ * more allocation: that is the price of the guarantees above. The incumbents that build first cost
+ * more than that anyway, and the two linked queues allocate a node per element on top of the
+ * element itself.
+ *
+ *
Refusing. Every arm that builds before it asks lands between 140 and 185ns with 24
+ * B/op, and the gap to the claim-first arms is the element, not the queue: the same ring that takes
+ * 142ns to refuse through {@code offer} refuses in 2.6ns through {@code fill}, which also claims a
+ * slot before calling its supplier. {@code refusedWorkQueue} is cheaper still, because refusal is a
+ * load against a counter no refusing thread writes -- see {@code ContendedAdmissionBenchmark}. A
+ * caller whose element already exists, or is free to build, keeps the ordering of this gap and not
+ * its size.
+ */
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Benchmark)
+public class AdmissionAlternativesBenchmark {
+
+ private static final int CAPACITY = 1024;
+
+ private static final String ELEMENT = "element";
+
+ /**
+ * What the incumbents build before they ask. Stands in for a {@code WafMetric} or a metric
+ * sample: not exact, but a real allocation with a stable footprint, so the arm that builds one
+ * can be told from the arm that does not.
+ */
+ static final class Payload {
+ final Object a;
+ final long timestamp;
+
+ Payload(Object a, long timestamp) {
+ this.a = a;
+ this.timestamp = timestamp;
+ }
+ }
+
+ /** Only ever called once a place is already claimed, which is the whole difference. */
+ private static final Producer BUILDER = () -> new Payload(ELEMENT, System.nanoTime());
+
+ /**
+ * The same build for the ring's own {@code fill}, which also claims a slot before calling it. It
+ * must not throw or return null: either would leave a claimed slot the consumer waits on forever.
+ */
+ private static final MessagePassingQueue.Supplier RAW_BUILDER =
+ () -> new Payload(ELEMENT, System.nanoTime());
+
+ private MpscArrayQueue steadyRaw;
+ private MessagePassingQueue steadyRawFill;
+ private BlockingQueue steadyArrayBlocking;
+ private BlockingQueue steadyLinkedBlocking;
+ private Queue steadyClq;
+ private AtomicInteger steadyClqCount;
+ private WorkQueue steadyWork;
+
+ private MpscArrayQueue fullRaw;
+ private MessagePassingQueue fullRawFill;
+ private BlockingQueue fullArrayBlocking;
+ private BlockingQueue fullLinkedBlocking;
+ private Queue fullClq;
+ private AtomicInteger fullClqCount;
+ private WorkQueue fullWork;
+
+ @Setup
+ public void setUp() {
+ steadyRaw = new MpscArrayQueue<>(CAPACITY);
+ steadyRawFill = Queues.mpscArrayQueue(CAPACITY);
+ steadyArrayBlocking = new ArrayBlockingQueue<>(CAPACITY);
+ steadyLinkedBlocking = new LinkedBlockingQueue<>(CAPACITY);
+ steadyClq = new ConcurrentLinkedQueue<>();
+ steadyClqCount = new AtomicInteger();
+ steadyWork = WorkQueues.createMpscQueue(CAPACITY);
+
+ fullRaw = new MpscArrayQueue<>(16);
+ fullRawFill = Queues.mpscArrayQueue(16);
+ fullArrayBlocking = new ArrayBlockingQueue<>(16);
+ fullLinkedBlocking = new LinkedBlockingQueue<>(16);
+ fullClq = new ConcurrentLinkedQueue<>();
+ fullClqCount = new AtomicInteger();
+ fullWork = WorkQueues.createMpscQueue(16);
+ for (int i = 0; i < 16; i++) {
+ Payload payload = new Payload(ELEMENT, i);
+ fullRaw.offer(payload);
+ fullRawFill.offer(payload);
+ fullArrayBlocking.offer(payload);
+ fullLinkedBlocking.offer(payload);
+ fullClq.offer(payload);
+ fullClqCount.incrementAndGet();
+ fullWork.tryPut(payload);
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Steady: admit one, drain one, at a single thread. Every alternative at its best.
+ // ---------------------------------------------------------------------------------------------
+
+ @Benchmark
+ @Threads(1)
+ public void steadyRawMpsc(Blackhole bh) {
+ bh.consume(steadyRaw.offer(new Payload(ELEMENT, System.nanoTime())));
+ bh.consume(steadyRaw.poll());
+ }
+
+ /**
+ * The ring claiming before it builds, through the same factory {@link WorkQueue} uses. Against
+ * {@code steadyWorkQueue}, this is what the permit counter and the API cost on top of it.
+ */
+ @Benchmark
+ @Threads(1)
+ public void steadyRawMpscFill(Blackhole bh) {
+ bh.consume(steadyRawFill.fill(RAW_BUILDER, 1));
+ bh.consume(steadyRawFill.poll());
+ }
+
+ /** {@code WafMetricCollector}: build, then offer, then find out. */
+ @Benchmark
+ @Threads(1)
+ public void steadyArrayBlocking(Blackhole bh) {
+ bh.consume(steadyArrayBlocking.offer(new Payload(ELEMENT, System.nanoTime())));
+ bh.consume(steadyArrayBlocking.poll());
+ }
+
+ /** {@code RumInjectorMetrics}: the same, over a linked queue. */
+ @Benchmark
+ @Threads(1)
+ public void steadyLinkedBlocking(Blackhole bh) {
+ bh.consume(steadyLinkedBlocking.offer(new Payload(ELEMENT, System.nanoTime())));
+ bh.consume(steadyLinkedBlocking.poll());
+ }
+
+ /** The hand-rolled bound: a read, then an increment, with a window between them. */
+ @Benchmark
+ @Threads(1)
+ public void steadyClqWithCounter(Blackhole bh) {
+ if (steadyClqCount.get() < CAPACITY) {
+ steadyClqCount.incrementAndGet();
+ bh.consume(steadyClq.offer(new Payload(ELEMENT, System.nanoTime())));
+ }
+ if (steadyClq.poll() != null) {
+ steadyClqCount.decrementAndGet();
+ }
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void steadyWorkQueue(Blackhole bh) {
+ bh.consume(steadyWork.tryPut(BUILDER));
+ steadyWork.process(bh::consume);
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Refused: a full queue, four threads. The boundary, where a bounded queue lives under load.
+ // ---------------------------------------------------------------------------------------------
+
+ @Benchmark
+ @Threads(4)
+ public void refusedRawMpsc(Blackhole bh) {
+ Payload payload = new Payload(ELEMENT, System.nanoTime());
+ bh.consume(fullRaw.offer(payload));
+ bh.consume(payload);
+ }
+
+ @Benchmark
+ @Threads(4)
+ public void refusedArrayBlocking(Blackhole bh) {
+ Payload payload = new Payload(ELEMENT, System.nanoTime());
+ bh.consume(fullArrayBlocking.offer(payload));
+ bh.consume(payload);
+ }
+
+ @Benchmark
+ @Threads(4)
+ public void refusedLinkedBlocking(Blackhole bh) {
+ Payload payload = new Payload(ELEMENT, System.nanoTime());
+ bh.consume(fullLinkedBlocking.offer(payload));
+ bh.consume(payload);
+ }
+
+ /**
+ * The one arm where the hand-rolled guard is doing what it was written for: refusing without
+ * touching the queue. It still builds the element first, because the caller had no way to know.
+ */
+ @Benchmark
+ @Threads(4)
+ public void refusedClqWithCounter(Blackhole bh) {
+ Payload payload = new Payload(ELEMENT, System.nanoTime());
+ if (fullClqCount.get() < 16) {
+ fullClqCount.incrementAndGet();
+ bh.consume(fullClq.offer(payload));
+ }
+ bh.consume(payload);
+ }
+
+ /** The ring's own claim-before-build: a full ring returns 0 without calling the supplier. */
+ @Benchmark
+ @Threads(4)
+ public void refusedRawMpscFill(Blackhole bh) {
+ bh.consume(fullRawFill.fill(RAW_BUILDER, 1));
+ }
+
+ /** Never asked to build, because the place is claimed first and there was none. */
+ @Benchmark
+ @Threads(4)
+ public void refusedWorkQueue(Blackhole bh) {
+ bh.consume(fullWork.tryPut(BUILDER));
+ }
+}
diff --git a/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionBenchmark.java
new file mode 100644
index 00000000000..2cdabd649b1
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionBenchmark.java
@@ -0,0 +1,231 @@
+package datadog.common.queue;
+
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Compares admission APIs and measures whether HotSpot removes the temporary reservation object
+ * through escape analysis.
+ *
+ *
If the reservation is eliminated, the reserve path is allocation-free and costs what the
+ * producer overloads cost, so {@link BiContextualProducer} would not have been needed for
+ * client-side stats. If it survives, the reserve path allocates per admission and the producer
+ * overloads earn their place. Compare its GC profiler results with those of the producer overloads.
+ *
+ *
{@code gc.alloc.rate.norm} is the number that answers it; the timings are secondary and are
+ * muddied on purpose, because every arm consumes the item it just admitted to keep the queue at
+ * steady state. The element itself is preallocated in every arm, including the producer ones, so
+ * what is being compared is the admission machinery and not the cost of building an element.
+ *
+ *
The {@code backings} parameter is the template-method question. {@code store} and {@code
+ * retrieve} are one call site each, shared by every backing in the process, so their receiver
+ * profile is global: {@code ONE} loads a single concrete subclass and C2 inlines through them,
+ * {@code BOTH} loads two and it still does behind a type guard, {@code THREE} is the cliff. Every
+ * arm drives two extra queues through the same sites for the same number of iterations, so the
+ * traffic is identical and only the number of distinct types in it changes.
+ *
+ *
JDK 25, {@code -Pjmh.forks=2}, ns/op. Every arm allocates 0.01 B/op or less, {@code THREE}
+ * included. Error bars are within ±0.6 except {@code reserveAndFill} at {@code BOTH} (±4.3) and
+ * {@code THREE} (±2.3), and {@code reserveRefused} at {@code ONE} (±1.9).
+ *
+ *
Two things to read off it. A second backing is free — every arm is flat from {@code ONE} to
+ * {@code BOTH}. A third is not free but is small: one to two nanoseconds on a twenty-one nanosecond
+ * operation, because an admit-and-drain pays two uncontended atomics and a ring compare-and-set,
+ * and an out-of-line call is little against memory ordering. {@code reserveRefused} is the control:
+ * it never reaches {@code store}, and it does not move.
+ *
+ *
{@code reserveAndFill} is the exception worth knowing about, at roughly 30%. Its {@code store}
+ * happens inside {@link Reservation#fill} on an object that only exists if escape analysis deletes
+ * it, so that arm is not paying for a virtual call so much as for a longer chain of optimizations
+ * having to survive one. The allocation still goes away — the reservation is still scalar-replaced
+ * at {@code THREE} — but the call it wraps no longer folds into the caller. A caller admitting
+ * through a reservation is the one with something to lose from a third backing.
+ *
+ *
The refusal-design figure that {@link BaseWorkQueue} cites lives here too, and is unchanged by
+ * any of the above: {@code reserveMixed} measured 12 B/op when a refused reservation was a shared
+ * singleton and 0 B/op when it is its own allocation, on JDK 17. Only that arm can tell the two
+ * designs apart — {@code reserveAndFill} and {@code reserveRefused} each see a single outcome, so
+ * C2 prunes the branch that never runs and there is no merge left to defeat escape analysis.
+ *
+ *
{@link MegamorphicCallSiteWorkQueue} is the third type, and lives in this source set rather
+ * than in the module: the question is what a third backing would cost, and shipping one to
+ * find out would answer a different question.
+ */
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Threads(1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Thread)
+public class AdmissionBenchmark {
+
+ public enum Backings {
+ ONE,
+ BOTH,
+ THREE
+ }
+
+ private static final String ELEMENT = "element";
+
+ private static final Producer PRODUCER = () -> ELEMENT;
+
+ private static final ContextualProducer CONTEXTUAL = context -> context;
+
+ private static final BiContextualProducer BI_CONTEXTUAL =
+ (first, second) -> first;
+
+ @Param({"ONE", "BOTH", "THREE"})
+ public Backings backings;
+
+ /** Alternates the reserving queue in {@link #reserveMixed}, so one site sees both outcomes. */
+ private int mixer;
+
+ /** The queue under test. */
+ private WorkQueue queue;
+
+ /** Kept full for the whole run, so its reservations are always refused. */
+ private WorkQueue full;
+
+ /**
+ * Two more queues driven through the same call sites as the queue under test, and the whole of
+ * what the parameter varies. Every arm allocates both and drives both, so the traffic arriving at
+ * {@code store} and {@code retrieve} is identical; only the number of distinct types in it
+ * changes. An arm that skipped the loop would also differ in how hard its call sites had been
+ * exercised before measurement, which is not the question being asked.
+ */
+ private WorkQueue second;
+
+ private WorkQueue third;
+
+ @Setup
+ public void setUp() {
+ queue = WorkQueues.createMpscQueue(1024);
+ full = WorkQueues.createMpscQueue(1);
+ full.tryPut(ELEMENT);
+ second =
+ backings == Backings.ONE
+ ? WorkQueues.createMpscQueue(1024)
+ : WorkQueues.createMpmcQueue(1024);
+ third =
+ backings == Backings.THREE
+ ? new MegamorphicCallSiteWorkQueue<>(1024)
+ : WorkQueues.createMpscQueue(1024);
+ }
+
+ /**
+ * Runs the other backings through the same methods, so every loaded type reaches the shared call
+ * sites. Repeated before every iteration rather than once per trial: the measured loop drives one
+ * type only, and a profile that saw the others just once at startup is not the profile a process
+ * with several live backings actually has.
+ */
+ @Setup(Level.Iteration)
+ public void pollute(Blackhole bh) {
+ for (int i = 0; i < 20_000; i++) {
+ second.tryPut(ELEMENT);
+ second.process(bh::consume);
+ third.tryPut(ELEMENT);
+ third.process(bh::consume);
+ }
+ }
+
+ @Benchmark
+ public void tryPutElement(Blackhole bh) {
+ bh.consume(queue.tryPut(ELEMENT));
+ queue.process(bh::consume);
+ }
+
+ @Benchmark
+ public void tryPutProducer(Blackhole bh) {
+ bh.consume(queue.tryPut(PRODUCER));
+ queue.process(bh::consume);
+ }
+
+ @Benchmark
+ public void tryPutContextual(Blackhole bh) {
+ bh.consume(queue.tryPut(ELEMENT, CONTEXTUAL));
+ queue.process(bh::consume);
+ }
+
+ @Benchmark
+ public void tryPutBiContextual(Blackhole bh) {
+ bh.consume(queue.tryPut(ELEMENT, ELEMENT, BI_CONTEXTUAL));
+ queue.process(bh::consume);
+ }
+
+ @Benchmark
+ public void reserveAndFill(Blackhole bh) {
+ Reservation place = queue.tryReserve();
+ try {
+ if (place.granted()) {
+ place.fill(ELEMENT);
+ }
+ } finally {
+ place.close();
+ }
+ queue.process(bh::consume);
+ }
+
+ /** The refusal path, which is where the shared singleton was supposed to be paying off. */
+ @Benchmark
+ public void reserveRefused(Blackhole bh) {
+ Reservation place = full.tryReserve();
+ try {
+ bh.consume(place.granted());
+ } finally {
+ place.close();
+ }
+ }
+
+ /**
+ * Both outcomes through one call site, which is the only shape where how a refusal is represented
+ * can cost anything.
+ *
+ *
The two arms above each see a single outcome, so C2 prunes the branch that never runs and
+ * there is no merge to defeat escape analysis — they read zero whether a refusal is a shared
+ * singleton or its own allocation, and neither one can tell the two designs apart. A caller whose
+ * queue is nearly always accepting is genuinely in that case. A caller that sits at the boundary,
+ * refusing about as often as it admits, is in this one.
+ */
+ @Benchmark
+ public void reserveMixed(Blackhole bh) {
+ WorkQueue target = (mixer++ & 1) == 0 ? queue : full;
+ Reservation place = target.tryReserve();
+ try {
+ if (place.granted()) {
+ place.fill(ELEMENT);
+ }
+ } finally {
+ place.close();
+ }
+ queue.process(bh::consume);
+ }
+}
diff --git a/utils/queue-utils/src/jmh/java/datadog/common/queue/BackingOverheadBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/BackingOverheadBenchmark.java
new file mode 100644
index 00000000000..ebc36e7d148
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/BackingOverheadBenchmark.java
@@ -0,0 +1,99 @@
+package datadog.common.queue;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import org.jctools.queues.MpmcArrayQueue;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * The two structures the multi-consumer backing could sit on, priced against each other. This is
+ * the measurement {@link MpmcWorkQueue} chose the array ring on.
+ *
+ *
The element is a preallocated singleton in every arm, so the only allocation measured is the
+ * structure's own — which is the point, because the linked queue builds a node per element and the
+ * ring builds nothing.
+ *
+ *
Uncontended the ring is about 38% cheaper and stops manufacturing 24 bytes of garbage per
+ * element, which at a million elements a second is 24MB/s the linked queue creates and the ring
+ * does not. The four-thread arms are ±287 and ±540 respectively and say nothing: each thread offers
+ * and polls the same queue, so both ends of it are thrashed and the measurement is of the harness.
+ * They are kept because leaving them out would imply the contended case was measured and settled.
+ *
+ *
What the ring costs in exchange is that it is not linearizable — see {@link MpmcWorkQueue} for
+ * why claiming a place first makes that survivable, and {@link Queues#mpmcArrayQueue} for why it
+ * would not be otherwise.
+ */
+@Fork(1)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Benchmark)
+public class BackingOverheadBenchmark {
+
+ private static final int CAPACITY = 1024;
+ private static final String ELEMENT = "element";
+
+ private Queue clq1;
+ private Queue mpmc1;
+ private Queue clqN;
+ private Queue mpmcN;
+
+ @Setup
+ public void setUp() {
+ clq1 = new ConcurrentLinkedQueue<>();
+ mpmc1 = new MpmcArrayQueue<>(CAPACITY);
+ clqN = new ConcurrentLinkedQueue<>();
+ mpmcN = new MpmcArrayQueue<>(CAPACITY);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void clqSteady(Blackhole bh) {
+ bh.consume(clq1.offer(ELEMENT));
+ bh.consume(clq1.poll());
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void mpmcSteady(Blackhole bh) {
+ bh.consume(mpmc1.offer(ELEMENT));
+ bh.consume(mpmc1.poll());
+ }
+
+ @Benchmark
+ @Threads(4)
+ public void clqContended(Blackhole bh) {
+ bh.consume(clqN.offer(ELEMENT));
+ bh.consume(clqN.poll());
+ }
+
+ @Benchmark
+ @Threads(4)
+ public void mpmcContended(Blackhole bh) {
+ bh.consume(mpmcN.offer(ELEMENT));
+ bh.consume(mpmcN.poll());
+ }
+}
diff --git a/utils/queue-utils/src/jmh/java/datadog/common/queue/BatchAdmissionBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/BatchAdmissionBenchmark.java
new file mode 100644
index 00000000000..03d8d796933
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/BatchAdmissionBenchmark.java
@@ -0,0 +1,177 @@
+package datadog.common.queue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * What one claim for a run of elements saves over one claim per element, with nothing else going
+ * on. The contended half of the question -- what the same batch costs the producers next to it --
+ * is {@link ContendedBatchAdmissionBenchmark}; this one is the ceiling on the saving.
+ *
+ *
Each arm admits {@code batchSize} elements and drains the same number, so the queue sits well
+ * off its bound and every admission succeeds. Timings are per call, which is per batch: divide by
+ * {@code batchSize} for the per-element figure the pair is actually comparing. The elements are
+ * built once in setup, so what is measured is admission and not construction.
+ *
+ *
What the saving can be. Admission costs a claim on the shared counter and a store into
+ * the backing. Batching removes all but one of the claims and none of the stores, and the MPSC
+ * ring's store is itself a compare-and-set on the producer index. So the ceiling here is a little
+ * under half the per-element cost on that backing, and rather more on the linked one, where the
+ * store is a node allocation and a CAS on the tail but the counter is the only bound there is.
+ *
+ *
Why the batch arm cannot simply win. It walks an iterator and refills a claim every
+ * {@code MAX_BATCH_CLAIM} elements, which the loop arm does not. The sizes run from 4 to 128 to
+ * find where that bookkeeping stops costing more than the claims it removes -- and the answer,
+ * below, is that at one thread it never does.
+ *
+ *
Results, and they go the other way. One fork, five iterations, JDK 25, on a machine
+ * with other work on it -- but the intervals here are tight (a percent or two), because there is no
+ * contention to amplify anything. The per-element columns are the call divided by {@code
+ * batchSize}.
+ *
+ *
Every arm allocates nothing measurable now that the multi-consumer backing is array-backed
+ * rather than linked. The MPMC rows above were within a nanosecond per element of these on the
+ * linked queue, which manufactured a node per element to get there -- 24 bytes an element, priced
+ * in {@link BackingOverheadBenchmark}. Time at parity, allocation gone, is the whole of what that
+ * swap bought; the 8ns per operation the structures differ by in isolation does not survive the
+ * admission machinery being laid on top of it.
+ *
+ *
The producer arms track the element arms to within a nanosecond per element at every size, on
+ * both backings, so they are left out of the table above; the claim is what they share and the
+ * producer call is inlined away.
+ *
+ *
Batching is slower here, by about 2ns per element on MPSC and not at all on MPMC. That
+ * is the honest ceiling on the saving, and it is negative: an uncontended atomic add is a few
+ * cycles, and removing 31 of them out of 32 does not pay for walking an iterator and refilling a
+ * claim. Some of that gap is the iterator itself rather than the claim bookkeeping -- the loop arm
+ * indexes an {@code ArrayList} and the batch arm cannot, because its argument is a {@code
+ * Collection}. It does not much matter which half it is: the size is the same either way and the
+ * sign does not change.
+ *
+ *
Nor does the sign change with batch size, which is the useful part. The saving does not turn
+ * positive at 128, so there is no crossover to find and no minimum batch size to recommend on this
+ * evidence. Whatever batching is worth, it is not worth anything to a producer that is alone.
+ *
+ *
Which is the point. The claim's cost is not in the instruction, it is in the line
+ * everyone else is reading -- and a benchmark with one thread has taken that out. Read this table
+ * as the floor and {@link ContendedBatchAdmissionBenchmark} as the case: there the same batching
+ * producer is between 1.3x and 3.3x faster than the same loop. A caller choosing between them
+ * should be asking how many threads share the queue, not how long its batches are.
+ */
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Threads(1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Thread)
+public class BatchAdmissionBenchmark {
+
+ public enum Backings {
+ MPSC,
+ MPMC
+ }
+
+ /** Comfortably above the largest batch, so no arm is admitting at the bound. */
+ private static final int CAPACITY = 1024;
+
+ private static final String ELEMENT = "element";
+
+ /**
+ * Turns a source element into the queue's element without allocating, as a real one would not.
+ */
+ private static final BiContextualProducer PRODUCER =
+ (source, context) -> source;
+
+ @Param({"MPSC", "MPMC"})
+ public Backings backings;
+
+ @Param({"4", "8", "32", "128"})
+ public int batchSize;
+
+ private WorkQueue queue;
+
+ /** Built once: the arms compare admission, not the cost of producing a list. */
+ private List elements;
+
+ @Setup
+ public void setUp() {
+ queue =
+ backings == Backings.MPSC
+ ? WorkQueues.createMpscQueue(CAPACITY)
+ : WorkQueues.createMpmcQueue(CAPACITY);
+ elements = new ArrayList<>(batchSize);
+ for (int i = 0; i < batchSize; i++) {
+ elements.add(ELEMENT);
+ }
+ }
+
+ /** One claim per run of elements. */
+ @Benchmark
+ public void batchOfElements(Blackhole bh) {
+ bh.consume(queue.tryPutBatch(elements));
+ bh.consume(queue.process(batchSize, bh::consume));
+ }
+
+ /** The same admissions, one claim each: what a caller writes without the batch API. */
+ @Benchmark
+ public void loopOfElements(Blackhole bh) {
+ for (int i = 0; i < batchSize; i++) {
+ bh.consume(queue.tryPut(elements.get(i)));
+ }
+ bh.consume(queue.process(batchSize, bh::consume));
+ }
+
+ /** The producer form, where the claim also gates whether an element is built at all. */
+ @Benchmark
+ public void batchOfProducers(Blackhole bh) {
+ bh.consume(queue.tryPutBatch(elements, ELEMENT, PRODUCER));
+ bh.consume(queue.process(batchSize, bh::consume));
+ }
+
+ @Benchmark
+ public void loopOfProducers(Blackhole bh) {
+ for (int i = 0; i < batchSize; i++) {
+ bh.consume(queue.tryPut(elements.get(i), ELEMENT, PRODUCER));
+ }
+ bh.consume(queue.process(batchSize, bh::consume));
+ }
+}
diff --git a/utils/queue-utils/src/jmh/java/datadog/common/queue/ContendedAdmissionBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/ContendedAdmissionBenchmark.java
new file mode 100644
index 00000000000..6b5ca2fc1f1
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/ContendedAdmissionBenchmark.java
@@ -0,0 +1,283 @@
+package datadog.common.queue;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.concurrent.TimeUnit;
+import org.jctools.queues.MpscArrayQueue;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Admission with more than one thread arriving at once, which is the only way two of this module's
+ * central costs become visible at all.
+ *
+ *
Allocation, as throughput. A single thread allocates almost for free — a pointer bump
+ * in a thread-local buffer — so a per-operation allocation shows up in {@code B/op} and barely at
+ * all in {@code ns/op}, which is how an allocation on a hot path gets waved through. Several
+ * threads allocating at once pay for it together: buffer refills, the memory bandwidth to touch
+ * fresh cache lines, and eventually collection. That turns the allocation into a throughput number,
+ * in the units a reviewer actually argues about. {@link AdmissionBenchmark} measured the
+ * reservation path at 0 B/op against 12 for a shared refusal singleton, and it measured that at one
+ * thread — where it is nearly free. This is where 12 B/op gets priced.
+ *
+ *
{@code refusedProducer} against {@code refusedBuildThenOffer} is this module's whole premise
+ * stated as a benchmark. Both sit on a full queue and admit nothing. The first hands over a
+ * producer and is never asked to build, because a place is claimed first and there was none; the
+ * second builds its element and then discovers there is no room, which is the shape {@link
+ * WorkQueue} exists to replace. At one thread the difference is an allocation that escape analysis
+ * may well erase anyway. Under load it is the difference the API is claiming.
+ *
+ *
Contention, as itself. {@link BaseWorkQueue#claimPlace} spends a place with one atomic
+ * decrement and gives it back with a second when there was none to spend, so a refused admission
+ * pays two read-modify-writes on one shared line — at the capacity boundary, where the most threads
+ * are arriving at once. {@code refusedRaw} is the baseline that prices it: on the MPSC backing
+ * jctools already enforces capacity through its own producer-index CAS, so a caller that never
+ * reserves is paying the counter for a bound it was getting free, and the delta is what that costs.
+ * That delta is what the relaxed read in {@code claimPlace} brought down from ~960ns to ~5ns. The
+ * linked backing has no such baseline — {@code ConcurrentLinkedQueue} is unbounded and the counter
+ * is the only thing bounding it, so there the comparison is against having no bound at all.
+ *
+ *
{@code steady} is the other half, and the commoner one: not full, with a consumer making room
+ * as fast as producers take it. It is the only arm where the counter is incremented by a drain
+ * while it is decremented by admission, contending on the same line from both directions. Its
+ * consumer is a thread this class owns rather than a {@code @Group} member; the arm's own note says
+ * why that distinction is load-bearing.
+ *
+ *
Neither cost is visible in {@link AdmissionBenchmark}, which is {@code @Threads(1)} and {@code
+ * Scope.Thread} — every thread there gets its own queue, so there is nothing to contend on and
+ * nothing to allocate alongside.
+ *
+ *
Results. Eight threads, one fork, {@code -Pjmh.profilers=gc}, JDK 25, on a machine with other
+ * work on it -- so the absolute numbers run high and the intervals are wide. They are an
+ * impression, not a baseline; {@code refusedRaw} is the control on every run.
+ *
+ *
The {@code before} column is the decrement-then-back-out admission this benchmark was written
+ * to price. The {@code after} column is the same run once {@link BaseWorkQueue#claimPlace} took to
+ * reading the count before spending from it.
+ *
+ *
The {@code before} column predates two changes, not one: the MPMC rows were taken when that
+ * backing was a {@link java.util.concurrent.ConcurrentLinkedQueue} rather than an array ring. They
+ * are kept because the column exists to show the 120x, which is a property of the counter and not
+ * of the structure underneath it -- the refusal never reaches the backing at all. The {@code after}
+ * column is a single fresh run of everything, so the arms are comparable with each other.
+ *
+ *
What this measured. Two read-modify-writes on one shared line, taken by eight threads
+ * at the capacity boundary, cost about 120x what the same rejection costs when the first of them is
+ * a load instead. A refusal is now ~7.3ns against ~2.5ns for jctools' own producer-index CAS, so
+ * the permit counter costs on the order of 5ns over a bound the ring was already enforcing --
+ * against ~960ns before, where it dwarfed everything else the API does. {@code steady} moved with
+ * it, and for the same reason: eight producers against one drain thread keep the queue saturated,
+ * so most of that arm is refusals too.
+ *
+ *
Treat the ratio with more suspicion than the direction. Two contended read-modify-writes
+ * should not cost 960ns on a quiet machine -- tens of nanoseconds is the expected order -- so some
+ * of that baseline is this machine's other work amplifying the contention, threads losing their
+ * slice mid-sequence with the line hot. The ~7.9ns is tight and the mechanism is not in doubt; a
+ * quiet run will likely show a smaller multiple against a smaller before.
+ *
+ *
{@code steady} measured 252.8ns per attempt on MPMC and 489.7ns on MPSC in this run. The timed
+ * operation is an admission attempt, including refusals; a separate, untimed thread drains the
+ * queue. These values do not isolate drain cost or successful handoff throughput. Both backings
+ * support this producer/consumer arrangement.
+ *
+ *
Attribution, since two changes landed together: this is the read, not the folding of the
+ * closed flag into the count. Removing a volatile boolean load cannot account for 950ns. Folding it
+ * was structural -- one word of state instead of two that have to agree.
+ *
+ *
The premise pair, which now goes the way the module argues. {@code refusedProducer}
+ * against {@code refusedBuildThenOffer} is reserve-before-build against building first and finding
+ * out after: ~8ns and 0 B/op against ~422ns and 32 B/op. Before the read it was the awkward result
+ * -- 0 B/op but slower in {@code ns/op} -- because the counter cost more than the allocation it
+ * avoided. It no longer does. Read the pair for what it is even so: the build-then-offer arm has no
+ * counter and the producer arm has no allocation, so it is not one variable. What it establishes is
+ * the ordering, and the ordering has reversed.
+ */
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Threads(4)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Benchmark)
+public class ContendedAdmissionBenchmark {
+
+ public enum Backings {
+ MPSC,
+ MPMC
+ }
+
+ /** Big enough that the steady arm is not living at the boundary by accident. */
+ private static final int CAPACITY = 1024;
+
+ private static final String ELEMENT = "element";
+
+ /**
+ * Stands in for the element a real producer builds — a client-stats {@code SpanSnapshot} and its
+ * tag arrays. Its size is not meant to be exact; what matters is that it is a real allocation
+ * with a stable footprint, so the arm that builds one can be compared against the arm that does
+ * not.
+ */
+ static final class Payload {
+ final Object a;
+ final Object b;
+ final long duration;
+ final int hash;
+
+ Payload(Object a, Object b, long duration) {
+ this.a = a;
+ this.b = b;
+ this.duration = duration;
+ this.hash = a.hashCode() ^ b.hashCode();
+ }
+ }
+
+ /** Allocates on every call, and is only ever called once a place is already claimed. */
+ private static final Producer BUILDER =
+ () -> new Payload(ELEMENT, ELEMENT, System.nanoTime());
+
+ @Param({"MPSC", "MPMC"})
+ public Backings backings;
+
+ /** Queue drained concurrently by the dedicated {@link #drain} thread. */
+ private WorkQueue queue;
+
+ /** Filled once in setup and never drained, so every admission is refused. */
+ private WorkQueue full;
+
+ /** The same, typed for the producer arm. */
+ private WorkQueue fullPayloads;
+
+ /**
+ * A full queue with no permit counter, for the two baselines. Unaffected by {@code backings} — it
+ * is the same number in both rows, and is what the MPSC backing would cost on the ring's own
+ * bound alone.
+ */
+ private MpscArrayQueue