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. But the variation is not the interesting part. The interesting part is a failure that lives in the gap between two decisions that each look fine on their own.

The seam. A telemetry collector does two things: it reads a counter destructively, and it hands the result to a bounded queue that may refuse. The counter design is sound. The queue usage is sound. The bug is the ordering — and the ordering belongs to neither half, so nothing in a review of either half catches it. WafMetricCollector has it fourteen times:

long counter = wafInputTruncatedCounter.getAndSet(i, 0);   // state destroyed
if (counter > 0) {
  if (!rawMetricsQueue.offer(new WafInputTruncated(counter, i))) {
    return;                                                 // ...and now lost for good
  }
}

The counter is already zeroed when the offer fails, so the value is gone permanently — and this fires precisely when telemetry is backing up because the agent is unreachable, i.e. when the data is worth the most. Nothing about getAndSet or offer is wrong. Their sequence is.

What the callback does about it. Because admission claims a place before it invokes the producer, the destructive read moves inside the producer, and the producer only ever runs on a place that has already been granted. You cannot destroy state you have not yet been granted somewhere to put it. The two decisions can no longer be made separately, because the API only offers them joined:

if (!queue.tryPut(i, this, (idx, self) ->
        new WafInputTruncated(self.wafInputTruncatedCounter.getAndSet(idx, 0), idx))) {
  break;   // no place was claimed, so the counter was never read and nothing was lost
}

That is the design's central claim, and it is a correctness argument rather than a performance one. It is why the contextual and batch producer forms are the shapes the API pushes you toward, and why a plain tryPut(element) — while supported — gives up the guarantee: an element built before the call is state already committed.

Two honest limits. It is available, not enforced — capturing the read outside the lambda puts the ordering back the way it was. What changes is which version is the natural one to write, which is why the recipe belongs in the primitive's own documentation rather than being left for each adopter to rediscover from the composition. And the guarantee is about ordering, not about the drain as a whole: a caller that stops on the first refusal still leaves the rest of its counters for the next cycle. That is the correct outcome — they are untouched — but it is the batch form, which reports how many of the intended elements it admitted, that makes the shortfall visible rather than implicit.

The allocation story is the same property, seen from the other side. Building work you then throw away is the benign case of the same ordering. Client-side stats is the clearest example: 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 — again at the moment of most pressure. Reserve-before-build means a full queue costs a read and nothing else.

Everything else the permit counter buys follows from wanting one bound that means the same thing everywhere: an exact capacity over both an MPSC ring and an unbounded linked queue, and an O(1) size(). Those 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 the price of 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

There is no in-tree caller on this branch — the API is 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.
• Two behaviour fixes from a @codex review pass, both narrow: size() clamped to the capacity it reports against (it could exceed it while a claimant was backing out of an overspend), and MaxRetries(n) now performs n retries rather than n - 1. A third finding — retry transiently-empty reads in the MPMC backing — did not hold: jctools' poll spins for a pending publish and returns null only when the producer and consumer indices agree, so there is no false empty to ride out. Three of our own comments claimed otherwise and are corrected; discardAll's bounded re-read stays, for the reason that is actually true (size counts places claimed by producers that have not stored yet).
• There is no drop counter, on purpose. The queue owns no dropped(); counting and publishing a refusal stays the caller's, which is the division FlagEvaluationWriterImpl already chose by hand. It was removed rather than shipped because ten of its eleven increment sites were redundant with a return the caller already gets — tryPut's boolean, tryPutBatch's admitted count, Reservation.granted(), the RejectHandler — and nothing in the tree read it, so it was a safety net for a caller who ignores returns rather than new information. It can come back in its own PR if an adopter needs it. The one thing that went with it and should not have: an item a RetryStrategy abandons is now lost with no report to the caller. processOrRetry returns only whether there was an item, so RetryStrategy.onFailure's true/false — resubmitted versus gave up — is the only signal, and nothing reads it; with a library strategy like MaxRetries the caller never learns. That is the single loss in this class with no synchronous channel back, and it wants either a changed processOrRetry return or the counter back. Both are follow-ups.
• A producer that returns null declines: its place goes back, and no RejectHandler is told, because nothing was lost. The boolean on the single-element tryPut forms still collapses a decline and a refusal into the same false, so a drain loop that skips zero-valued counters cannot tell "nothing to send" from "no room" without the batch form. The javadoc gap that went with it is fixed — all three producer overloads now say so at the point of use — but whether that return should distinguish the two is a design question worth a reviewer's opinion.
• 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	13.90 s	13.95 s	[-1.5%; +0.7%] (no difference)
startup:insecure-bank:tracing:Agent	12.95 s	12.97 s	[-0.9%; +0.6%] (no difference)
startup:petclinic:appsec:Agent	16.91 s	16.93 s	[-1.0%; +0.7%] (no difference)
startup:petclinic:iast:Agent	16.86 s	16.99 s	[-1.5%; -0.1%] (maybe better)
startup:petclinic:profiling:Agent	16.04 s	16.82 s	[-10.1%; +0.8%] (unstable)
startup:petclinic:sca:Agent	16.88 s	16.71 s	[+0.2%; +2.0%] (maybe worse)
startup:petclinic:tracing:Agent	15.66 s	16.00 s	[-6.2%; +1.9%] (no difference)

Commit: 663192be · 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]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 95005d0740

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[dougqh]

To Claude - I'm not sure if we need that level of detail in the Javadoc. I think we can just indicate that this is backed by JCTools MPMC ring and then if someone is sufficiently interested they can consult the JCTools documentation.

[dougqh]

Done in 5fc2d2d — the heading and its three paragraphs are now one, saying only what a reader of this class needs: the ring can refuse a queue that is not full, the bound does not live in the ring so a refusal can only mean not yet, the retry is bounded so a wrong accounting cannot become a hang, and a pointer to Queues.mpmcArrayQueue for the general argument.

Nothing was lost outright: the 0.24% refusal-with-room measurement and the reason an ordinary caller cannot use the ring this way stay in Queues.mpmcArrayQueue, where the general warning belongs, and storeRetrying keeps its one-line frequency note beside the loop it justifies.

It also removed a wrong sentence for free — the trimmed text claimed the ring's poll reports a false empty, which it does not (see the Codex thread on retrieve()).

[dougqh]

I'm debating calling this createBoundedMpmcQueue. I'm curious to hear what others think.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 77144cacdf

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Preserve claimed elements while the MPMC publisher is stalled

When another ring producer is descheduled after claiming its slot but before publishing it, all 64 offers can complete before that thread runs again, causing this method to return false even though the permit counter already reserved capacity. Producer-based admission has already invoked its producer at this point, so destructively collected state can be lost despite the API's reserve-before-build guarantee; Reservation.fill is worse because it ignores the failure and silently discards an element its contract says cannot be rejected. An already-claimed MPMC store must not be abandoned merely because this short retry budget expires.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Make reservation completion safe across producer threads

If a granted reservation is handed to another producer thread, concurrent fill/fill or fill/close calls can both observe done == false because this flag is neither atomic nor synchronized. Two fills can then publish two elements against one permit, while a racing close can return the permit and still allow an element to be stored, corrupting the capacity accounting and eventually causing over-admission or unrelated store failures. The public Reservation contract does not require thread confinement, so completion must be atomic or that restriction must be made explicit and enforced.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Reject null callbacks before removing an item

When a caller passes a null consumer or failure policy to processOrRetry/processOrHandle, the null is reused as an internal dispatch sentinel rather than rejected as required by the public contract. A null policy makes a throwing consumer behave like plain process and propagate its original exception, while a null consumer produces an internal NullPointerException that a non-null retry strategy or handler can swallow or repeatedly retry; in both cases the item has already been removed. Validate the public callbacks before take() instead of letting their nullness select an internal processing mode.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Stop claiming that refused reservations count their drop

A refused reservation does not increment any counter: tryReserve() only attempts claimPlace() and returns an inert reservation when that fails. This statement therefore tells callers that silently filling a refused reservation has already recorded the loss, even though the queue exposes no such record and the broader WorkQueue documentation says refusal accounting remains the caller's responsibility. A caller relying on this contract can lose work without either observing granted() or maintaining its own drop metric, so remove the counting claim or implement the promised accounting.

Useful? React with 👍 / 👎.

[datadog-prod-us1-5]

The bounded MPMC queue can drop an element after it claims capacity and calls its producer. The 64-retry limit can end before a delayed producer publishes its ring slot.

_{Open Bits AI session}

_{🤖 Datadog Autotest · Commit 77144ca · What is Autotest? · @DataDog review to ask questions · Any feedback? Reach out in #autotest}

[datadog-prod-us1-5]

Bounded MPMC retries can drop a produced element

The queue can lose work or destructively read state without a report to the caller.

Assertion details

• Input: A bounded MPMC queue has a delayed producer and concurrent activity that causes 64 transient offer failures.
• Expected: After the queue claims capacity and calls the producer, it must store each non-null element.
• Actual: The queue returns false after 64 failed offers. It releases the permit and drops the produced element.

Was this helpful? React 👍 or 👎
_{🤖 Datadog Autotest · What is Autotest? · @DataDog review to ask questions · Any feedback? Reach out in #autotest · Open Bits AI session}