Add WorkQueue<T> with MPSC and linked-queue backings

[dougqh]

What Does This Do

Adds datadog.common.queue.WorkQueue<T> in a new utils/queue-utils module: a bounded handoff between producers and a consumer whose defining property is that admission claims capacity before it invokes any producer, so an element that is going to be rejected is never constructed.

Admission, in the order you should reach for it:

• tryPut(element) — for something already built.
• tryPut(producer), tryPut(context, producer), tryPut(c1, c2, producer) — the queue claims a place, then asks the producer to build. The context parameters exist so a producer can stay a non-capturing, bound-once field instead of a lambda allocated per call; the ladder stops at two arguments deliberately.
• tryPutBatch(source, context, producer) and a RejectHandler overload — the queue walks a source collection and transforms as it goes, returning how many it admitted. A producer returning null declines that source element: its place goes back and nothing is counted, so intended - admitted is the caller's exact shortfall.
• tryPutBatch(elements) / tryPutBatch(T...) — return the elements that were refused.
• tryReserve() → Reservation<T> — the escape hatch for work that will not fold into a callback. A refusal is a reservation that reports granted() == false, never null.

Consumption is synchronous, in the caller's frame: process(consumer), process(limit, consumer), context-carrying variants, and processOrRetry / processOrHandle for callers that want to say what a consumer failure means (RetryStrategy, RetryQueue, MaxRetries, ExceptionHandler). Plain process propagates; the queue never logs and takes no view on a failure it was not given one for.

Two backings behind one bound, chosen by WorkQueues.createMpscQueue / createMpmcQueue / createUnboundedMpmcQueue: jctools MpscArrayQueue, and a ConcurrentLinkedQueue for callers that need multiple consumers. Both are bounded by the same permit counter, which also makes size() O(1) rather than a list walk.

Tests: WorkQueueContractTest runs the shared contract against every backing, plus MpscWorkQueueStressTest. AdmissionBenchmark covers the reservation path.

Motivation

Queue users in the tree hand-roll the same few things directly against jctools — a capacity check, a drop counter, a drain loop — and each one answers the same questions slightly differently.

The specific cost this is aimed at is building work you then throw away. Client-side stats is the clearest case: it builds a SpanSnapshot per eligible span, with peer-tag and additional-tag arrays, and discards it when the inbox turns out to be full — precisely the moment the process is under the most pressure. Reserve-before-build inverts that, so a full queue costs a read and nothing else.

The aim is safety, and behaving better when things are already going wrong. That is what the permit counter is for: one exact bound with the same meaning over an MPSC ring and an unbounded linked queue, a drop count that moves once per lost item, and a producer that is never asked to build for a queue with no room. Those properties are worth most under backpressure, which is exactly where a hand-rolled capacity check is least likely to have been thought through. Read that way, the counter's ~5ns over the ring's own bound is a price for the guarantee rather than an overhead to be removed — which is why the cheaper options that trade the guarantee away (per-backing admission, an approximate bound from batched thread-local permits) are not pursued here.

Additional Notes

Deliberately a draft, and there is no in-tree caller on this branch — the API is currently exercised only by its own tests and benchmark. #12339 stacks the client-side-stats adoption on top as a trial, so the surface can be reviewed against real use. Reviewing the two together is more informative than reviewing this one alone.

Known gaps, in case they are load-bearing for your read:

