He-Pin opened a new pull request, #3063: URL: https://github.com/apache/pekko/pull/3063
### Motivation BroadcastHub's `findAndRemoveConsumer` used `ArrayList.removeIf` which is O(k) per event with lambda allocation on every call. In high-fan-out scenarios (thousands of consumers clustered in the same wheel slot), this creates a producer backpressure bottleneck: the head can only advance after the head slot is empty, and draining a large slot requires k linear scans each of O(k) cost. Inspired by [akkadotnet/akka.net#8264](https://github.com/akkadotnet/akka.net/pull/8264) which replaced `ImmutableList<Consumer>[]` with `Dictionary<long, Consumer>[]`. ### Modification Replace `Array[java.util.ArrayList[Consumer]]` with `Array[LongMap[Consumer]]` keyed by `Consumer.id`: - **O(1) keyed removal**: `LongMap.getOrNull` + `-=` (two primitive hash lookups) replaces `ArrayList.removeIf` (O(k) linear scan with lambda allocation) - **Zero heap allocation per cycle**: `getOrNull` returns raw `V` (not `Option[V]`), avoiding `Some`/`None` allocation on the hot path - **Zero Long boxing**: `LongMap` stores primitive `long` keys in contiguous open-addressing arrays, unlike `HashMap[Long, _]` which boxes to `java.lang.Long` - **Lazy slot allocation**: empty slots are `null` (no backing map), reducing baseline memory from ~40 bytes × `bufferSize×2` to 0 - **GC-friendly release**: when a slot drains to zero consumers, the `LongMap` reference is nulled for immediate GC - **Null guards**: `Advance`/`NeedWakeup` handlers now guard against `findAndRemoveConsumer` returning null, preventing latent NPE - **Skipped empty slots**: `onUpstreamFailure` and `wakeupIdx` skip null (empty) slots during iteration ### Result | Operation | Before (`ArrayList`) | After (`LongMap`) | |-----------|---------------------|-------------------| | Remove by ID | O(k) linear scan + lambda alloc | **O(1)**, zero alloc | | Add | O(1) amortized | O(1) amortized | | Slot empty check | O(1) | O(1) (null short-circuit) | | Empty slot memory | ~40 bytes × bufferSize×2 | **0** (null) | | GC pressure per cycle | Long boxing + lambda + Entry | **0** | High-consumer lockstep scenarios (1000+ consumers) see dramatically reduced producer backpressure from wheel slot contention. No public API changes. Binary compatibility preserved. ### Tests - `sbt "stream-tests/Test/testOnly *HubSpec"` → 50 passed, 0 failed (includes 2 new high-consumer tests) - `sbt "++3.3.8; stream/compile"` → success - `sbt "stream/mimaReportBinaryIssues"` → no issues - `sbt "bench-jmh/compile"` → success - `scalafmt --mode diff-ref=origin/main` → clean - `git diff --check` → no whitespace errors JMH benchmark (`BroadcastHubBenchmark`) updated with: - `@Param` consumer counts: 64, 256, 1000, 2000 - Two benchmark methods: `broadcast` (buffer=64, max slot pressure) and `broadcastLargeBuffer` (buffer=256, moderate pressure) - Fixed buffer sizes decouple consumer count from buffer size, isolating wheel slot contention ### References Inspired by [akkadotnet/akka.net#8264](https://github.com/akkadotnet/akka.net/pull/8264). Pekko uses `scala.collection.mutable.LongMap` instead of `Dictionary`/`HashMap` for zero boxing on `Long` keys and contiguous open-addressing memory layout. -- 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]
