Skip to content

fix: do not let gather request more upstream items than its downstream demand - #2195

Open
jnbdz wants to merge 1 commit into
smallrye:mainfrom
SiteNetSoft:multi-gather-backpressure
Open

fix: do not let gather request more upstream items than its downstream demand#2195
jnbdz wants to merge 1 commit into
smallrye:mainfrom
SiteNetSoft:multi-gather-backpressure

Conversation

@jnbdz

@jnbdz jnbdz commented Sep 7, 2026

Copy link
Copy Markdown

MultiGather.request() forwards one upstream.request(1) for every downstream request() call, in addition to the request(1) issued after each emitted item. When the downstream replenishes its demand from inside onItem (which bounded-buffer adapters such as the Vert.x bindings' ReadStreamSubscriber do every time their buffer drops below half), each of those calls adds a credit at the source that nobody consumes. With a Multi<Buffer> piped into a Vert.x HttpClientRequest the surplus reaches ~11,000 credits per 100,000 items. As soon as the downstream pauses, the source keeps serving the surplus and onItem forwards every extracted item without looking at the demand, so the paused subscriber buffers tens of thousands of items and the application runs out of memory. This is the root cause of quarkusio/quarkus#53099.

This PR makes the operator request a single upstream item at a time, and only after reserving a unit of downstream demand for it:

  • requestNextUpstreamItem() skips when a request is already in flight, otherwise takes one unit from demand (left untouched at Long.MAX_VALUE) and requests one item;
  • when an upstream item does not produce an extraction, the reserved unit carries over and the next item is requested directly;
  • unused reservations are given back to demand on upstream completion so drainRemainingElements() sees the right count.

MultiGatherTest#requestsIssuedWhileDeliveringAnItemMustNotOverRequestUpstream drives the operator with an emitter served one item at a time and a subscriber that replenishes a 16-item buffer from onNext; before the fix the operator requests ~11% more items than the downstream did and delivers ~1,100 items after the subscriber stopped requesting.

Verified with the reporter's Quarkus reproducer: with the current operator it runs out of a 1 GB heap within seconds, with this patch it streams for a minute with the source requested count equal to the item count. The MultiGatherTckTest and the full implementation suite (12,255 tests) pass.

…m demand

`MultiGather` requested one upstream item for every downstream `request()`
call, on top of the one it requested after each emitted item. When the
downstream replenishes its demand while an item is being delivered (as
bounded-buffer adapters such as the Vert.x `ReadStreamSubscriber` do), the
extra upstream requests accumulate as credits at the source. Once the
downstream pauses, the source keeps serving those credits and the operator
forwards every item without checking its demand, so a bounded subscriber
ends up buffering an unbounded number of items.

Request a single upstream item at a time, and only after reserving a unit
of downstream demand for it. A reserved unit carries over when an upstream
item does not produce an extraction, and unused reservations are given back
to the demand when the upstream completes.

Reported in quarkusio/quarkus#53099

@jponge jponge left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks.

  • demand can go negative at completion. requestNextUpstreamItem() now CAS-decrements demand even after onCompletion() has already run and drained against it, leaving demand negative and reserved permanently
    stranded. The original code called upstream.request(1L) unconditionally, a harmless no-op on a completed source, but the new path modifies demand, which is unsafe post-completion.

  • Non-atomic guard on reserved. reserved.get() > 0 and reserved.incrementAndGet() are not atomic. Two concurrent callers can both pass the guard and issue upstream.request(1L). Use reserved.compareAndSet(0L, 1L) as
    the gate instead.

Also missing: tests for the no-extraction accumulation path and for the onCompletion demand restoration.

jponge added a commit to jponge/smallrye-mutiny that referenced this pull request Sep 10, 2026
Gate upstream requests with an AtomicBoolean CAS so at most one item is in
flight at a time. Fixes unbounded credit accumulation when the downstream
replenishes demand from inside onNext (quarkusio/quarkus#53099).

Supersedes smallrye#2195.
jponge added a commit to jponge/smallrye-mutiny that referenced this pull request Sep 10, 2026
Gate upstream requests with an AtomicBoolean CAS so at most one item is in
flight at a time. Fixes unbounded credit accumulation when the downstream
replenishes demand from inside onNext (quarkusio/quarkus#53099).

Supersedes smallrye#2195.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants