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/AdmissionBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionBenchmark.java
new file mode 100644
index 00000000000..571ae4be1f9
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/AdmissionBenchmark.java
@@ -0,0 +1,185 @@
+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.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;
+
+/**
+ * The three admission forms measured against each other on a real queue, to answer one question:
+ * does the reservation object survive escape analysis? If it does, the reserve route costs what the
+ * producer routes cost and {@link BiContextualProducer} was never needed for client-side stats. If
+ * it does not, the producer callbacks are earning their contortions.
+ *
+ *
{@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 ONE} loads a single
+ * concrete subclass, so {@code store} is monomorphic and C2 inlines it outright; {@code BOTH} loads
+ * two, which C2 still inlines behind a type guard. A third backing would be the cliff. Measuring
+ * both is how we find out whether the inheritance layout costs anything today, or only threatens
+ * to.
+ *
+ *
Results, filled in as they are measured:
+ *
+ *
+ * Benchmark (backings) ns/op B/op
+ * tryPutElement ONE ? ?
+ * tryPutElement BOTH ? ?
+ * tryPutContextual ONE ? ?
+ * tryPutContextual BOTH ? ?
+ * tryPutBiContextual ONE ? ?
+ * tryPutBiContextual BOTH ? ?
+ * reserveAndFill ONE 20.4 0
+ * reserveAndFill BOTH 20.6 0
+ * reserveRefused ONE 13.8 0
+ * reserveRefused BOTH 13.9 0
+ * reserveMixed ONE 13.6 0 (12 with a shared refusal singleton)
+ * reserveMixed BOTH 13.6 0 (12 with a shared refusal singleton)
+ *
+ *
+ *
JDK 17, one machine, {@code -Pjmh.forks=1}. The single-outcome arms cannot distinguish the two
+ * refusal designs; only {@code reserveMixed} can.
+ */
+@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
+ }
+
+ 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"})
+ 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;
+
+ /**
+ * Present only to put a second concrete subclass into the profile. Its call sites are the same
+ * ones the queue under test uses, which is exactly the pollution being measured.
+ */
+ private WorkQueue other;
+
+ @Setup
+ public void setUp(Blackhole bh) {
+ queue = WorkQueues.createMpscQueue(1024);
+ full = WorkQueues.createMpscQueue(1);
+ full.tryPut(ELEMENT);
+ if (backings == Backings.BOTH) {
+ other = WorkQueues.createMpmcQueue(1024);
+ // Warm the other backing through the same methods, so both types reach the call sites.
+ for (int i = 0; i < 20_000; i++) {
+ other.tryPut(ELEMENT);
+ other.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/main/java/datadog/common/queue/BaseWorkQueue.java b/utils/queue-utils/src/main/java/datadog/common/queue/BaseWorkQueue.java
new file mode 100644
index 00000000000..7a3a4915ecd
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/BaseWorkQueue.java
@@ -0,0 +1,519 @@
+package datadog.common.queue;
+
+import static java.util.Collections.emptyList;
+
+import datadog.trace.api.function.Strategy;
+import datadog.trace.api.function.StrategyConsumer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+/**
+ * Everything a {@link WorkQueue} does that does not depend on how elements are stored: the bound,
+ * admission, reservations, the closed flag, drop counting, and the consume-and-maybe-retry cycle.
+ *
+ *
Subclasses supply two storage primitives, {@link #store} and {@link #retrieve}, and neither
+ * needs to enforce anything. The bound lives here, as a count of places still available: admission
+ * spends one before it builds or stores anything, consumption returns one, and a reservation is
+ * simply a spent place with nothing in it yet. That is why the storage primitives can be as thin as
+ * they are, and why both backings admit and reserve through exactly the same code.
+ *
+ *
The counter costs one atomic add per admission and one per consumption. On a backing that
+ * could have leaned on its own bound that is a real tax, paid for a uniform contract: every backing
+ * can reserve capacity, nothing has to hold a position open, so no consumer can be stalled by a
+ * reservation and no reservation can deadlock a thread that also consumes.
+ */
+abstract class BaseWorkQueue implements WorkQueue {
+
+ /**
+ * Wraps an item that has already failed, carrying its attempt count back into the queue. Only
+ * allocated on the failure path, so the common case stores the element itself.
+ */
+ private static final class Retry {
+ final T item;
+ final int attempt;
+
+ Retry(T item, int attempt) {
+ this.item = item;
+ this.attempt = attempt;
+ }
+ }
+
+ /** Non-capturing adapters, so the producer forms share one admission path without allocating. */
+ private static final ContextualProducer, Object> PRODUCE = Producer::produce;
+
+ /**
+ * The answer to every refused claim: a reservation that holds nothing, discards whatever is
+ * filled into it, and has nothing to give back. It holds no state, so one instance serves every
+ * queue and every element type.
+ *
+ *
Filling it is a no-op rather than a throw. The queue is full exactly when a caller can least
+ * afford a surprise, and an exception raised only under backpressure is a bug that waits for
+ * production to appear. The drop is already counted, by {@link #tryReserve} at the moment of
+ * refusal.
+ */
+ private final LongAdder dropped = new LongAdder();
+
+ private volatile boolean closed;
+
+ /**
+ * Places still available, not places used. The bound is then a comparison against zero rather
+ * than against a capacity that has to be loaded and that an unbounded queue has to be branched
+ * around: seeded with {@link Integer#MAX_VALUE} it is a queue no backlog can exhaust, on the same
+ * code path as any other.
+ */
+ private final AtomicInteger available;
+
+ private final int capacity;
+
+ BaseWorkQueue(int capacity) {
+ this.capacity = capacity;
+ this.available = new AtomicInteger(capacity);
+ }
+
+ /**
+ * Stores an element in a place already claimed for it, so this can only fail if the backing
+ * refuses for a reason of its own.
+ *
+ * @return whether the element was stored
+ */
+ /**
+ * The one call site every backing funnels through, which is why the count of backings loaded in a
+ * process is an admission cost and not only a dispatch cost. At one or two implementations this
+ * site is free; a third makes it megamorphic, measured at 24 bytes and roughly three times the
+ * time per call — paid by callers that only ever touch one backing. A third backing is therefore
+ * a decision about every existing caller, and the point at which to replace this template method
+ * with a per-caller strategy so the sites stay separate.
+ */
+ abstract boolean store(Object element);
+
+ /**
+ * @return the next stored object, or {@code null} if there was none
+ */
+ abstract Object retrieve();
+
+ /**
+ * Spends a place, and gives it back if there was none to spend, rather than looping on a
+ * compare-and-set. Admission costs one atomic add, with a second only on the path that was going
+ * to be rejected anyway — and no retry under contention, which is where a CAS loop is at its
+ * worst.
+ *
+ *
The bound itself is exact: the queue never holds more than {@code capacity} elements and
+ * open reservations together. What is approximate is who gets turned away. Claimants racing at
+ * the boundary can drive the count below zero between them and all give their places back, so an
+ * admission can be rejected while the queue is a place or two short of full. That only happens
+ * when it is already at the boundary, where the caller is dropping work regardless.
+ */
+ private boolean claimPlace() {
+ if (available.decrementAndGet() >= 0) {
+ return true;
+ }
+ available.incrementAndGet();
+ return false;
+ }
+
+ private void releasePlace() {
+ available.incrementAndGet();
+ }
+
+ private boolean admit(Object element) {
+ if (!claimPlace()) {
+ return false;
+ }
+ if (store(element)) {
+ return true;
+ }
+ releasePlace();
+ return false;
+ }
+
+ @StrategyConsumer
+ private boolean admit(
+ C context, @Strategy ContextualProducer producer) {
+ if (!claimPlace()) {
+ return false;
+ }
+ T element;
+ try {
+ element = producer.produce(context);
+ } catch (Throwable t) {
+ releasePlace();
+ throw t;
+ }
+ return storeOrRelease(element);
+ }
+
+ @StrategyConsumer
+ private boolean admit(
+ C1 first,
+ C2 second,
+ @Strategy BiContextualProducer producer) {
+ if (!claimPlace()) {
+ return false;
+ }
+ T element;
+ try {
+ element = producer.produce(first, second);
+ } catch (Throwable t) {
+ releasePlace();
+ throw t;
+ }
+ return storeOrRelease(element);
+ }
+
+ private boolean storeOrRelease(T element) {
+ if (element != null && store(element)) {
+ return true;
+ }
+ releasePlace();
+ return false;
+ }
+
+ /**
+ * A place spent ahead of the element that will use it. Filling can only ever store, because the
+ * room was already taken; abandoning gives the room back. Nothing is held open in the backing, so
+ * a consumer never has to wait on one.
+ *
+ *
Both outcomes come from one allocation site. A refusal could be a shared singleton, and that
+ * is the more obvious design: it saves the allocation on the path that already lost. What it
+ * costs is paid by a caller that sees both outcomes at one site. Returning either a fresh
+ * reservation or a static merges an allocation with a globally reachable reference at a phi, and
+ * escape analysis gives up on the merge, so a reservation that would have been scalar-replaced
+ * away is allocated for real — {@code AdmissionBenchmark.reserveMixed} measures 12 bytes per call
+ * that way and zero this way, on JDK 17. JDK 21's allocation-merge support does not rescue it:
+ * that covers merges of non-escaping allocations and null, never a static.
+ *
+ *
The condition matters, because it is not every caller. A site that only ever sees one
+ * outcome — a queue that is effectively always accepting, or the drain loop's always-full
+ * counterpart — has its other branch pruned, and there is no merge left to defeat anything; both
+ * designs measure zero there. So this is insurance for the caller sitting at the capacity
+ * boundary rather than a saving for everyone. It is free insurance, which is the reason to take
+ * it: one allocation site is no worse anywhere, and it also keeps {@link #fill} and {@link
+ * #close} monomorphic for callers that never see a refusal, and drops an unchecked cast.
+ *
+ *
A refused reservation starts out {@code done}, which is what makes it inert: there is no
+ * place to give back and nothing to store, and both methods already short-circuit on that flag.
+ *
+ *
Static, with the queue handed in, rather than an inner class holding it implicitly. The
+ * reference is a field of this object either way, so nothing changes at runtime; what changes is
+ * that a reader can see it. That matters here more than it usually would, because the shape above
+ * is asking escape analysis to delete this object and promote its fields to locals — so the field
+ * count is the subject, and a hidden field is a hidden part of the subject.
+ */
+ private static final class PlaceReservation implements Reservation {
+ private final BaseWorkQueue queue;
+ private final boolean granted;
+ private boolean done;
+
+ PlaceReservation(BaseWorkQueue queue, boolean granted) {
+ this.queue = queue;
+ this.granted = granted;
+ this.done = !granted;
+ }
+
+ @Override
+ public boolean granted() {
+ return granted;
+ }
+
+ @Override
+ public void fill(T element) {
+ // Before the null check, so that filling a refusal stays silent: a caller that skipped
+ // building an element has nothing but null to offer, and the refused path never throws.
+ if (done) {
+ return;
+ }
+ if (element == null) {
+ throw new NullPointerException("a queue cannot hold null");
+ }
+ done = true;
+ queue.store(element);
+ }
+
+ @Override
+ public void close() {
+ // Only the reserving thread fills or closes, so a plain flag orders the two correctly.
+ if (!done) {
+ done = true;
+ queue.releasePlace();
+ }
+ }
+ }
+
+ private Object take() {
+ Object element = retrieve();
+ if (element != null) {
+ releasePlace();
+ }
+ return element;
+ }
+
+ private void discardAll() {
+ while (take() != null) {
+ // give every place back as it goes
+ }
+ }
+
+ @Override
+ public final int size() {
+ // Claimants at the boundary can transiently drive the count below zero before backing out.
+ return Math.max(0, capacity - available.get());
+ }
+
+ @Override
+ public final boolean tryPut(T element) {
+ return record(!closed && admit(element));
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public final boolean tryPut(Producer producer) {
+ return record(!closed && admit(producer, (ContextualProducer) PRODUCE));
+ }
+
+ @Override
+ public final boolean tryPut(C context, ContextualProducer producer) {
+ return record(!closed && admit(context, producer));
+ }
+
+ @Override
+ public final boolean tryPut(
+ C1 first, C2 second, BiContextualProducer producer) {
+ return record(!closed && admit(first, second, producer));
+ }
+
+ @Override
+ @SafeVarargs
+ public final Collection tryPutBatch(T... elements) {
+ List rejected = null;
+ for (int i = 0; i < elements.length; i++) {
+ T element = elements[i];
+ if (!tryPut(element)) {
+ if (rejected == null) {
+ // Refusals run to the end far more often than not: once the queue is full it stays full
+ // for the rest of the pass unless a consumer intervenes. Sizing for the remainder is an
+ // exact fit in that case and an over-fit in the other, and either beats regrowing.
+ rejected = new ArrayList<>(elements.length - i);
+ }
+ rejected.add(element);
+ }
+ }
+ return rejected == null ? emptyList() : rejected;
+ }
+
+ @Override
+ public final Collection tryPutBatch(Collection elements) {
+ List rejected = null;
+ int remaining = elements.size();
+ for (T element : elements) {
+ if (!tryPut(element)) {
+ if (rejected == null) {
+ rejected = new ArrayList<>(remaining);
+ }
+ rejected.add(element);
+ }
+ remaining--;
+ }
+ return rejected == null ? emptyList() : rejected;
+ }
+
+ @Override
+ public final Reservation tryReserve() {
+ boolean granted = !closed && claimPlace();
+ if (!granted) {
+ dropped.increment();
+ }
+ return new PlaceReservation<>(this, granted);
+ }
+
+ @Override
+ public final boolean process(Consumer consumer) {
+ return processOrRetry(consumer, null);
+ }
+
+ @Override
+ public final boolean processOrHandle(
+ Consumer consumer, ExceptionHandler exceptionHandler) {
+ Object raw = take();
+ if (raw == null) {
+ return false;
+ }
+ consume(raw, consumer, null, null, null, exceptionHandler);
+ return true;
+ }
+
+ @Override
+ public final boolean processOrRetry(
+ Consumer consumer, RetryStrategy retryStrategy) {
+ Object raw = take();
+ if (raw == null) {
+ return false;
+ }
+ consume(raw, consumer, null, null, retryStrategy, null);
+ return true;
+ }
+
+ @Override
+ public final boolean process(C context, BiConsumer consumer) {
+ return processOrRetry(context, consumer, null);
+ }
+
+ @Override
+ public final boolean processOrRetry(
+ C context, BiConsumer consumer, RetryStrategy retryStrategy) {
+ Object raw = take();
+ if (raw == null) {
+ return false;
+ }
+ consume(raw, null, context, consumer, retryStrategy, null);
+ return true;
+ }
+
+ @Override
+ public final boolean processOrHandle(
+ C context,
+ BiConsumer consumer,
+ ExceptionHandler exceptionHandler) {
+ Object raw = take();
+ if (raw == null) {
+ return false;
+ }
+ consume(raw, null, context, consumer, null, exceptionHandler);
+ return true;
+ }
+
+ @Override
+ public final int process(int limit, Consumer consumer) {
+ return process(limit, consumer, null, null);
+ }
+
+ @Override
+ public final int process(int limit, C context, BiConsumer consumer) {
+ return process(limit, null, context, consumer);
+ }
+
+ private int process(
+ int limit,
+ Consumer consumer,
+ C context,
+ BiConsumer biConsumer) {
+ int consumed = 0;
+ while (consumed < limit) {
+ Object raw = take();
+ if (raw == null) {
+ break;
+ }
+ // Counted before the consumer runs: a throw carries the count away with it either way, and
+ // an item handed over is consumed whether or not the consumer made anything of it.
+ consumed++;
+ consume(raw, consumer, context, biConsumer, null, null);
+ }
+ return consumed;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void consume(
+ Object raw,
+ Consumer consumer,
+ C context,
+ BiConsumer biConsumer,
+ RetryStrategy retryStrategy,
+ ExceptionHandler exceptionHandler) {
+ T item;
+ int attempt;
+ if (raw instanceof Retry) {
+ Retry retried = (Retry) raw;
+ item = retried.item;
+ attempt = retried.attempt;
+ } else {
+ item = (T) raw;
+ attempt = 0;
+ }
+ if (retryStrategy == null && exceptionHandler == null) {
+ // No strategy means no opinion about failure: the throw travels out to the caller's own
+ // frame, where its existing error handling already lives. Swallowing it here would make a
+ // queue the arbiter of an error policy nobody handed it.
+ if (consumer != null) {
+ consumer.accept(item);
+ } else {
+ biConsumer.accept(context, item);
+ }
+ return;
+ }
+ try {
+ if (consumer != null) {
+ consumer.accept(item);
+ } else {
+ biConsumer.accept(context, item);
+ }
+ } catch (Throwable failure) {
+ if (exceptionHandler != null) {
+ dropped.increment();
+ exceptionHandler.handle(item, failure);
+ } else if (!retryStrategy.onFailure(item, attempt + 1, failure, lease(attempt + 1))) {
+ dropped.increment();
+ }
+ }
+ }
+
+ /** Allocated only once a consumer has thrown, and never escapes {@link #onFailure}. */
+ private RetryQueue lease(int attempt) {
+ return new RetryQueue() {
+ @Override
+ public boolean retry(T item) {
+ if (closed || !admit(new Retry<>(item, attempt))) {
+ dropped.increment();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean retry(T... items) {
+ boolean all = items.length > 0;
+ for (T item : items) {
+ all &= retry(item);
+ }
+ return all;
+ }
+ };
+ }
+
+ private boolean record(boolean admitted) {
+ if (!admitted) {
+ dropped.increment();
+ }
+ return admitted;
+ }
+
+ @Override
+ public final long dropped() {
+ return dropped.sum();
+ }
+
+ @Override
+ public final void close() {
+ closed = true;
+ }
+
+ @Override
+ public final boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public final void clear() {
+ discardAll();
+ }
+
+ @Override
+ public final void shutdown() {
+ closed = true;
+ discardAll();
+ }
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/BiContextualProducer.java b/utils/queue-utils/src/main/java/datadog/common/queue/BiContextualProducer.java
new file mode 100644
index 00000000000..cfc8fc72cbb
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/BiContextualProducer.java
@@ -0,0 +1,22 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * A {@link Producer} that derives its element from two caller-supplied contexts.
+ *
+ *
Two rather than one because the second context is typically a value the call site hoisted out
+ * of a loop — a schema, a clock reading, a per-batch buffer — that the item alone cannot recover.
+ * Carrying it as a parameter is what keeps the producer a non-capturing bound-once field and keeps
+ * the hoist visible where it happens, instead of a per-iteration capture or a cached binding that
+ * can silently go stale.
+ *
+ *
The ladder stops here on purpose. A third context is usually derivable from the item, and a
+ * primitive one has to be boxed to ride a generic parameter, which costs more than re-deriving it.
+ * A call site that genuinely needs more should close over what it needs once per scope.
+ */
+@Strategy
+@FunctionalInterface
+public interface BiContextualProducer {
+ T produce(C1 first, C2 second);
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/ContextualProducer.java b/utils/queue-utils/src/main/java/datadog/common/queue/ContextualProducer.java
new file mode 100644
index 00000000000..9e556b84697
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/ContextualProducer.java
@@ -0,0 +1,15 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * A {@link Producer} that derives its element from a caller-supplied context.
+ *
+ *
The context parameter is what lets the producer stay non-capturing: state the element needs is
+ * passed in at the call site rather than closed over.
+ */
+@Strategy
+@FunctionalInterface
+public interface ContextualProducer {
+ T produce(C context);
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/ExceptionHandler.java b/utils/queue-utils/src/main/java/datadog/common/queue/ExceptionHandler.java
new file mode 100644
index 00000000000..7d2d996bd0a
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/ExceptionHandler.java
@@ -0,0 +1,24 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Deals with a consumer's failure and lets the item go, for a caller who wants to see what went
+ * wrong without deciding whether to try again. The item is dropped either way, and the failure does
+ * not reach the caller of {@code processOrHandle}.
+ *
+ *
The narrow half of {@link RetryStrategy}: reach for that one when the answer to a failure is
+ * sometimes "again", and this one when it is only ever "record it and move on". The item comes
+ * along because the consumer that threw is in no position to say which one died.
+ *
+ * @see WorkQueue#processOrHandle(java.util.function.Consumer, ExceptionHandler)
+ */
+@Strategy
+@FunctionalInterface
+public interface ExceptionHandler {
+ /**
+ * Called on the consuming thread, in place of propagating. A handler that throws propagates in
+ * the failure's stead.
+ */
+ void handle(T item, Throwable failure);
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/LinkedWorkQueue.java b/utils/queue-utils/src/main/java/datadog/common/queue/LinkedWorkQueue.java
new file mode 100644
index 00000000000..0cd4cec2808
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/LinkedWorkQueue.java
@@ -0,0 +1,39 @@
+package datadog.common.queue;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * A {@link WorkQueue} over a {@link ConcurrentLinkedQueue}: multi-producer, multi-consumer,
+ * optionally bounded.
+ *
+ *
This backing exists to give call sites that cannot yet take an MPSC ring — because they have
+ * several consumers, or no defensible capacity — the admission and lifecycle contract anyway, so
+ * they can be migrated behind {@link WorkQueue} first and re-backed later. It keeps the linked
+ * queue's per-element node, so it does not deliver the allocation win; prefer {@link
+ * MpscWorkQueue}.
+ *
+ *
Storage only: the bound lives in {@link BaseWorkQueue}, which is what replaces the hand-rolled
+ * cap plus O(n) {@code ConcurrentLinkedQueue.size()} walk such a call site otherwise pays on every
+ * admission, and makes {@link #size()} constant-time.
+ */
+final class LinkedWorkQueue extends BaseWorkQueue {
+
+ private final ConcurrentLinkedQueue