• Contention at the boundary is measured, and it is no longer the dominant cost. ContendedAdmissionBenchmark (8 threads, gc profiler, JDK 25, loaded machine — directional) first priced a refused admission at ~960ns against ~3.4ns for the same rejection on jctools' own producer-index CAS: two read-modify-writes on one shared line, taken by every thread at the capacity boundary. claimPlace() now reads the count before spending from it, so a full or closed queue refuses with a load. Refusal is ~7.9ns against ~2.9ns, putting the permit counter at roughly 5ns over a bound the MPSC ring already enforced — an accepted cost for what the API does, not an open question. The decrement stays authoritative, so the bound is unchanged.
• Reserve-before-build now wins on both axes. refusedProducer vs refusedBuildThenOffer is ~8ns / 0 B/op against ~422ns / 32 B/op. Before the read it was the awkward result — 0 B/op but slower in ns/op, because the counter cost more than the allocation it avoided. The two arms are not one variable (no counter in one, no allocation in the other); what reversed is the ordering.
• Trust the direction more than the ratio: 960ns is too expensive for two contended RMWs on a quiet machine, so some of that baseline was this machine's other work amplifying contention. A quiet run should show a smaller multiple against a smaller before. The ~7.9ns is tight (±0.35).
• The closed flag is gone, folded into the permit count as a large negative bias — so the check disappears from all seven admission sites rather than getting cheaper, and closed and capacity can no longer be read out of step. A producer can no longer see an open flag and then claim a place close() already revoked. The count is a long because createUnboundedMpmcQueue seeds it with Integer.MAX_VALUE, which leaves an int no headroom for the bias — close() there would have silently done nothing.
• AdmissionBenchmark remains @Threads(1) and Scope.Thread, so its tryPut* rows are still unmeasured; only the reservation arms are filled in. Those answered their own question: building a fresh refusal measures 0 B/op where a shared refusal singleton costs 12, because the allocation-merged-with-a-static phi defeats escape analysis (JDK 17).
• Writing that benchmark turned up an API hazard worth a reviewer's eye: createMpscQueue's "Single Consumer" is a requirement, and a second consumer neither throws nor is rejected — the two spin in the ring's gap-wait, which presents as a hang. A JMH @Group with @GroupThreads(1) silently produced two consumers under -Pjmh.threads=8 and wedged a run for 28 minutes. The factory javadoc now states the consequence; nothing enforces it.
• Unexercised by any caller anywhere: tryPutBatch(Collection), tryPutBatch(T...), the RejectHandler overload, tryReserve / Reservation, RetryStrategy / RetryQueue / MaxRetries, processOrRetry / processOrHandle. That is a lot of surface per adopter and a reasonable thing to push back on.
• shutdown() is not atomic, and its javadoc now says so rather than claiming otherwise. It is closed = true; discardAll(); — a producer already past the closed check, or an in-flight retry lease, can still store an element after the discard, and that element then sits in a queue nothing will drain. Ordering the flag first bounds the survivors to those already in flight rather than eliminating them. Making it genuinely atomic means re-reading closed after every producer returns and before its element is stored — four admission sites, a per-admission cost, and a behaviour change that deserves its own PR and its own measurement. Correcting the doc is this PR; the guard is not.
• The permit counter's bound is exact — the queue never holds more than capacity elements and open reservations — but who gets turned away is approximate at the boundary, where claimants racing can back out together and refuse an admission while the queue is a place or two short of full.

[dougqh]

Update: two backings added

Both are package-private and reachable only through Queues factories — mpscQueue(capacity), mpmcQueue(capacity), unboundedMpmcQueue(). Shared admission/lifecycle/retry logic is in BaseQueue. 28 tests, all passing.

Reserve-first is real, not emulated

Worth recording, because it decides whether the central guarantee is achievable by wrapping JCTools rather than forking it. I decompiled fill(Supplier, int) in 4.0.6:

• MpscArrayQueue (Unsafe, Java 8 path) — casProducerIndex at bytecode 141, Supplier.get() at 180.
• MpscVarHandleArrayQueue (Java 25+ path, what Queues selects) — casProducerIndex at 128, Supplier.get() at 167.

Both CAS-claim the slot first, and both return early without calling the supplier at all when there's no capacity. So fill(supplier, 1) == 1 is exact reserve-first admission, on both paths. The stress test asserts this directly: 8 threads × 20k producing admissions against a permanently full queue invoke the producer zero times.

A consequence the ticket didn't note

The claim is published (producer index advanced) before the element is stored. A consumer that reaches that slot waits for the element to appear. So reserve-first converts "producer allocates" into "producer allocates while holding a slot the consumer may be blocked on" — a slow producer now stalls the consumer, where before it only stalled itself. Fine for SpanSnapshot; not fine for anything that blocks or does I/O in produce(). I documented this on Queue, but it's a contract the use-case cards should be checked against — OkHttpSink (APMLP-1652) builds a Request including a full buffer copy inside what would become the producer, which is the largest candidate.

Two API defects found by the compiler

