fix: do not let gather request more upstream items than its downstream demand - #2195
fix: do not let gather request more upstream items than its downstream demand#2195jnbdz wants to merge 1 commit into
Conversation
…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
left a comment
There was a problem hiding this comment.
Thanks.
-
demandcan go negative at completion.requestNextUpstreamItem()now CAS-decrementsdemandeven afteronCompletion()has already run and drained against it, leavingdemandnegative andreservedpermanently
stranded. The original code calledupstream.request(1L)unconditionally, a harmless no-op on a completed source, but the new path modifiesdemand, which is unsafe post-completion. -
Non-atomic guard on
reserved.reserved.get() > 0andreserved.incrementAndGet()are not atomic. Two concurrent callers can both pass the guard and issueupstream.request(1L). Usereserved.compareAndSet(0L, 1L)as
the gate instead.
Also missing: tests for the no-extraction accumulation path and for the onCompletion demand restoration.
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.
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.
MultiGather.request()forwards oneupstream.request(1)for every downstreamrequest()call, in addition to therequest(1)issued after each emitted item. When the downstream replenishes its demand from insideonItem(which bounded-buffer adapters such as the Vert.x bindings'ReadStreamSubscriberdo every time their buffer drops below half), each of those calls adds a credit at the source that nobody consumes. With aMulti<Buffer>piped into a Vert.xHttpClientRequestthe surplus reaches ~11,000 credits per 100,000 items. As soon as the downstream pauses, the source keeps serving the surplus andonItemforwards 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 fromdemand(left untouched atLong.MAX_VALUE) and requests one item;demandon upstream completion sodrainRemainingElements()sees the right count.MultiGatherTest#requestsIssuedWhileDeliveringAnItemMustNotOverRequestUpstreamdrives the operator with an emitter served one item at a time and a subscriber that replenishes a 16-item buffer fromonNext; 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
MultiGatherTckTestand the fullimplementationsuite (12,255 tests) pass.