You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2026/07/04 15:13:20 UTC
[PR] Batch elements across internal async stream boundaries [pekko]
He-Pin opened a new pull request, #3288:
URL: https://github.com/apache/pekko/pull/3288
### Motivation
Async stream islands currently send every element across the internal actor boundary as its own boundary event. This keeps ordering and backpressure simple, but it also creates per-element allocation and mailbox traffic.
### Modification
- Add an internal boundary subscriber marker so only Pekko-internal stream boundaries use batching.
- Accumulate elements produced by an output boundary during one interpreter run and flush them when the interpreter parks.
- Use the existing single-element event for `count == 1`, send `OnNextBatch` for `count > 1`, and keep external Reactive Streams subscribers on the existing path.
- Flush pending elements before completion and clear pending elements on cancellation or failure.
- Add a regression test for repeated async boundary batches and include a fused baseline parameter in the JMH benchmark.
### Result
Internal async boundary crossings allocate fewer boundary events and actor messages while preserving element ordering and Reactive Streams behavior.
Short JMH smoke run on JDK 25.0.2, G1, `-wi 1 -i 3 -w 5s -r 5s -f 1 -prof gc`:
| async boundaries | before ops/s | after ops/s | throughput delta | before B/op | after B/op | allocation delta |
| --- | ---: | ---: | ---: | ---: | ---: | ---: |
| 1 | 1,681,800 | 4,149,104 | +146.7% | 194.227 | 80.141 | -58.7% |
| 3 | 1,348,579 | 1,907,827 | +41.5% | 388.307 | 160.271 | -58.7% |
| 10 | 1,166,508 | 1,537,495 | +31.8% | 1,067.599 | 440.539 | -58.7% |
The first baseline smoke run produced a failed fork for `asyncBoundaries=3`; the table uses a separate single-parameter baseline rerun with the same JMH settings for that row.
### Tests
- `sbt "stream / Compile / compile"` -- passed
- `sbt "stream-tests / Test / testOnly org.apache.pekko.stream.FusingSpec org.apache.pekko.stream.scaladsl.PublisherSinkSpec"` -- passed
- `sbt "bench-jmh / Compile / compile"` -- passed
- `sbt "bench-jmh/Jmh/run -wi 1 -i 3 -w 5s -r 5s -f 1 -p asyncBoundaries=1,3,10 -prof gc .*AsyncBoundaryThroughputBenchmark.async_boundary_throughput"` -- passed for current branch
- `git diff --check` -- passed
- `scalafmt --list --mode diff-ref=origin/main` -- passed
### References
None - internal stream boundary throughput optimization
--
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: [PR] Batch elements across internal async stream boundaries [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #3288:
URL: https://github.com/apache/pekko/pull/3288#discussion_r3523766089
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala:
##########
@@ -136,7 +154,8 @@ import org.reactivestreams.Subscription
publisher: Publisher[Any],
val internalPortName: String)
extends UpstreamBoundaryStageLogic[Any]
- with OutHandler {
+ with OutHandler
+ with BoundarySubscriber {
Review Comment:
saved an allocation
--
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: [PR] Batch elements across internal async stream boundaries [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #3288:
URL: https://github.com/apache/pekko/pull/3288
--
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