1. RetryStrategy<? super T> doesn't typecheck. A strategy over a supertype S would receive a RetryQueue<S>, and the queue can't satisfy that — it only accepts T. Changed process to take an invariant RetryStrategy<T>. The variance in the ticket's converged signature isn't merely too permissive, it's uninhabitable.
2. The overload ambiguity is real, not theoretical. process(consumer, null) doesn't compile — ambiguous between process(Consumer, RetryStrategy) and process(C, BiConsumer). Currently worked around with a cast at the internal call site, but every caller passing a literal null hits it too. This strengthens the case for renaming the context-taking forms.

Deviations from the ticket, deliberate

• The retry "lease" is not honoured. The ticket says a single-item retry "cannot fail on capacity" because the consumer still owns the slot. poll() releases the slot before the consumer runs, so there is no slot left to reuse — a retry is an ordinary re-admission and can be rejected if producers refilled the queue meanwhile (counted as a drop). Honouring the ticket would need a claim-based consumer, i.e. forking the backing rather than wrapping it, or reserving a hidden retry slot. Flagging rather than quietly weakening it.
• Batch admission is element-wise, not one atomic reservation. tryPutBatch / tryPut(Collection) / RetryQueue.retry(T...) can therefore partially admit. fill(supplier, n) would make this atomic for the MPSC backing; left out of the draft.
• shutdown() is not atomic — it sets closed, then discards. A producer already past the closed check can still land an element. Narrowing the window needs backing-level support.
• put(BatchProducer) is lossless by construction, which I think is the intended reading: an element is pulled only once a slot is claimed, so whatever doesn't fit is still held by the producer. It stops early rather than blocking, and the caller checks hasNext(). No drops are counted for stopping at capacity.

On LinkedQueue

It keeps the per-element node, so it doesn't deliver the allocation win — it exists so multi-consumer or as-yet-unbounded call sites can adopt the interface first and be re-backed later. Its size counter is what makes the bound enforceable and size() O(1), which is precisely what DependencyResolverQueue (APMLP-1654) wants in place of a hand-rolled cap plus an O(n) ConcurrentLinkedQueue.size() walk on every admission.

Still open from the original description

The static-routine bypass for hot call sites — and with it the interface-vs-final-class dispatch question — is still not addressed. With two backings live behind Queue<T>, admission call sites that see both are now genuinely bimorphic.

[dd-octo-sts]

🟢 Java Benchmark SLOs — All performance SLOs passed

Suite	Status
Startup	🟢 pass

SLO thresholds are defined here based on automatically generated metrics. A warning is raised when results are within 5% of the threshold.

PR vs. master results

Scenario	Candidate	master	Δ (95% CI of mean)
startup:insecure-bank:iast:Agent	14.05 s	14.04 s	[-0.6%; +0.8%] (no difference)
startup:insecure-bank:tracing:Agent	12.94 s	12.92 s	[-0.7%; +1.0%] (no difference)
startup:petclinic:appsec:Agent	17.34 s	16.60 s	[+0.3%; +8.6%] (maybe worse)
startup:petclinic:iast:Agent	16.81 s	17.47 s	[-8.0%; +0.5%] (no difference)
startup:petclinic:profiling:Agent	17.34 s	17.29 s	[-1.1%; +1.7%] (no difference)
startup:petclinic:sca:Agent	17.39 s	16.66 s	[+0.1%; +8.7%] (maybe worse)
startup:petclinic:tracing:Agent	16.13 s	16.59 s	[-6.9%; +1.2%] (no difference)

Commit: 6fdbe965 · CI Pipeline · Benchmarking Platform UI

---

Load and DaCapo benchmarks can be triggered manually in the GitLab pipeline. Results will appear in the Benchmarking Platform UI after completion.

[dougqh]

Renamed: Queue<T> → WorkQueue<T>, factories split out

Set apart from the raw JCTools factory, and the java.util.Queue collision (open question 1) is gone — no caller has to qualify an import now.

Queue            -> WorkQueue          (+ WorkQueues factory)
BaseQueue        -> BaseWorkQueue
MpscBoundedQueue -> MpscWorkQueue
LinkedQueue      -> LinkedWorkQueue

The two factory surfaces are now separate classes, because they answer different questions for the caller:

	returns	caller's job
