He-Pin opened a new pull request, #3730: URL: https://github.com/apache/pekko/pull/3730
### Motivation The GraphStage fanout publisher introduced in #2874 still uses `ResizableMultiReaderRingBuffer`. When N subscribers advance together, each read of the oldest element scans all N cursors, making cursor maintenance O(N²) per published element. ### Modification - Track the number of readers at the oldest buffer position and scan only when the last one advances or cancels. Discover the initial count lazily to preserve `SubscriberManagement` construction order. - Maintain the count when a reader joins; preserve buffer growth, element reclamation, backpressure and existing terminal signals. - Add a directional complexity regression, cursor lifecycle coverage, a publisher-level joining/cancellation/backpressure test, and JMH micro/end-to-end benchmarks. - Document the existing shared-buffer and late-subscriber behavior. No public signatures or dependencies change. ### Result Cursor maintenance for equally paced readers is linear. With 32 readers, the regression reports 1088 cursor reads on the original code and passes the 128-read upper bound with this change. Local JMH on macOS ARM64, Temurin 17.0.21, Scala 2.13.18; 2 forks, 3 × 1 s warmups and 5 × 1 s measurements per fork. Throughput below is input elements per second, with each element delivered to every subscriber: | Subscribers | Publisher before | Publisher after | Ratio | | --- | ---: | ---: | ---: | | 1 | 2,341,840 | 2,259,025 | 0.96× | | 2 | 2,219,048 | 2,271,793 | 1.02× | | 8 | 1,813,387 | 2,082,464 | 1.15× | | 32 | 440,255 | 1,543,406 | 3.51× | | 128 | 28,539 | 758,668 | 26.58× | The 1- and 2-subscriber confidence intervals overlap. The ring-buffer microbenchmark improves 13.95× at 32 readers and 62.15× at 128; its single-reader mean is 5% lower, with overlapping confidence intervals. These benchmarks use continuous demand (`Long.MaxValue`); they do not establish the same gains for per-element requests or slow subscribers. Demand accounting still scans subscriptions. Reproduce on both revisions with the same benchmark source: ```text sbt "bench-jmh/Jmh/run -wi 3 -w 1s -i 5 -r 1s -f 2 -t 1 -rf json -rff /tmp/fanout.json .*Fanout.*Benchmark.*" ``` ### Tests - Original implementation: new complexity regression fails as intended (1088 > 128); other 20 focused tests pass. - Updated implementation: `stream-tests / Test / testOnly org.apache.pekko.stream.impl.ResizableMultiReaderRingBufferSpec org.apache.pekko.stream.impl.FanoutPublisherBehaviorSpec` — 22 passed on JDK 17. - Before/after JMH runs — passed, including exact subscriber element-count checks. - Expanded Scala/Java stream regression suites (`FlowSpec`, `SinkSpec`, `PublisherSinkSpec`, `FlowPublisherSinkSpec`, `SinkTest`, and the two focused specs) — 115 passed. - `stream-tests-tck / Test / testOnly org.apache.pekko.stream.tck.FanoutPublisherTest` — 31 passed; 7 TCK cases marked untested were skipped. - `headerCreateAll`, native scalafmt, `sortImports` in `bench-jmh`, `checkCodeStyle` (including Java formatting and `+headerCheckAll`), and `git diff --check` — passed. - `+mimaReportBinaryIssues` — passed across the repository for Scala 2.13.18 and 3.3.9. - Fresh independent adversarial review — no must-fix findings. A separate probe against the compiled ring buffer passed 9,928,224 randomized operations, including 103 integer wraparounds and 100 completion scenarios. - `docs/paradox` — passed (existing duplicate-anchor warnings in release notes). - Qoder CLI stdout review — no must-fix production-code findings. Its requested `sbt +bench-jmh/compile` check passed on Scala 2.13.18 and 3.3.9. - `validatePullRequest` — stopped locally; full validation is delegated to GitHub CI. The initial attempt encountered an unrelated macOS ARM64 limitation: `leveldbjni-all:1.8` contains an x86_64/i386 native library, causing `JournalNoCompactionSpec` to exit with `UnsatisfiedLinkError`. ### References Refs #2874. Related prior BroadcastHub exploration: #1705. -- 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]
