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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
