He-Pin opened a new pull request, #3735: URL: https://github.com/apache/pekko/pull/3735
### Motivation After #3730 removes repeated oldest-cursor scans, `Sink.asPublisher(fanout = true)` still scans every subscription to calculate upstream demand even when outstanding requests already reserve all available buffer space, or one subscriber's demand already covers that space. This is stacked on #3730 at `9bed9a0e523da8c7356f46a5928306459584c018`. The temporary base branch `stack/pr-3730-fanout-publisher-cursor-scan` points to that exact commit so this PR shows only the additional change. Merge #3730 first, then rebase/retarget this PR to `main` before merging it. ### Modification - Skip the demand scan when outstanding upstream requests cover the available buffer capacity. - Stop scanning once a subscriber's demand reaches the full available capacity. Keep the existing request arithmetic and subscription lifecycle; no cached demand or request batching is introduced. - Add eight focused tests for outstanding requests, capacity one, finite and overflowing demand, cancellation, joining readers, and completion with retained elements. - Extend the existing publisher benchmark with request batches of 1, 16, and `Long.MaxValue`. ### Result The scan can visit zero subscriptions when capacity is already reserved, or stop at the first subscriber whose demand covers the capacity. Small requests can still require a full scan; this does not remove the worst-case repeated scans for many subscribers requesting one element at a time. Exploratory JMH results, comparing the parent commit with this change using identical benchmark sources on macOS arm64 / Temurin 17.0.21 / Scala 2.13.18. One JMH thread, two forks, 2 x 1 s warmup and 3 x 1 s measurement per fork. Throughput counts input elements delivered to every subscriber, not total subscriber deliveries. | Subscribers | Request batch | Parent elements/s | This change elements/s | Ratio | | ---: | ---: | ---: | ---: | ---: | | 1 | 1 | 2,089,613 | 1,999,753 | 0.957 | | 32 | 1 | 245,838 | 243,633 | 0.991 | | 1 | 16 | 2,291,339 | 2,289,714 | 0.999 | | 32 | 16 | 1,243,455 | 1,436,254 | 1.155 | | 1 | unbounded | 2,381,075 | 2,329,985 | 0.979 | | 32 | unbounded | 1,590,420 | 1,710,606 | 1.076 | All corresponding 99.9% confidence intervals overlap. To investigate the lower single-subscriber, batch-1 mean, a separate longer comparison reversed the execution order (this change, then parent), with three forks, 5 x 1 s warmup and 5 x 2 s measurement. It measured 1,988,046 +/- 123,156 elements/s for the parent and 1,950,412 +/- 89,989 for this change, a ratio of 0.981. These intervals also overlap. A small performance loss in that workload remains possible; these measurements do not establish a universal throughput improvement or absence of regression. ### Tests - Scala 2.13.18: `SubscriberManagementSpec` (8 tests), plus `FanoutPublisherBehaviorSpec`, `ResizableMultiReaderRingBufferSpec`, and `FlowSpec` (64 tests): passed. The new demand tests also pass against the parent implementation. - `stream-tests-tck / Test / testOnly org.apache.pekko.stream.tck.FanoutPublisherTest`: 31 passed; 7 standard untested TCK cases skipped. - Scala 3.3.9: `SubscriberManagementSpec` and `FanoutPublisherBehaviorSpec`: 22 passed. - `+bench-jmh/compile` and `+stream/mimaReportBinaryIssues`: passed on Scala 2.13.18 and 3.3.9. - `sbt headerCreateAll`, scoped header checks for `stream`, `stream-tests / Test`, and `bench-jmh`, native scalafmt, and `git diff --check`: passed. - Import sorting: `sortImports` is a root command alias, so the module-scoped invocation is unavailable. The scoped equivalent (`scalafixEnable; stream-tests/scalafixAll SortImports`) fails with the existing rule's `NoSuchMethodError: scala.meta.package$.Input()`. Incidental edits were discarded; native scalafmt, including its SortImports rewrite, passed. - Independent adversarial review and Qoder CLI stdout review: no blocker or must-fix findings. Qoder's first attempt ended with a service error and a second attachment-only attempt stalled and was terminated; the final review with full source context included directly in the prompt completed successfully. All attempts retained stdout logs. - Full local validation was intentionally not run at the requester's direction; the remaining build, style, header, documentation, and test matrix validation is delegated to remote CI. ### References Refs #3730. This PR is the second change in the fanout publisher optimization stack. -- 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]
