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/ContendedAdmissionBenchmark.java b/utils/queue-utils/src/jmh/java/datadog/common/queue/ContendedAdmissionBenchmark.java
new file mode 100644
index 00000000000..4eca521f412
--- /dev/null
+++ b/utils/queue-utils/src/jmh/java/datadog/common/queue/ContendedAdmissionBenchmark.java
@@ -0,0 +1,272 @@
+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.
+ *
+ *
+ * ./gradlew :utils:queue-utils:jmh -Pjmh.includes=ContendedAdmission -Pjmh.threads=8 -Pjmh.profilers=gc
+ *
+ *
+ * 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.
+ *
+ *
+ * Benchmark (backings) before ns/op after ns/op B/op
+ * refusedProducer MPSC 1035.8 8.2 0
+ * refusedProducer LINKED 1162.5 9.0 0
+ * refusedQueue MPSC 963.7 7.9 0
+ * refusedQueue LINKED 1229.0 7.9 0
+ * refusedBuildThenOffer MPSC 448.9 422.0 32
+ * refusedBuildThenOffer LINKED 460.0 422.0 32
+ * refusedRaw MPSC 3.4 2.9 0
+ * refusedRaw LINKED 3.4 3.0 0
+ * steady MPSC 798.7 125.4 0
+ * steady LINKED 1008.4 146.8 1.4
+ *
+ *
+ * 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.9ns against ~2.9ns 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.
+ *
+ *
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,
+ LINKED
+ }
+
+ /** 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", "LINKED"})
+ public Backings backings;
+
+ /** Never full, drained concurrently by {@link #consume}. */
+ 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 raw;
+
+ /** The one consumer for {@link #steady}. Owned here, not by JMH -- see that arm's note. */
+ private Thread drain;
+
+ private volatile boolean draining;
+
+ @Setup
+ public void setUp() {
+ queue = create(CAPACITY);
+ full = create(16);
+ while (full.tryPut(ELEMENT)) {
+ // fill it, so claimPlace always has to back out
+ }
+ fullPayloads =
+ backings == Backings.MPSC ? WorkQueues.createMpscQueue(16) : WorkQueues.createMpmcQueue(16);
+ while (fullPayloads.tryPut(new Payload(ELEMENT, ELEMENT, 0L))) {
+ // same, for the producer arm
+ }
+ raw = new MpscArrayQueue<>(16);
+ while (raw.offer(ELEMENT)) {
+ // same, through jctools' own rejection
+ }
+ draining = true;
+ drain =
+ new Thread(
+ () -> {
+ while (draining) {
+ // Not timed, and deliberately not throttled: the point is to keep the steady arm
+ // off the capacity boundary and to keep the counter's increment side busy.
+ queue.process(CAPACITY, e -> {});
+ }
+ },
+ "contended-admission-drain");
+ drain.setDaemon(true);
+ drain.start();
+ }
+
+ @TearDown
+ public void tearDown() throws InterruptedException {
+ draining = false;
+ drain.join(SECONDS.toMillis(5));
+ }
+
+ private WorkQueue create(int capacity) {
+ return backings == Backings.MPSC
+ ? WorkQueues.createMpscQueue(capacity)
+ : WorkQueues.createMpmcQueue(capacity);
+ }
+
+ /** Reserve-before-build: the producer is never asked, so nothing is allocated. */
+ @Benchmark
+ public void refusedProducer(Blackhole bh) {
+ bh.consume(fullPayloads.tryPut(BUILDER));
+ }
+
+ /**
+ * Build-then-offer: what the same rejection costs when the element is constructed before the
+ * queue gets a say. Consumed through the blackhole so it genuinely escapes, the way an element
+ * handed to a queue in another class does — otherwise escape analysis erases the allocation this
+ * arm exists to charge for.
+ */
+ @Benchmark
+ public void refusedBuildThenOffer(Blackhole bh) {
+ Payload payload = new Payload(ELEMENT, ELEMENT, System.nanoTime());
+ bh.consume(raw.offer(payload));
+ bh.consume(payload);
+ }
+
+ /** Two RMWs on the shared counter, every call, from every thread. */
+ @Benchmark
+ public void refusedQueue(Blackhole bh) {
+ bh.consume(full.tryPut(ELEMENT));
+ }
+
+ /** The bound jctools gives for free, for the delta. */
+ @Benchmark
+ public void refusedRaw(Blackhole bh) {
+ bh.consume(raw.offer(ELEMENT));
+ }
+
+ /**
+ * The commoner half: not full, with a consumer making room about as fast as producers take it, so
+ * the counter is incremented by a drain while it is decremented by admission -- contending on the
+ * same line from both directions.
+ *
+ * Every JMH thread is a producer here, and the consumer is the dedicated {@link #drain} thread
+ * started in setup rather than a {@code @Group} member. That is not a stylistic choice. A group's
+ * {@code @GroupThreads(1)} fixes the consumer count *per group*, and JMH instantiates as many
+ * groups as the thread count allows -- so {@code -Pjmh.threads=8} against a group of 4 yields two
+ * consumers. Two concurrent {@code poll()}s on the single-consumer MPSC ring do not fail; they
+ * spin inside jctools' gap-wait and the iteration never ends. Owning the consumer outright makes
+ * the arm correct at any thread count, which is what the project's spot-check flags hand it.
+ */
+ @Benchmark
+ public void steady(Blackhole bh) {
+ bh.consume(queue.tryPut(ELEMENT));
+ }
+}
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..e6b7be90244
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/BaseWorkQueue.java
@@ -0,0 +1,654 @@
+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.AtomicLong;
+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 state, 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();
+
+ /**
+ * Subtracted from {@link #state} once, by {@link #close()}. Closing then costs no flag of its
+ * own: it drives the permit count so far negative that no claim can ever succeed again, so a
+ * closed queue refuses through the same comparison that a full one does and admission has one
+ * word to read instead of two that have to agree.
+ *
+ *
Large enough to be unreachable from either side. Permits start at no more than {@link
+ * Integer#MAX_VALUE} and only places actually claimed are ever given back, so releases after a
+ * close cannot climb the offset; and the count is a {@code long} precisely because an unbounded
+ * queue seeds it with {@code Integer.MAX_VALUE}, which leaves an {@code int} no room above the
+ * bound to put this.
+ */
+ private static final long CLOSED_OFFSET = 1L << 40;
+
+ /** Any state below this has {@link #CLOSED_OFFSET} applied to it, and nothing else can be. */
+ private static final long CLOSED_MARK = -(1L << 39);
+
+ /**
+ * Places still available, not places used, biased by {@link #CLOSED_OFFSET} once closed. 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 AtomicLong state;
+
+ private final int capacity;
+
+ BaseWorkQueue(int capacity) {
+ this.capacity = capacity;
+ this.state = new AtomicLong(capacity);
+ }
+
+ /** The places left, with the closed bias taken back off. */
+ private static long permits(long state) {
+ return state < CLOSED_MARK ? state + CLOSED_OFFSET : state;
+ }
+
+ /**
+ * 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.
+ *
+ *
A plain read comes first, and it is what makes a refusal cheap. Without it a rejected
+ * admission paid two read-modify-writes on the one line every producer is already fighting over,
+ * at the capacity boundary, which is exactly where the most threads arrive at once -- {@code
+ * ContendedAdmissionBenchmark} priced that at roughly 960ns against 3.4ns for the same rejection
+ * taken on the backing's own producer index. A queue that is full, or closed, now turns a
+ * claimant away with a load. The decrement stays authoritative, so the bound is unaffected: the
+ * read can only cause a refusal, never an admission.
+ *
+ *
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 (state.get() < 1) {
+ return false;
+ }
+ if (state.decrementAndGet() >= 0) {
+ return true;
+ }
+ state.incrementAndGet();
+ return false;
+ }
+
+ private void releasePlace() {
+ state.incrementAndGet();
+ }
+
+ /**
+ * Counts nothing, unlike the producer admissions below. This one is shared with the retry path,
+ * where a refusal is a step rather than an outcome: a strategy handed a refused retry may still
+ * place the item somewhere else, and only its {@link RetryStrategy#onFailure} return says whether
+ * the item was finally lost. Each caller counts its own outcome, once.
+ */
+ 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()) {
+ dropped.increment();
+ 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()) {
+ dropped.increment();
+ return false;
+ }
+ T element;
+ try {
+ element = producer.produce(first, second);
+ } catch (Throwable t) {
+ releasePlace();
+ throw t;
+ }
+ return storeOrRelease(element);
+ }
+
+ /**
+ * The per-source-element half of {@link #tryPutBatch(Collection, Object, BiContextualProducer)}.
+ *
+ * Three outcomes, two of which report as not admitted for different reasons. A decline is the
+ * caller's own decision, so it gives its place back and counts nothing. A refusal — no place to
+ * claim, or a backing that would not take what was produced — gives the place back where there is
+ * one and counts a drop.
+ *
+ * @return whether the source element was admitted
+ */
+ @StrategyConsumer
+ private boolean admitEach(
+ E element,
+ C context,
+ @Strategy BiContextualProducer producer,
+ @Strategy RejectHandler onRejected) {
+ if (!claimPlace()) {
+ reject(element, onRejected);
+ return false;
+ }
+ T produced;
+ try {
+ produced = producer.produce(element, context);
+ } catch (Throwable t) {
+ releasePlace();
+ throw t;
+ }
+ if (produced == null) {
+ // Declined. The place goes back and nothing is counted, because nothing was lost.
+ releasePlace();
+ return false;
+ }
+ if (store(produced)) {
+ return true;
+ }
+ releasePlace();
+ reject(element, onRejected);
+ return false;
+ }
+
+ /** {@code null} rather than a no-op handler, so the count-only form adds a test and no call. */
+ @StrategyConsumer
+ private void reject(E element, @Strategy RejectHandler onRejected) {
+ dropped.increment();
+ if (onRejected != null) {
+ onRejected.onRejected(element);
+ }
+ }
+
+ /**
+ * The tail of every producer admission. A {@code null} is the producer declining, which is the
+ * caller's own decision: the place goes back and nothing is counted, because nothing was lost. A
+ * backing that would not take what was produced is a refusal, and is counted. Same three outcomes
+ * as {@link #admitEach}, which walks a source instead of taking one element.
+ */
+ private boolean storeOrRelease(T element) {
+ if (element == null) {
+ releasePlace();
+ return false;
+ }
+ if (store(element)) {
+ return true;
+ }
+ releasePlace();
+ dropped.increment();
+ 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;
+ }
+ requireElement(element);
+ 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 (int) Math.max(0, capacity - permits(state.get()));
+ }
+
+ @Override
+ public final boolean tryPut(T element) {
+ requireElement(element);
+ if (!admit(element)) {
+ dropped.increment();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public final boolean tryPut(Producer producer) {
+ return admit(producer, (ContextualProducer) PRODUCE);
+ }
+
+ @Override
+ public final boolean tryPut(C context, ContextualProducer producer) {
+ return admit(context, producer);
+ }
+
+ @Override
+ public final boolean tryPut(
+ C1 first, C2 second, BiContextualProducer producer) {
+ return 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 int tryPutBatch(
+ Collection source,
+ C context,
+ BiContextualProducer producer) {
+ return tryPutBatch(source, context, producer, null);
+ }
+
+ @Override
+ public final int tryPutBatch(
+ Collection source,
+ C context,
+ BiContextualProducer producer,
+ RejectHandler onRejected) {
+ int admitted = 0;
+ for (E element : source) {
+ if (admitEach(element, context, producer, onRejected)) {
+ admitted++;
+ }
+ }
+ return admitted;
+ }
+
+ @Override
+ public final Reservation tryReserve() {
+ boolean granted = 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) {
+ // No counting here. A refused retry is one step of a decision the strategy is still
+ // making; the item is counted lost exactly once, when onFailure reports it gave up.
+ return admit(new Retry<>(item, attempt));
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean retry(T... items) {
+ boolean all = items.length > 0;
+ for (T item : items) {
+ all &= retry(item);
+ }
+ return all;
+ }
+ };
+ }
+
+ /**
+ * The one place the module says what a {@code null} element is. Neither backing can hold one, so
+ * there is no outcome to report and nothing to count -- only a caller with a bug. Thrown before a
+ * place is claimed, so a rejected call costs the queue nothing.
+ */
+ private static void requireElement(Object element) {
+ if (element == null) {
+ throw new NullPointerException("a queue cannot hold null");
+ }
+ }
+
+ @Override
+ public final long dropped() {
+ return dropped.sum();
+ }
+
+ @Override
+ public final void close() {
+ long current;
+ do {
+ current = state.get();
+ if (current < CLOSED_MARK) {
+ // Already closed. Applying the offset twice would walk the state toward a second
+ // threshold nothing checks, and the second close has nothing left to say.
+ return;
+ }
+ } while (!state.compareAndSet(current, current - CLOSED_OFFSET));
+ }
+
+ @Override
+ public final boolean isClosed() {
+ return state.get() < CLOSED_MARK;
+ }
+
+ @Override
+ public final void clear() {
+ discardAll();
+ }
+
+ @Override
+ public final void shutdown() {
+ close();
+ 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 queue = new ConcurrentLinkedQueue<>();
+
+ /**
+ * @param capacity the bound, or {@link Integer#MAX_VALUE} to leave the queue unbounded
+ */
+ LinkedWorkQueue(int capacity) {
+ super(capacity);
+ }
+
+ @Override
+ boolean store(Object element) {
+ return queue.offer(element);
+ }
+
+ @Override
+ Object retrieve() {
+ return queue.poll();
+ }
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/MaxRetries.java b/utils/queue-utils/src/main/java/datadog/common/queue/MaxRetries.java
new file mode 100644
index 00000000000..bfe8bce7964
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/MaxRetries.java
@@ -0,0 +1,15 @@
+package datadog.common.queue;
+
+/** A {@link RetryStrategy} that resubmits an item until a fixed attempt count is reached. */
+public final class MaxRetries implements RetryStrategy {
+ private final int maxRetries;
+
+ public MaxRetries(int maxRetries) {
+ this.maxRetries = maxRetries;
+ }
+
+ @Override
+ public boolean onFailure(T item, int attempt, Throwable failure, RetryQueue retryQueue) {
+ return attempt < maxRetries && retryQueue.retry(item);
+ }
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/MpscWorkQueue.java b/utils/queue-utils/src/main/java/datadog/common/queue/MpscWorkQueue.java
new file mode 100644
index 00000000000..95b8d087f27
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/MpscWorkQueue.java
@@ -0,0 +1,43 @@
+package datadog.common.queue;
+
+import org.jctools.queues.MessagePassingQueue;
+
+/**
+ * A {@link WorkQueue} over a JCTools MPSC array queue: many producers, one consumer, no per-element
+ * node. The preferred backing.
+ *
+ * Storage only. The bound and the reserve-before-construct guarantee both live in {@link
+ * BaseWorkQueue}, which spends a place before it calls any producer, so by the time an element
+ * reaches {@link #store} the ring is known to have room for it.
+ *
+ *
That the ring could have enforced its own bound, inside a CAS it was performing anyway, is the
+ * cost of this arrangement — see {@link BaseWorkQueue} for what it buys. What it avoids is holding
+ * a ring position open across a caller-controlled gap: the ring reports a claimed-but-unfilled
+ * position as empty, so a reservation that held one would stall the consumer, and would need a
+ * placeholder object per reservation for the consumer to tell an abandoned position from a pending
+ * one.
+ */
+final class MpscWorkQueue extends BaseWorkQueue {
+
+ private final MessagePassingQueue queue;
+
+ MpscWorkQueue(int requestedCapacity) {
+ this(Queues.mpscArrayQueue(requestedCapacity));
+ }
+
+ /** Takes the queue already built, so the bound can be the capacity it actually rounded up to. */
+ private MpscWorkQueue(MessagePassingQueue queue) {
+ super(queue.capacity());
+ this.queue = queue;
+ }
+
+ @Override
+ boolean store(Object element) {
+ return queue.offer(element);
+ }
+
+ @Override
+ Object retrieve() {
+ return queue.poll();
+ }
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/Producer.java b/utils/queue-utils/src/main/java/datadog/common/queue/Producer.java
new file mode 100644
index 00000000000..8b384c7daea
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/Producer.java
@@ -0,0 +1,22 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Produces an element for admission into a {@link WorkQueue}.
+ *
+ * A producer is only invoked once a place has been claimed, so it is never called for an element
+ * that will be rejected. Implementations must be non-capturing — a {@code static final} constant of
+ * the concrete type, or a lambda that closes over nothing — which is what {@link Strategy} marks.
+ *
+ *
That is not a preference, it is the whole reason this form exists. A capturing lambda
+ * allocates once per call, and so does a {@link Reservation}; the reservation is straight-line code
+ * that keeps whatever the call site had hoisted and needs no context parameters. So a producer that
+ * captures is strictly worse than the reserve form it was meant to improve on. If the state will
+ * not fit the context parameters, use {@link WorkQueue#tryReserve} rather than closing over it.
+ */
+@Strategy
+@FunctionalInterface
+public interface Producer {
+ T produce();
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/RejectHandler.java b/utils/queue-utils/src/main/java/datadog/common/queue/RejectHandler.java
new file mode 100644
index 00000000000..afab1042290
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/RejectHandler.java
@@ -0,0 +1,27 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Sees the source elements a batch admission could not take, for a caller that has somewhere to put
+ * them: a resubmission list, a spill buffer, a per-kind counter.
+ *
+ * A handler rather than a returned collection, so a caller that only wanted the count is charged
+ * nothing for one it would have thrown away, and a caller that wants the elements chooses where
+ * they go instead of receiving a list it has to copy out of. The admission side's answer to {@link
+ * ExceptionHandler}, which does the same thing for a consumer's failures.
+ *
+ *
Only refusals reach a handler. An element the producer declined by returning {@code null} was
+ * the caller's own decision and is not a rejection. The queue cannot hold that line perfectly at
+ * the boundary, though: a place is claimed before the producer is asked, so once the queue is full
+ * a handler sees source elements the producer would have declined, indistinguishable from the rest.
+ * A caller resubmitting what it is handed should apply its own decline rule again.
+ *
+ * @see WorkQueue#tryPutBatch(java.util.Collection, Object, BiContextualProducer, RejectHandler)
+ */
+@Strategy
+@FunctionalInterface
+public interface RejectHandler {
+ /** Called on the admitting thread, once per source element that could not be admitted. */
+ void onRejected(E element);
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/Reservation.java b/utils/queue-utils/src/main/java/datadog/common/queue/Reservation.java
new file mode 100644
index 00000000000..e901ab86bf2
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/Reservation.java
@@ -0,0 +1,57 @@
+package datadog.common.queue;
+
+/**
+ * A claimed place in a {@link WorkQueue}, for a caller whose work between claiming and filling
+ * cannot be expressed as a {@link Producer}.
+ *
+ * What is claimed is capacity, never a position: the element joins the queue where it is filled,
+ * not where it was claimed, so an open reservation holds no place a consumer could be waiting on
+ * and cannot stall one. A reservation that is neither filled nor closed does leak its capacity,
+ * quietly and permanently, which is why this is an {@link AutoCloseable} meant for
+ * try-with-resources.
+ *
+ *
A refused claim is a reservation too, rather than a {@code null}, and one that quietly
+ * discards whatever is filled into it. Nothing about the failed path throws, so the shortest
+ * correct call site is also the obvious one:
+ *
+ *
{@code
+ * try (Reservation place = queue.tryReserve()) {
+ * place.fill(buildTask());
+ * }
+ * }
+ *
+ * Consulting {@link #granted} first is what buys the reserve-first guarantee — skip the build
+ * and nothing is allocated for a queue that had no room for it:
+ *
+ *
{@code
+ * try (Reservation place = queue.tryReserve()) {
+ * if (place.granted()) {
+ * place.fill(buildTask());
+ * }
+ * }
+ * }
+ */
+public interface Reservation extends AutoCloseable {
+
+ /**
+ * Whether a place was actually claimed. Worth asking before building anything expensive: a
+ * refused reservation accepts a fill and throws it away, so checking is what turns
+ * allocate-then-drop into never-allocate.
+ *
+ * @return whether a fill will be kept
+ */
+ boolean granted();
+
+ /**
+ * Publishes {@code element} into the claimed place. A granted place is already paid for, so this
+ * cannot be rejected; a refused one discards the element, having already counted the drop.
+ */
+ void fill(T element);
+
+ /**
+ * Gives the place back if it was never filled, immediately. Filling first makes this a no-op, and
+ * nothing is ever consumed for a released place.
+ */
+ @Override
+ void close();
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/RetryQueue.java b/utils/queue-utils/src/main/java/datadog/common/queue/RetryQueue.java
new file mode 100644
index 00000000000..feba695044a
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/RetryQueue.java
@@ -0,0 +1,36 @@
+package datadog.common.queue;
+
+/**
+ * The capability to resubmit work after a consumer failure.
+ *
+ * Only obtainable inside {@link RetryStrategy#onFailure}, never from a plain consumer, so
+ * re-enqueue-after-failure stays visibly distinct from ordinary admission.
+ */
+public interface RetryQueue {
+ /**
+ * Resubmits the failed item.
+ *
+ * The failed item's place was given back when it was consumed, so this claims a place like any
+ * other admission and can be rejected if the queue filled up behind it. A refusal is not itself
+ * counted as a drop: the item is counted once, when {@link RetryStrategy#onFailure} returns
+ * {@code false} to say the strategy gave up. A strategy that cannot resubmit must therefore
+ * report that, or the item is lost without being counted. This is the overload every ordinary
+ * strategy wants: it resubmits without allocating the array the varargs form needs.
+ *
+ * @return whether the item was resubmitted
+ */
+ boolean retry(T item);
+
+ /**
+ * Resubmits several items in place of the failed item.
+ *
+ *
Each piece claims its own place, so a partition can be admitted only in part, and the return
+ * value reports whether all of them made it. As with the single-item overload, a refusal is not
+ * counted here; a strategy that partially resubmits and returns {@code true} is telling the queue
+ * the remainder was its own to lose.
+ *
+ * @return whether every item was resubmitted
+ */
+ @SuppressWarnings("unchecked")
+ boolean retry(T... items);
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/RetryStrategy.java b/utils/queue-utils/src/main/java/datadog/common/queue/RetryStrategy.java
new file mode 100644
index 00000000000..e8da41df964
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/RetryStrategy.java
@@ -0,0 +1,21 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Decides what happens to an item whose consumer threw.
+ *
+ *
Invoked only on failure — a successful consumption needs no callback. The return value reports
+ * the decision; it does not report whether the item will eventually succeed. Logging and counting
+ * are the caller's to compose here: this API performs neither.
+ */
+@Strategy
+@FunctionalInterface
+public interface RetryStrategy {
+ /**
+ * @param attempt how many times this item has been consumed unsuccessfully, including now, so the
+ * first failure reports {@code 1}
+ * @return {@code true} if the item was resubmitted, {@code false} if the strategy gave up
+ */
+ boolean onFailure(T item, int attempt, Throwable failure, RetryQueue retryQueue);
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/WorkQueue.java b/utils/queue-utils/src/main/java/datadog/common/queue/WorkQueue.java
new file mode 100644
index 00000000000..3c61087b900
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/WorkQueue.java
@@ -0,0 +1,294 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+import datadog.trace.api.function.StrategyConsumer;
+import java.util.Collection;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+/**
+ * A bounded handoff point between producers and a consumer, with admission that never builds an
+ * element it is going to reject.
+ *
+ * Capacity is fixed by construction. A queue never grows in response to fullness: full means
+ * drop and count. Admission claims a place before invoking any producer, so a rejected element is
+ * never constructed at all — the guarantee that makes it safe to hand this a producer that
+ * allocates heavily, since the allocation cannot happen on the path where it would be wasted.
+ *
+ *
What is claimed is capacity, never a position, so a producer never holds up a consumer. It
+ * does hold a place other producers could have used, though: work that blocks, or that takes
+ * appreciably longer than an allocation, is paid for by everyone else admitting to this queue.
+ * Producers should build their element and nothing else.
+ *
+ *
Prefer the producer forms, and treat {@link #tryReserve} as the fallback. The distinction is
+ * {@code forEach} against {@code Iterator}: with a producer the queue owns the sequence, claiming
+ * and building in the order it knows to be safe, and there is no protocol for a caller to get
+ * wrong. A reservation hands that loop back — the caller must check {@link Reservation#granted},
+ * must fill or close, and an abandoned one is capacity nobody can see or reclaim, the same way a
+ * half-consumed iterator is state its collection cannot account for. Reach for it when the work
+ * between claiming and filling genuinely will not fold into a callback, and use {@code tryPut}
+ * everywhere else.
+ *
+ *
Consumption is synchronous and happens in the caller's frame; the boolean returned by the
+ * {@code process} methods reports whether there was an item to work on, which is the signal a drain
+ * loop needs, and says nothing about whether the consumer succeeded. A consumer that throws throws
+ * out of {@code process} — the queue takes no view on failure it was not given one for, and never
+ * logs. Say what should happen instead by calling {@link #processOrRetry} with a {@link
+ * RetryStrategy}, or {@link #processOrHandle} with an {@link ExceptionHandler} when the answer is
+ * only ever to record it and move on. Those are separate names rather than overloads because a
+ * lambda or method reference cannot always tell two same-arity callbacks apart.
+ *
+ *
Nulls carry meaning in three places and are a bug everywhere else. A context may be
+ * null: the queue carries it to a producer or consumer and never looks at it, so an absent one is
+ * the caller's business. An optional {@link RejectHandler} may be null, which says exactly what the
+ * overload without it says. And a producer's return may be null, which is that producer
+ * declining the element it was asked to build — the place goes back, nothing is admitted, and
+ * nothing is counted against {@link #dropped}, because a decision is not a loss.
+ *
+ *
Everything else is required. An element is never null, because neither backing can hold
+ * one: there is no outcome to report, so {@code tryPut} and {@link Reservation#fill fill} throw
+ * instead of returning, and they throw before claiming a place so that a call with a bug in it
+ * costs the queue nothing. A null inside a batch throws partway through, abandoning the rest.
+ * Producers, consumers, retry strategies and exception handlers are required too — a null
+ * there has no sensible reading, and it surfaces as the thrown {@link NullPointerException} of the
+ * call that would have used it, with any place already claimed given back first.
+ */
+public interface WorkQueue {
+
+ /**
+ * @return whether the element was admitted
+ * @throws NullPointerException if the element is null, thrown before a place is claimed
+ */
+ boolean tryPut(T element);
+
+ /**
+ * Admits an element, constructing it only once a slot is reserved.
+ *
+ * @return whether the element was admitted
+ */
+ @StrategyConsumer
+ boolean tryPut(@Strategy Producer producer);
+
+ /**
+ * Admits an element derived from {@code context}, constructing it only once a slot is reserved.
+ *
+ * @return whether the element was admitted
+ */
+ @StrategyConsumer
+ boolean tryPut(C context, @Strategy ContextualProducer producer);
+
+ /**
+ * Admits an element derived from two contexts, constructing it only once a slot is reserved.
+ *
+ * @return whether the element was admitted
+ * @see BiContextualProducer
+ */
+ @StrategyConsumer
+ boolean tryPut(
+ C1 first,
+ C2 second,
+ @Strategy BiContextualProducer producer);
+
+ /**
+ * @return the elements that were not admitted, empty if all were
+ */
+ @SuppressWarnings("unchecked")
+ Collection tryPutBatch(T... elements);
+
+ /**
+ * @return the elements that were not admitted, empty if all were
+ */
+ Collection tryPutBatch(Collection elements);
+
+ /**
+ * Admits an element per source element, constructing each only once a slot is reserved for it.
+ * The queue owns the walk, so the producer is asked only for elements there is already room for,
+ * and a caller batching work this way never holds capacity of its own.
+ *
+ * The producer may decline a source element by returning {@code null}. That is an explicit
+ * decision by the caller rather than a loss, so a declined element is not counted against {@link
+ * #dropped()} and does not count as admitted; the place claimed for it is simply given back.
+ *
+ *
A count rather than the refused source elements, because the count is the number a caller
+ * can act on and the elements are not. A caller that knows how many it meant to admit gets the
+ * exact shortfall by subtraction, with its own declines excluded from both sides. The refused
+ * elements cannot be that precise: a place is claimed before the producer is asked, so a full
+ * queue cannot tell a genuine refusal from an element the producer would have declined anyway,
+ * and hands back — and counts against {@link #dropped()} — some of each.
+ *
+ *
{@code context} is the one value the whole batch shares and a source element cannot recover
+ * on its own — a schema, a clock reading, a per-batch buffer. It is read once here rather than
+ * per element, which is the hoist the single-element form spells out in {@link
+ * BiContextualProducer}.
+ *
+ *
{@link Collection} rather than {@link Iterable} because admission runs while there is room,
+ * and a queue with a live consumer keeps making room: a source with no end would not terminate.
+ *
+ *
Reach for this only when the walk exists to admit and nothing else. The queue stops asking
+ * once it runs out of room, so the producer is the only per-source-element hook a caller gets and
+ * it is reached only for elements there was room for. A loop that also carries something across
+ * its iterations — a count of what it considered, a flag OR-ed over the whole source, a decision
+ * about the batch as a whole — needs every source element regardless of admission, and hands back
+ * more per element than a producer can return. Such a caller keeps its own loop and admits one
+ * element at a time; that is not a shortcoming of the loop.
+ *
+ * @return how many elements were admitted
+ * @see BiContextualProducer
+ */
+ @StrategyConsumer
+ int tryPutBatch(
+ Collection source,
+ C context,
+ @Strategy BiContextualProducer producer);
+
+ /**
+ * As {@link #tryPutBatch(Collection, Object, BiContextualProducer)}, handing each source element
+ * it could not admit to {@code onRejected} on the way past.
+ *
+ * Elements the producer declined do not reach the handler; refusals do. See {@link
+ * RejectHandler} for the one place that line blurs.
+ *
+ * @return how many elements were admitted
+ * @see RejectHandler
+ */
+ @StrategyConsumer
+ int tryPutBatch(
+ Collection source,
+ C context,
+ @Strategy BiContextualProducer producer,
+ @Strategy RejectHandler onRejected);
+
+ /**
+ * Claims a place without supplying its element, for a caller whose work between claiming and
+ * filling cannot be expressed as a {@link Producer}.
+ *
+ * What is reserved is capacity, not a position — {@link Reservation#fill} cannot be rejected,
+ * and the element joins the queue where it is filled. Nothing is held open that a consumer could
+ * be waiting on, so a thread may safely reserve and consume, but a reservation that is never
+ * filled or closed leaks its capacity for good. Use try-with-resources.
+ *
+ *
Never {@code null}: a refusal comes back as a reservation that reports {@link
+ * Reservation#granted} as {@code false} and discards anything filled into it. Nothing on the
+ * refused path throws, so the try-with-resources is always safe; checking {@code granted} is what
+ * lets the caller skip building an element the queue had no room for.
+ *
+ * @return the claimed capacity, or a refused reservation if there was none to claim
+ */
+ Reservation tryReserve();
+
+ /**
+ * Consumes one item, if there is one. A throwing consumer propagates.
+ *
+ * @return whether there was an item to consume
+ */
+ boolean process(Consumer consumer);
+
+ /**
+ * Consumes one item, if there is one, handing a throwing consumer's failure to {@code
+ * retryStrategy} rather than propagating it.
+ *
+ * @return whether there was an item to consume
+ */
+ boolean processOrRetry(Consumer consumer, @Strategy RetryStrategy retryStrategy);
+
+ /**
+ * Consumes one item, if there is one, handing a throwing consumer's failure to {@code
+ * exceptionHandler} rather than propagating it. The item is dropped.
+ *
+ * @return whether there was an item to consume
+ */
+ boolean processOrHandle(
+ Consumer consumer, @Strategy ExceptionHandler exceptionHandler);
+
+ /**
+ * Consumes one item, if there is one. A throwing consumer propagates.
+ *
+ * @return whether there was an item to consume
+ */
+ boolean process(C context, BiConsumer consumer);
+
+ /**
+ * Consumes one item, if there is one, handing a throwing consumer's failure to {@code
+ * retryStrategy} rather than propagating it.
+ *
+ * @return whether there was an item to consume
+ */
+ boolean processOrRetry(
+ C context,
+ BiConsumer consumer,
+ @Strategy RetryStrategy retryStrategy);
+
+ /**
+ * Consumes one item, if there is one, handing a throwing consumer's failure to {@code
+ * exceptionHandler} rather than propagating it. The item is dropped.
+ *
+ * @return whether there was an item to consume
+ */
+ boolean processOrHandle(
+ C context,
+ BiConsumer consumer,
+ @Strategy ExceptionHandler exceptionHandler);
+
+ /**
+ * Consumes up to {@code limit} items, stopping early when the queue runs dry.
+ *
+ * The limit is required, and there is no consume-until-empty form. Against live producers that
+ * has no reason to ever return; on an unbounded backing there is not even a capacity to fall back
+ * on as an implicit bound; and a {@link RetryStrategy} re-admits behind a consumer that is still
+ * draining, so only a caller-named ceiling guarantees the batch ends. The limit is also the
+ * caller's latency knob: a drain occupies its thread until it is done, which matters most where
+ * that thread is shared with other subsystems.
+ *
+ *
A throwing consumer propagates, abandoning the rest of the batch. Items already consumed
+ * stay consumed and the count is lost with the stack unwind, so a caller that needs it should
+ * drain in smaller batches or handle failure per item with a {@link RetryStrategy}.
+ *
+ * @return how many items were consumed, which is {@code limit} when the batch filled and there
+ * may be more waiting
+ */
+ int process(int limit, Consumer consumer);
+
+ /**
+ * Consumes up to {@code limit} items, stopping early when the queue runs dry.
+ *
+ * @return how many items were consumed
+ * @see #process(int, Consumer)
+ */
+ int process(int limit, C context, BiConsumer consumer);
+
+ int size();
+
+ /**
+ * @return how many elements have been rejected on admission, or abandoned by a {@link
+ * RetryStrategy}, over this queue's lifetime
+ */
+ long dropped();
+
+ /**
+ * Stops future admission, leaving current contents alone so a consumer can finish its backlog.
+ *
+ * A caller distinguishes "transiently full, worth retrying" from "permanently done" by asking
+ * {@link #isClosed()}; the {@code boolean} returned by admission does not carry the difference.
+ */
+ void close();
+
+ boolean isClosed();
+
+ /** Discards current contents without affecting admission. */
+ void clear();
+
+ /**
+ * {@link #close() Closes} and then {@link #clear() clears} — the flag before the discard, so a
+ * producer that has not started yet cannot begin.
+ *
+ *
Not atomic, and not made atomic by being one call. A producer already past the closed check,
+ * or an in-flight retry lease, can still store its element after the discard has run, and that
+ * element then sits in a queue nothing will drain again. Ordering the flag first bounds the
+ * survivors to those already in flight rather than eliminating them.
+ *
+ *
A caller that needs the queue provably empty has to quiesce its producers first and shut
+ * down after. The queue cannot do that half on the caller's behalf: it knows when it is closed,
+ * but not who is still holding a place or how long they mean to hold it.
+ */
+ void shutdown();
+}
diff --git a/utils/queue-utils/src/main/java/datadog/common/queue/WorkQueues.java b/utils/queue-utils/src/main/java/datadog/common/queue/WorkQueues.java
new file mode 100644
index 00000000000..bc9e1b84146
--- /dev/null
+++ b/utils/queue-utils/src/main/java/datadog/common/queue/WorkQueues.java
@@ -0,0 +1,64 @@
+package datadog.common.queue;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Factory methods for {@link WorkQueue} buffers: bounded handoff points that count what they drop
+ * and never build an element they are going to reject.
+ *
+ *
Distinct from {@link Queues}, which hands back a raw JCTools queue for the caller to drive
+ * itself. A buffer created here owns its backing — which implementation it is stays an
+ * implementation detail, so a call site can be re-backed without changing.
+ */
+public final class WorkQueues {
+
+ private WorkQueues() {}
+
+ /**
+ * Creates a bounded Multiple Producer, Single Consumer buffer backed by an MPSC array queue.
+ *
+ *
The preferred backing: no per-element node, constant-time {@link WorkQueue#size()}, and
+ * admission that claims a slot before invoking a producer, so an element that will not fit is
+ * never built.
+ *
+ *
Single Consumer is a requirement, not a characteristic. Producers may be any number of
+ * threads, but every call that takes elements out -- any {@code process}, {@code processOrRetry}
+ * or {@code processOrHandle} overload, plus {@link WorkQueue#clear} and {@link
+ * WorkQueue#shutdown} -- must come from one thread. A second consumer is not rejected and does
+ * not throw: the two can spin inside the ring's gap-wait indefinitely, which presents as a hang
+ * rather than a failure. Use {@link #createMpmcQueue} where more than one thread drains.
+ *
+ * @param requestedCapacity the bound. Will be rounded to the next power of two.
+ */
+ public static WorkQueue createMpscQueue(int requestedCapacity) {
+ return new MpscWorkQueue<>(requestedCapacity);
+ }
+
+ /**
+ * Creates a bounded Multiple Producer, Multiple Consumer buffer backed by a {@link
+ * ConcurrentLinkedQueue}.
+ *
+ * For call sites that need several consumers. It keeps the linked queue's per-element node, so
+ * it buys the admission and lifecycle contract, an enforceable bound and a constant-time {@link
+ * WorkQueue#size()}, but not the allocation win — prefer {@link #createMpscQueue} where a single
+ * consumer is possible.
+ *
+ * @param capacity the bound
+ */
+ public static WorkQueue createMpmcQueue(int capacity) {
+ return new LinkedWorkQueue<>(capacity);
+ }
+
+ /**
+ * Creates an unbounded Multiple Producer, Multiple Consumer buffer backed by a {@link
+ * ConcurrentLinkedQueue}.
+ *
+ * Unbounded means admission never rejects and {@link WorkQueue#dropped()} only ever counts
+ * items abandoned by a retry strategy. Intended as a migration step for call sites that are
+ * unbounded today: adopt the interface here, then pick a bound and move to {@link
+ * #createMpscQueue}.
+ */
+ public static WorkQueue createUnboundedMpmcQueue() {
+ return new LinkedWorkQueue<>(Integer.MAX_VALUE);
+ }
+}
diff --git a/utils/queue-utils/src/test/java/datadog/common/queue/MpscWorkQueueStressTest.java b/utils/queue-utils/src/test/java/datadog/common/queue/MpscWorkQueueStressTest.java
new file mode 100644
index 00000000000..18b29593d9f
--- /dev/null
+++ b/utils/queue-utils/src/test/java/datadog/common/queue/MpscWorkQueueStressTest.java
@@ -0,0 +1,261 @@
+package datadog.common.queue;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerArray;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Contention tests for the MPSC backing, where the admission contract actually has to hold: many
+ * producers claiming slots against a single consumer freeing them.
+ *
+ * What is being checked is conservation. Every element a producer was told it admitted must
+ * reach the consumer exactly once, and every element it was told was rejected must be counted as
+ * dropped — so admitted plus dropped accounts for everything offered, with nothing lost, duplicated
+ * or invented in between.
+ */
+class MpscWorkQueueStressTest {
+
+ private static final int PRODUCERS = 8;
+ private static final int PER_PRODUCER = 20_000;
+ private static final int TOTAL = PRODUCERS * PER_PRODUCER;
+ private static final int CAPACITY = 128;
+ private static final long TIMEOUT_SECONDS = 60;
+
+ @Test
+ void conservesEveryElementUnderContention() throws Exception {
+ WorkQueue queue = WorkQueues.createMpscQueue(CAPACITY);
+ AtomicIntegerArray timesSeen = new AtomicIntegerArray(TOTAL);
+ AtomicInteger admitted = new AtomicInteger();
+ AtomicInteger consumed = new AtomicInteger();
+
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch producersDone = new CountDownLatch(PRODUCERS);
+
+ for (int p = 0; p < PRODUCERS; p++) {
+ final int producer = p;
+ Thread thread =
+ new Thread(
+ () -> {
+ await(start);
+ try {
+ for (int i = 0; i < PER_PRODUCER; i++) {
+ int value = producer * PER_PRODUCER + i;
+ if (queue.tryPut(value)) {
+ admitted.incrementAndGet();
+ }
+ }
+ } finally {
+ producersDone.countDown();
+ }
+ },
+ "producer-" + p);
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ AtomicBoolean consumerFailed = new AtomicBoolean();
+ Thread consumer =
+ new Thread(
+ () -> {
+ boolean producersFinished = false;
+ while (true) {
+ boolean hadWork = queue.process(value -> timesSeen.incrementAndGet(value));
+ if (hadWork) {
+ consumed.incrementAndGet();
+ } else if (producersFinished) {
+ return;
+ } else {
+ producersFinished = producersDone.getCount() == 0;
+ Thread.yield();
+ }
+ }
+ },
+ "consumer");
+ consumer.setDaemon(true);
+ consumer.setUncaughtExceptionHandler((t, e) -> consumerFailed.set(true));
+ consumer.start();
+
+ start.countDown();
+ assertTrue(
+ producersDone.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "producers did not finish in time");
+ consumer.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertFalse(consumer.isAlive(), "consumer did not finish in time");
+ assertFalse(consumerFailed.get(), "consumer thread threw");
+
+ assertEquals(
+ admitted.get(), consumed.get(), "every admitted element reaches the consumer once");
+ assertEquals(TOTAL - admitted.get(), queue.dropped(), "every rejection is counted");
+ assertEquals(0, queue.size());
+
+ for (int value = 0; value < TOTAL; value++) {
+ int seen = timesSeen.get(value);
+ assertTrue(seen <= 1, "element " + value + " was consumed " + seen + " times");
+ }
+ }
+
+ /**
+ * The reserve-before-construct guarantee under contention: producers race for a capacity that is
+ * never freed, so no producer may ever run.
+ */
+ @Test
+ void neverInvokesProducerWhileFull() throws Exception {
+ WorkQueue queue = WorkQueues.createMpscQueue(CAPACITY);
+ while (queue.tryPut(0)) {
+ // fill it, and leave it full — nothing consumes
+ }
+ // the loop above ends on a rejection, which is itself a drop
+ long droppedWhileFilling = queue.dropped();
+
+ AtomicInteger produced = new AtomicInteger();
+ AtomicInteger admittedAfterFull = new AtomicInteger();
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(PRODUCERS);
+
+ for (int p = 0; p < PRODUCERS; p++) {
+ Thread thread =
+ new Thread(
+ () -> {
+ await(start);
+ try {
+ for (int i = 0; i < PER_PRODUCER; i++) {
+ boolean landed =
+ queue.tryPut(
+ produced,
+ counter -> {
+ counter.incrementAndGet();
+ return 1;
+ });
+ if (landed) {
+ admittedAfterFull.incrementAndGet();
+ }
+ }
+ } finally {
+ done.countDown();
+ }
+ },
+ "producer-" + p);
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ start.countDown();
+ assertTrue(done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "producers did not finish in time");
+
+ assertEquals(0, admittedAfterFull.get(), "a full queue admits nothing");
+ assertEquals(0, produced.get(), "no element may be built for a slot that was never claimed");
+ assertEquals(
+ droppedWhileFilling + (long) PRODUCERS * PER_PRODUCER,
+ queue.dropped(),
+ "every rejected admission is counted");
+ }
+
+ /**
+ * Reservations mixed into ordinary admission under contention: the consumer has to tell a place
+ * that is still being filled from an element that is ready, and from a place that was abandoned,
+ * without losing or duplicating anything behind it.
+ */
+ @Test
+ void conservesElementsWhenProducersReserve() throws Exception {
+ WorkQueue queue = WorkQueues.createMpscQueue(CAPACITY);
+ AtomicIntegerArray timesSeen = new AtomicIntegerArray(TOTAL);
+ AtomicInteger admitted = new AtomicInteger();
+ AtomicInteger consumed = new AtomicInteger();
+
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch producersDone = new CountDownLatch(PRODUCERS);
+
+ for (int p = 0; p < PRODUCERS; p++) {
+ final int producer = p;
+ Thread thread =
+ new Thread(
+ () -> {
+ await(start);
+ try {
+ for (int i = 0; i < PER_PRODUCER; i++) {
+ int value = producer * PER_PRODUCER + i;
+ switch (i % 3) {
+ case 0:
+ if (queue.tryPut(value)) {
+ admitted.incrementAndGet();
+ }
+ break;
+ case 1:
+ try (Reservation place = queue.tryReserve()) {
+ if (place.granted()) {
+ place.fill(value);
+ admitted.incrementAndGet();
+ }
+ }
+ break;
+ default:
+ // claimed and then abandoned: the consumer must skip it
+ try (Reservation place = queue.tryReserve()) {
+ // no fill
+ }
+ break;
+ }
+ }
+ } finally {
+ producersDone.countDown();
+ }
+ },
+ "producer-" + p);
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ AtomicBoolean consumerFailed = new AtomicBoolean();
+ Thread consumer =
+ new Thread(
+ () -> {
+ boolean producersFinished = false;
+ while (true) {
+ boolean hadWork = queue.process(value -> timesSeen.incrementAndGet(value));
+ if (hadWork) {
+ consumed.incrementAndGet();
+ } else if (producersFinished) {
+ return;
+ } else {
+ producersFinished = producersDone.getCount() == 0;
+ Thread.yield();
+ }
+ }
+ },
+ "consumer");
+ consumer.setDaemon(true);
+ consumer.setUncaughtExceptionHandler((t, e) -> consumerFailed.set(true));
+ consumer.start();
+
+ start.countDown();
+ assertTrue(
+ producersDone.await(TIMEOUT_SECONDS, TimeUnit.SECONDS), "producers did not finish in time");
+ consumer.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertFalse(consumer.isAlive(), "consumer did not finish in time");
+ assertFalse(consumerFailed.get(), "consumer thread threw");
+
+ assertEquals(
+ admitted.get(), consumed.get(), "every filled place reaches the consumer exactly once");
+ assertEquals(0, queue.size(), "no abandoned place is left holding capacity");
+
+ for (int value = 0; value < TOTAL; value++) {
+ int seen = timesSeen.get(value);
+ assertTrue(seen <= 1, "element " + value + " was consumed " + seen + " times");
+ }
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+}
diff --git a/utils/queue-utils/src/test/java/datadog/common/queue/WorkQueueContractTest.java b/utils/queue-utils/src/test/java/datadog/common/queue/WorkQueueContractTest.java
new file mode 100644
index 00000000000..61b2307f897
--- /dev/null
+++ b/utils/queue-utils/src/test/java/datadog/common/queue/WorkQueueContractTest.java
@@ -0,0 +1,983 @@
+package datadog.common.queue;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.IntFunction;
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/** The behaviour every backing must share, exercised against each of them. */
+class WorkQueueContractTest {
+
+ private static final int CAPACITY = 4;
+
+ static Stream boundedQueues() {
+ return Stream.of(
+ Arguments.of("mpsc", (IntFunction>) WorkQueues::createMpscQueue),
+ Arguments.of("mpmc", (IntFunction>) WorkQueues::createMpmcQueue));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void admitsUpToCapacityThenDrops(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ assertEquals(CAPACITY, queue.size());
+ assertFalse(queue.tryPut("overflow"));
+ assertEquals(CAPACITY, queue.size());
+ assertEquals(1, queue.dropped());
+ }
+
+ /** The point of the whole API: a rejected element is never built. */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void doesNotInvokeProducerWhenFull(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ AtomicBoolean produced = new AtomicBoolean();
+ assertFalse(
+ queue.tryPut(
+ produced,
+ flag -> {
+ flag.set(true);
+ return "built";
+ }));
+ assertFalse(produced.get(), "producer ran for an element that could not be admitted");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void invokesProducerWhenThereIsRoom(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertTrue(queue.tryPut("ctx", context -> context + "-built"));
+ List consumed = consumeAll(queue);
+ assertEquals(Arrays.asList("ctx-built"), consumed);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void batchAdmissionReportsRejectedElements(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ Collection rejected = queue.tryPutBatch("a", "b", "c", "d", "e", "f");
+ assertEquals(Arrays.asList("e", "f"), new ArrayList<>(rejected));
+ assertEquals(2, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void collectionAdmissionReportsRejectedElements(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ Collection rejected = queue.tryPutBatch(Arrays.asList("a", "b", "c", "d", "e", "f"));
+ assertEquals(Arrays.asList("e", "f"), new ArrayList<>(rejected));
+ assertEquals(2, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void transformingBatchAdmissionAppliesTheContextToEverySourceElement(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ int admitted =
+ queue.tryPutBatch(Arrays.asList(1, 2, 3), "x", (source, suffix) -> source + suffix);
+ assertEquals(3, admitted);
+ assertEquals(Arrays.asList("1x", "2x", "3x"), consumeAll(queue));
+ assertEquals(0, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aDeclinedSourceElementIsNeitherAdmittedNorDropped(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ // Every other element declined. Returning null is the caller's own decision, so it counts
+ // against neither the admitted total nor dropped(): the caller already knows it declined.
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2, 3, 4, 5, 6),
+ "x",
+ (source, suffix) -> source % 2 == 0 ? null : source + suffix);
+ assertEquals(3, admitted);
+ assertEquals(Arrays.asList("1x", "3x", "5x"), consumeAll(queue));
+ assertEquals(0, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void decliningLeavesTheClaimedPlaceAvailableToTheRestOfTheBatch(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ // Nearly twice capacity in source elements, the even ones declined: the place claimed for a
+ // declined element has to go back, or the batch would run out of room after CAPACITY source
+ // elements rather than after CAPACITY admitted ones.
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2, 3, 4, 5, 6, 7),
+ "x",
+ (source, suffix) -> source % 2 == 0 ? null : source + suffix);
+ assertEquals(CAPACITY, admitted);
+ assertEquals(Arrays.asList("1x", "3x", "5x", "7x"), consumeAll(queue));
+ assertEquals(0, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void theShortfallIsExactWhenTheCallerKnowsWhatItMeantToAdmit(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ List asked = new ArrayList<>();
+ int intended = 6;
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2, 3, 4, 5, 6),
+ "x",
+ (source, suffix) -> {
+ asked.add(source);
+ return source + suffix;
+ });
+ assertEquals(CAPACITY, admitted);
+ // The whole point of the count: a caller that declined nothing gets its loss by subtraction.
+ assertEquals(2, intended - admitted);
+ // And the producer was only ever asked about elements there was already room for.
+ assertEquals(Arrays.asList(1, 2, 3, 4), asked);
+ assertEquals(2, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aSourceElementTheProducerWouldHaveDeclinedIsStillDroppedOnceFull(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ // The odd elements fill the queue exactly, so element 8 never gets a place -- even though the
+ // producer would have declined it. The place is claimed before the producer is asked, so the
+ // queue cannot know that, and counts what is true from where it stands: it could not ask.
+ // This is why dropped() is approximate for a declining producer and the shortfall is not.
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8),
+ "x",
+ (source, suffix) -> source % 2 == 0 ? null : source + suffix);
+ assertEquals(CAPACITY, admitted);
+ assertEquals(Arrays.asList("1x", "3x", "5x", "7x"), consumeAll(queue));
+ assertEquals(1, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void transformingBatchAdmissionAdmitsNothingOnceClosed(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.close();
+ AtomicBoolean asked = new AtomicBoolean();
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2),
+ "x",
+ (source, suffix) -> {
+ asked.set(true);
+ return source + suffix;
+ });
+ assertEquals(0, admitted);
+ assertFalse(asked.get(), "a closed queue must not ask the producer for anything");
+ assertEquals(2, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aRejectHandlerSeesEverySourceElementThatCouldNotBeAdmitted(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ List rejected = new ArrayList<>();
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2, 3, 4, 5, 6),
+ "x",
+ (source, suffix) -> source + suffix,
+ rejected::add);
+ assertEquals(CAPACITY, admitted);
+ assertEquals(Arrays.asList(5, 6), rejected);
+ assertEquals(2, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aRejectHandlerDoesNotSeeElementsTheProducerDeclined(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ List rejected = new ArrayList<>();
+ // Six source elements, three declined, three admitted -- the queue never fills, so nothing was
+ // refused and the handler is never called. A decline is not a rejection.
+ int admitted =
+ queue.tryPutBatch(
+ Arrays.asList(1, 2, 3, 4, 5, 6),
+ "x",
+ (source, suffix) -> source % 2 == 0 ? null : source + suffix,
+ rejected::add);
+ assertEquals(3, admitted);
+ assertTrue(rejected.isEmpty(), "a declined element is the caller's own decision");
+ assertEquals(0, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aThrowingTransformGivesBackItsPlaceAndPropagates(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ List source = Arrays.asList(1, 2, 3);
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ queue.tryPutBatch(
+ source,
+ "x",
+ (element, suffix) -> {
+ if (element == 2) {
+ throw new IllegalStateException("boom");
+ }
+ return element + suffix;
+ }));
+ // The place claimed for the failed element went back, so the queue still holds capacity for
+ // three more admissions beyond the one that succeeded.
+ assertEquals(1, queue.size());
+ assertTrue(queue.tryPutBatch("a", "b", "c").isEmpty());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void exceptionHandlerSeesTheFailureAndTheItemIsDropped(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+
+ List seen = new ArrayList<>();
+ assertTrue(
+ queue.processOrHandle(
+ item -> {
+ throw new IllegalStateException("boom");
+ },
+ (item, failure) -> seen.add(item + ":" + failure.getMessage())));
+
+ assertEquals(Arrays.asList("a:boom"), seen, "the handler is told which item died");
+ assertEquals(1, queue.dropped());
+ assertEquals(0, queue.size());
+ assertFalse(queue.process(item -> fail("nothing should be left")));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void exceptionHandlerIsNotCalledWhenTheConsumerSucceeds(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+
+ List consumed = new ArrayList<>();
+ assertTrue(
+ queue.processOrHandle(
+ consumed::add,
+ (item, failure) -> fail("handler ran for a consumer that did not throw")));
+
+ assertEquals(Arrays.asList("a"), consumed);
+ assertEquals(0, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processReportsWhetherThereWasWork(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertFalse(queue.process(item -> {}), "empty queue has no work");
+ queue.tryPut("a");
+ assertTrue(queue.process(item -> {}));
+ assertFalse(queue.process(item -> {}));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processPropagatesAConsumerFailureWhenGivenNoStrategy(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+
+ IllegalStateException thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ queue.process(
+ item -> {
+ throw new IllegalStateException("boom");
+ }),
+ "without a strategy the queue takes no view on failure");
+
+ assertEquals("boom", thrown.getMessage());
+ assertEquals(0, queue.dropped(), "a failure the caller sees is not a silent drop");
+ assertEquals(0, queue.size(), "the item was still consumed off the queue");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processReportsWorkEvenWhenTheStrategyGivesUp(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ RetryStrategy giveUp = (item, attempt, failure, retryQueue) -> false;
+ assertTrue(
+ queue.processOrRetry(
+ item -> {
+ throw new IllegalStateException("boom");
+ },
+ giveUp),
+ "the return value reports work found, not consumer success");
+ assertEquals(1, queue.dropped(), "an abandoned item is counted");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void retriesUntilTheStrategyGivesUp(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ AtomicInteger attempts = new AtomicInteger();
+ List reported = new ArrayList<>();
+
+ RetryStrategy strategy =
+ (item, attempt, failure, retryQueue) -> {
+ reported.add(attempt);
+ return attempt < 2 && retryQueue.retry(item);
+ };
+
+ while (queue.processOrRetry(
+ item -> {
+ attempts.incrementAndGet();
+ throw new IllegalStateException("boom");
+ },
+ strategy)) {
+ // drain until the strategy stops resubmitting
+ }
+
+ assertEquals(2, attempts.get(), "consumed twice: original plus one retry");
+ assertEquals(Arrays.asList(1, 2), reported, "attempt counts survive re-admission");
+ assertEquals(1, queue.dropped(), "giving up loses the item");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void maxRetriesBoundsResubmission(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ AtomicInteger attempts = new AtomicInteger();
+ RetryStrategy strategy = new MaxRetries<>(3);
+
+ while (queue.processOrRetry(
+ item -> {
+ attempts.incrementAndGet();
+ throw new IllegalStateException("boom");
+ },
+ strategy)) {
+ // drain
+ }
+
+ assertEquals(3, attempts.get());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void closeStopsAdmissionButKeepsBacklog(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ queue.close();
+
+ assertTrue(queue.isClosed());
+ assertFalse(queue.tryPut("b"));
+ assertEquals(1, queue.size(), "already-admitted work survives so a consumer can finish");
+ assertEquals(Arrays.asList("a"), consumeAll(queue));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void clearDiscardsContentsButLeavesAdmissionOpen(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPutBatch("a", "b");
+ queue.clear();
+
+ assertEquals(0, queue.size());
+ assertFalse(queue.isClosed());
+ assertTrue(queue.tryPut("c"), "clear does not close");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void shutdownClosesAndDiscards(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPutBatch("a", "b");
+ queue.shutdown();
+
+ assertEquals(0, queue.size());
+ assertTrue(queue.isClosed());
+ assertFalse(queue.tryPut("c"));
+ }
+
+ @org.junit.jupiter.api.Test
+ void unboundedQueueNeverRejects() {
+ WorkQueue queue = WorkQueues.createUnboundedMpmcQueue();
+ for (int i = 0; i < 1000; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ assertEquals(1000, queue.size());
+ assertEquals(0, queue.dropped());
+ }
+
+ @org.junit.jupiter.api.Test
+ void unboundedQueueStillCloses() {
+ WorkQueue queue = WorkQueues.createUnboundedMpmcQueue();
+ queue.close();
+ assertFalse(queue.tryPut("a"));
+ }
+
+ /**
+ * Closing is a bias applied to the permit count rather than a flag beside it, so the three ways
+ * that encoding could leak are worth pinning: applying it twice, reading a size through it, and
+ * giving places back underneath it.
+ *
+ * These pin the behaviour; none of them currently catches its own implementation slip, and it
+ * is worth being straight about why. The offset is a multiple of 2^32, so {@code size()}'s cast
+ * back to {@code int} erases the bias whether or not the unbiasing is there; the offset is far
+ * enough from either threshold that neither repeated closes nor a full queue's worth of returned
+ * places can reach it. They are guards against a future change to the offset or to the width of
+ * either, which is when all three become reachable at once.
+ */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void closingTwiceSaysWhatClosingOnceSaid(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+
+ queue.close();
+ queue.close();
+ queue.close();
+
+ assertTrue(queue.isClosed(), "still closed, not closed three times over");
+ assertFalse(queue.tryPut("a"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aClosedQueueStillReportsWhatItHolds(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertTrue(queue.tryPut("a"));
+ assertTrue(queue.tryPut("b"));
+
+ queue.close();
+
+ assertEquals(2, queue.size(), "closing must not be visible as a size");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void drainingAfterCloseDoesNotReopen(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ queue.close();
+
+ // Every drained element hands a place back, so a full queue's worth of releases runs the
+ // count as far back toward the bias as it can go.
+ List drained = new ArrayList<>();
+ while (queue.process(drained::add)) {
+ // drain it dry
+ }
+
+ assertEquals(CAPACITY, drained.size(), "close does not stop consumption");
+ assertEquals(0, queue.size());
+ assertTrue(queue.isClosed(), "returned places must not climb out of the closed state");
+ assertFalse(queue.tryPut("after"));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void retryCanPartitionFailedWorkIntoSeveralItems(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("ab");
+ List consumed = new ArrayList<>();
+ RetryStrategy split =
+ (item, attempt, failure, retryQueue) -> retryQueue.retry("a", "b");
+
+ while (queue.processOrRetry(
+ item -> {
+ if (item.length() > 1) {
+ throw new IllegalStateException("too big to handle in one piece");
+ }
+ consumed.add(item);
+ },
+ split)) {
+ // drain until the pieces are through
+ }
+
+ assertEquals(Arrays.asList("a", "b"), consumed);
+ assertEquals(0, queue.dropped(), "partitioned work is not lost");
+ }
+
+ // A reservation claims capacity on every backing; only the array backing also holds position.
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void reservationClaimsCapacityUpFront(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ try (Reservation place = queue.tryReserve()) {
+ assertNotNull(place);
+ assertEquals(1, queue.size(), "the claim costs capacity before the element exists");
+ for (int i = 0; i < CAPACITY - 1; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ assertFalse(queue.tryPut("overflow"), "the claimed place is not available to anyone else");
+ place.fill("reserved");
+ }
+ assertTrue(
+ consumeAll(queue).contains("reserved"), "filling a claimed place cannot be rejected");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void abandonedReservationYieldsNothingAndGivesTheCapacityBack(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ Reservation place = queue.tryReserve();
+ assertNotNull(place);
+ place.close();
+
+ // The array backing reclaims the slot as the consumer passes over it rather than at close, so
+ // the capacity is back once the queue has been drained, not necessarily the instant it is
+ // abandoned. What both backings promise is that nothing is ever consumed for it.
+ assertTrue(consumeAll(queue).isEmpty(), "an abandoned place produces no element");
+ assertEquals(0, queue.size());
+ assertEquals(0, queue.dropped(), "abandoning a place the caller claimed is not a rejection");
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i), "the abandoned capacity is usable again");
+ }
+ assertEquals(CAPACITY, consumeAll(queue).size());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void reserveFailsWhenThereIsNoRoom(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ Reservation refused = queue.tryReserve();
+ assertFalse(refused.granted(), "a refusal is a reservation, never null");
+ assertEquals(1, queue.dropped(), "a place that could not be claimed counts like a rejection");
+
+ refused.fill("discarded");
+ refused.close();
+ assertEquals(CAPACITY, queue.size(), "filling a refusal changes nothing and does not throw");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void reserveFailsOnceClosed(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.close();
+ assertFalse(queue.tryReserve().granted());
+ }
+
+ /** The array backing claims a slot, so the element keeps the position it was reserved at. */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void reservationJoinsWhereItIsFilledRatherThanWhereItWasClaimed(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("first");
+ List consumed = new ArrayList<>();
+
+ try (Reservation place = queue.tryReserve()) {
+ assertTrue(queue.tryPut("behind"), "the rest of the queue stays open for admission");
+ assertEquals(
+ Arrays.asList("first", "behind"),
+ consumeAll(queue),
+ "a reservation holds no position, so nothing is held in front of the consumer");
+ place.fill("filled late");
+ }
+
+ consumed.addAll(consumeAll(queue));
+ assertEquals(Arrays.asList("filled late"), consumed, "the order is the fill order");
+ }
+
+ /**
+ * The hazard a position-holding reservation would have: one thread that reserves and then drains
+ * would be waiting on itself.
+ */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aThreadMayReserveAndConsume(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("waiting");
+
+ try (Reservation place = queue.tryReserve()) {
+ assertEquals(1, queue.process(10, item -> {}), "consumption is not blocked by the claim");
+ place.fill("filled");
+ }
+
+ assertEquals(Arrays.asList("filled"), consumeAll(queue));
+ }
+
+ @org.junit.jupiter.api.Test
+ void unboundedReservationAlwaysSucceeds() {
+ WorkQueue queue = WorkQueues.createUnboundedMpmcQueue();
+ for (int i = 0; i < 1000; i++) {
+ try (Reservation place = queue.tryReserve()) {
+ assertNotNull(place);
+ place.fill("e" + i);
+ }
+ }
+ assertEquals(1000, queue.size());
+ assertEquals(0, queue.dropped());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processStopsAtTheLimit(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ for (int i = 0; i < CAPACITY; i++) {
+ queue.tryPut("e" + i);
+ }
+ List consumed = new ArrayList<>();
+
+ assertEquals(2, queue.process(2, consumed::add));
+
+ assertEquals(Arrays.asList("e0", "e1"), consumed);
+ assertEquals(CAPACITY - 2, queue.size(), "the rest of the batch is still there");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processStopsWhenTheQueueRunsDry(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ queue.tryPut("b");
+ List consumed = new ArrayList<>();
+
+ assertEquals(
+ 2,
+ queue.process(100, consumed::add),
+ "a count short of the limit is how a caller learns there is no more work");
+
+ assertEquals(Arrays.asList("a", "b"), consumed);
+ assertEquals(0, queue.process(100, consumed::add), "and an empty queue drains nothing");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processConsumesNothingForAnEmptyBatch(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+
+ assertEquals(0, queue.process(0, item -> fail("nothing may be consumed")));
+
+ assertEquals(1, queue.size());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processPassesTheContextToEveryItem(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ queue.tryPut("b");
+ List consumed = new ArrayList<>();
+
+ assertEquals(2, queue.process(10, consumed, List::add));
+
+ assertEquals(Arrays.asList("a", "b"), consumed);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void processAbandonsTheRestOfTheBatchWhenTheConsumerThrows(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("a");
+ queue.tryPut("b");
+ queue.tryPut("c");
+ List consumed = new ArrayList<>();
+
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ queue.process(
+ 10,
+ item -> {
+ consumed.add(item);
+ if ("b".equals(item)) {
+ throw new IllegalStateException("boom");
+ }
+ }));
+
+ assertEquals(Arrays.asList("a", "b"), consumed, "the failing item was handed over");
+ assertEquals(1, queue.size(), "what was behind it is left for the next drain");
+ assertEquals(0, queue.dropped(), "a failure the caller sees is not a drop");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void openReservationHoldsCapacityWithoutHoldingUpTheBatch(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ queue.tryPut("first");
+ List consumed = new ArrayList<>();
+
+ try (Reservation place = queue.tryReserve()) {
+ queue.tryPut("behind");
+
+ assertEquals(2, queue.process(10, consumed::add), "the batch runs past the open claim");
+ assertEquals(Arrays.asList("first", "behind"), consumed);
+ assertEquals(1, queue.size(), "the claimed place is still spent");
+
+ place.fill("reserved");
+ }
+
+ assertEquals(1, queue.process(10, consumed::add));
+ assertEquals(Arrays.asList("first", "behind", "reserved"), consumed);
+ assertEquals(0, queue.size());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void admitsFromTwoContexts(String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+
+ assertTrue(queue.tryPut("a", "b", (first, second) -> first + second));
+
+ assertEquals(Arrays.asList("ab"), consumeAll(queue));
+ }
+
+ /** The point of the whole API, in its two-context form: a rejected element is never built. */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void doesNotInvokeTwoContextProducerWhenFull(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ for (int i = 0; i < CAPACITY; i++) {
+ queue.tryPut("e" + i);
+ }
+ AtomicBoolean produced = new AtomicBoolean();
+
+ assertFalse(
+ queue.tryPut(
+ produced,
+ "unused",
+ (flag, ignored) -> {
+ flag.set(true);
+ return "built";
+ }));
+
+ assertFalse(produced.get(), "a full queue must not build what it is going to reject");
+ assertEquals(1, queue.dropped());
+ }
+
+ // --- What a null means, one test per place it can appear. ---
+
+ /**
+ * The leak this guards against is silent and permanent: claiming a place and then throwing out of
+ * the backing would shrink capacity by one for the life of the queue, once per call.
+ */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aNullElementThrowsWithoutSpendingAPlace(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ // A null-valued variable, not a literal: a bare tryPut(null) does not compile, because it
+ // cannot tell tryPut(T) from tryPut(Producer). Real callers reach this path through a field.
+ String absent = null;
+ assertThrows(NullPointerException.class, () -> queue.tryPut(absent));
+ assertEquals(0, queue.size());
+ assertEquals(0, queue.dropped(), "a caller's bug is not a dropped element");
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i), "the refused call must not have cost the queue a place");
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aNullElementThrowsOutOfABatchAndAbandonsTheRest(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertThrows(
+ NullPointerException.class, () -> queue.tryPutBatch(Arrays.asList("a", null, "b")));
+ assertEquals(Arrays.asList("a"), consumeAll(queue), "what came before the null is admitted");
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void fillingAReservationWithNullThrowsAndTheReservationStillReleases(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ Reservation place = queue.tryReserve();
+ assertTrue(place.granted());
+ assertThrows(NullPointerException.class, () -> place.fill(null));
+ place.close();
+ assertEquals(0, queue.size());
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i));
+ }
+ }
+
+ /**
+ * A producer declining means the same thing in the single-element forms as it does in a batch:
+ * nothing was lost, so nothing is counted. The place has to come back either way.
+ */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aProducerDecliningIsNeitherAdmittedNorDropped(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertFalse(queue.tryPut(() -> null));
+ assertFalse(queue.tryPut("ctx", ctx -> null));
+ assertFalse(queue.tryPut("one", "two", (first, second) -> null));
+ assertEquals(0, queue.size());
+ assertEquals(0, queue.dropped(), "a decline is a decision, not a loss");
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i), "every declined place must have been given back");
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aNullProducerThrowsAndGivesBackTheClaimedPlace(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertThrows(NullPointerException.class, () -> queue.tryPut("ctx", null));
+ for (int i = 0; i < CAPACITY; i++) {
+ assertTrue(queue.tryPut("e" + i), "the place claimed before the call must not be stranded");
+ }
+ }
+
+ /** A context is the caller's own value; the queue carries it and never looks at it. */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aNullContextIsCarriedThroughToTheProducer(
+ String name, IntFunction> factory) {
+ WorkQueue queue = factory.apply(CAPACITY);
+ assertTrue(queue.tryPut((String) null, ctx -> ctx == null ? "absent" : "present"));
+ assertTrue(queue.tryPut(null, null, (first, second) -> first == null ? "both" : "neither"));
+ assertEquals(
+ 1,
+ queue.tryPutBatch(
+ Arrays.asList(1), null, (source, context) -> context == null ? "null ctx" : "ctx"));
+ assertEquals(Arrays.asList("absent", "both", "null ctx"), consumeAll(queue));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("boundedQueues")
+ void aNullRejectHandlerSaysWhatOmittingItSays(
+ String name, IntFunction> factory) {
+ WorkQueue