[He-Pin]

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:

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.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has insufficient remaining quota.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has insufficient remaining quota.

[Copilot]

🟢 Approval recommended

No active review issues are reported for the change.

No open issues 🎉

_{Review effort: Lite}

[pjfanning]

lgtm