You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "pjfanning (via GitHub)" <gi...@apache.org> on 2026/10/08 16:10:57 UTC
[I] MergeLatest does not backpressure upstreams and buffers without bound [pekko]
pjfanning opened a new issue, #3592:
URL: https://github.com/apache/pekko/issues/3592
`MergeLatest` (used by `Source.mergeLatest`, `Flow.mergeLatest` and `Source.combine` with `MergeLatest`) does not backpressure its upstreams when downstream is slow. It keeps pulling all inputs and buffers emitted elements without bound.
In `stream/src/main/scala/org/apache/pekko/stream/scaladsl/MergeLatest.scala` the input handler does:
```scala
override def onPush(): Unit = {
messages.update(index, grab(input))
activeStreams.add(index)
if (allMessagesReady) emit(out, buildElem(messages.asInstanceOf[Array[T]]))
tryPull(input)
}
```
`tryPull(input)` is called on every push regardless of downstream demand, and `emit` queues the element when `out` is not available. With fast upstreams and a slow (or stalled) downstream, the number of queued `emit`s grows without limit, which can lead to unbounded memory use. Every queued element is also eventually emitted, so a slow consumer receives a backlog of stale combinations rather than the latest one.
The Scaladoc and the operator page (`docs/src/main/paradox/stream/operators/Source-or-Flow/mergeLatest.md`) have no "Backpressures when" line, and the stream docs PR #3590 deliberately does not add one because the operator does not backpressure today.
Possible fixes to discuss:
- only pull an input again when downstream has demand (and the element for that input has been used), so the operator backpressures like other fan-in operators, or
- keep only the latest value per input and emit the latest combination when downstream pulls (conflate-like), which matches the operator's name and avoids a backlog.
Either way, the Scaladoc and operator page should then document the backpressure behaviour.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [I] MergeLatest does not backpressure upstreams and buffers without bound [pekko]
Posted by "dajiaohuang (via GitHub)" <gi...@apache.org>.
dajiaohuang commented on issue #3592:
URL: https://github.com/apache/pekko/issues/3592#issuecomment-6426466867
I am taking a look at a focused implementation of the proposed latest-value default with a legacy compatibility mode, including tests and documentation. I will link a PR here if the implementation is ready.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [I] MergeLatest does not backpressure upstreams and buffers without bound [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on issue #3592:
URL: https://github.com/apache/pekko/issues/3592#issuecomment-6082306357
## Pros and cons of the two proposed fixes
For context: the unbounded `emit` + unconditional `tryPull` in `onPush` has been there since the operator was first added (2018), so every Pekko release, 1.x included, has this behaviour. Some users may be relying on getting every combination without realising it.
### Option 1: real backpressure (lossless)
Don't pull an input again until the combination produced by its last push has gone downstream. If another input pushes while a combination is still waiting, leave that element in its port (not grabbed) and process it on the next downstream pull. At most one element per input is held at a time.
**Pros**
- Lossless: downstream gets the same sequence of combinations as today, only paced by demand. Users who count or audit every combination see no change.
- Behaves like the other fan-in operators (`Merge`, `Zip`). "Backpressures when downstream backpressures" is easy to document and explain.
- Slow consumers throttle the producers, so no CPU is spent producing values that are never used.
**Cons**
- At odds with the name: a slow consumer still gets stale combinations, just a bounded backlog of them instead of an unbounded one.
- One slow consumer stalls every input. Hot or timer sources (`Source.tick`, `Source.queue`, `Source.actorRef`) then drop or buffer upstream, which only moves the problem.
- Harder to implement: elements waiting per port, and completion (especially `eagerComplete`) has to handle elements that were pushed but never grabbed.
### Option 2: keep only the latest value per input (conflate-like, lossy)
Keep pulling every input and overwrite that input's stored value, setting a "changed" flag. On a downstream pull, if the flag is set and every input has a value, emit the current combination and clear the flag. Memory is O(number of inputs).
**Pros**
- Matches the name: a slow consumer always gets the newest state of all inputs. That's what the typical uses want (price × quantity as in the docs example, dashboards, merging config/state).
- With a fast downstream the output is identical to today's.
- Simple to implement and reason about.
- Upstreams are never stalled, so it works well with hot and timer sources.
- Clear to document: "Backpressures when: never (conflates, like `conflate`)".
**Cons**
- Lossy: with a slow consumer, the combinations in between are dropped. Code or tests that assert the full sequence under backpressure will see a difference. This is the main compatibility risk.
- Expensive upstreams are never throttled.
- Completion needs care: a changed combination that hasn't been emitted yet must still go out before the stage completes.
### Proposal
Option 2 as the default in 2.0.0, with a config switch so users can go back to the existing behaviour if something breaks:
```hocon
pekko.stream.materializer.merge-latest {
# "latest" (default): keep one value per input and emit the newest combination
# when downstream pulls.
# "legacy": pre-2.0 behaviour; emit every combination, buffering without bound
# when downstream is slow. Deprecated, to be removed in a future major.
mode = "latest"
}
```
- The setting would be read per materialization (in `preStart`, from the system config) rather than once globally, so it respects per-`ActorSystem` config.
- Possibly also a per-stage `Attributes` override, so one stream can be switched back without changing the whole system.
- Neither change affects binary compatibility: the `MergeLatest` constructor stays the same.
- The Scaladoc and operator page would get a "Backpressures when" line, and the 2.0 migration notes would mention the change and the switch.
If there's demand for a lossless bounded mode, Option 1 could be added later as another `mode` value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org