WorkQueues.createMpscQueue(n)	WorkQueue<T>	hand work over; backing is hidden and re-backable
Queues.mpscArrayQueue(n)	MessagePassingQueue<T>	drive the raw queue yourself

Queues keeps only the raw factories and is otherwise untouched, so its eleven existing callers — which include OkHttpSink and ClientStatsAggregator, two of the ticket's own use cases — are unaffected until they migrate deliberately.

Usage now reads:

WorkQueue<SpanSnapshot> inbox = WorkQueues.createMpscQueue(1024);
inbox.tryPut(ctx, SNAPSHOT);
inbox.process(this::publish, new MaxRetries<>(3));

28 tests still green.

Open questions, updated

1. Queue collides with java.util.Queue — resolved by the rename.
2. Overload ambiguity still stands, and is still real rather than theoretical: process(consumer, null) does not compile. Worth deciding whether the context-taking forms get distinct names.
3. RetryQueue.retry(T...) unchecked warning — unchanged. (Note it kept its name: it is the retry capability, not a WorkQueue.)
4. Module placement — settled: utils/queue-utils, alongside Queues.
5. Static-routine bypass / dispatch — still open, and now the sharper question of the two, since two backings sit behind WorkQueue<T>.

[dougqh]

Dropped BatchProducer and put() — deferred to the SCA follow-on

Removed from this PR. The reasoning, so the follow-on doesn't relitigate it:

Why it goes now. Across all five use cases on APMLP-1642 — client-side stats, PendingTrace, OkHttpSink, health metrics, DependencyResolverQueue — every admission is a single element. Batch admission had no caller, so it was a public type, a method, and a lossless-admission contract that every future backing would have to honour, carried on spec.

Why it comes back with SCA. SCA's partition-on-failure (the #11977 Reachability case) is the genuine caller. Landing it there means its real access pattern picks the shape, instead of us guessing between pull, push, and reservation.

The name was wrong too, which is what started this. BatchProducer sat in a family with Producer.produce() and ContextualProducer.produce(ctx) but had hasNext()/next() — so it read as "produces batches" while actually being pulled one element at a time, advertising the opposite of the mechanism. It was also structurally java.util.Iterator<T> with no remove(), which has had a throwing default since Java 8.

Agreed shape for when it returns

Not a callback taking the WorkQueue itself. Handing out the full interface exposes close(), shutdown(), clear() and process() to arbitrary caller code, and it lets the filler stash the reference past the call. It also inverts loop ownership: the filler learns about capacity only from tryPut returning false, by which point the eager tryPut(T) form has already built the element — reintroducing build-then-drop inside the one method meant to prevent it.

Instead, a scoped admission-only capability in the manner of RetryQueue — obtainable only inside the call, dead on return:

interface Admission<T> {                 // no lifecycle, no consumption
  boolean tryPut(T element);
  <C> boolean tryPut(C context, ContextualProducer<? super C, ? extends T> producer);
}

Note this is close to the reserve(int) / Reservation<T> that the ticket deliberately kept private — "only genuinely needed when caller-owned work must happen between claiming capacity and filling it." SCA bisection may be exactly that case, in which case the follow-on is really the argument for making a bounded, scoped form of reservation public. Worth deciding there with the use case in hand.

24 tests, all passing. tryPutBatch and tryPut(Collection) are still present and also have no caller today; happy to trim those too, though they cost no new type.

[dougqh]

To Claude - I think I prefer the present tense Retry to Retried

[dougqh]

For Claude - is this a location where batch claiming could work?
There would need to a limit on batch size, so it doesn't created starvation. But at least here, there's some safety since the rejects are returned to be used again.

I suppose the alternative is to registered the elements in a simple Generator, but we don't have Generator support yet.

[dougqh]

Should we estimate the reject count based on availability cap and number of elements that we're trying to submit?

[dougqh]

Same question about estimating reject count?

[dougqh]

For Claude, I tend to want to make methods final for both code cleanest and performance reasons. Is that possible here?

[dougqh]

For Claude - We should probably call tryPutBatch, too

[dougqh]

I think we should probably also have...
boolean process(Consumer, ExceptionHandler)

interface ExceptionHandler {
void handle(Throwable t);
}

I mostly just want that for the case where someone wants to log, etc without retrying.