Skip to content

feat(broadcast): add a bounded MPMC broadcast channel - #253

Open
onenewcode wants to merge 1 commit into
apache:mainfrom
onenewcode:feat/broadcast-mpmc-bounded
Open

feat(broadcast): add a bounded MPMC broadcast channel#253
onenewcode wants to merge 1 commit into
apache:mainfrom
onenewcode:feat/broadcast-mpmc-bounded

Conversation

@onenewcode

@onenewcode onenewcode commented Aug 31, 2026

Copy link
Copy Markdown

Closes #213.

Adds broadcast::mpmc::bounded, the bounded half of the lossless MPMC broadcast pair. The unbounded half shipped in #117; this reuses its retention machinery without blurring either contract.

Contract

  • Lossless. Every accepted value stays readable by every subscription that was active when it was accepted. A receive never reports lag; nothing slides, overwrites, or is dropped to make room.
  • Capacity is the shared backlog, not a per-receiver queue. It measures what the slowest active subscription still holds, so adding subscriptions never consumes capacity — falling behind does. With no active subscription the channel retains nothing and a send never waits.
  • Backpressure. send waits until the slowest subscription consumes a retained message or is dropped; try_send reports TrySendError::Full and hands the value back untouched.
  • One commit order. tail advances and the message lands in the buffer inside the same critical section, so with several producers a later claim can never surface ahead of an earlier one, and every subscription observes the identical sequence.
  • Fresh subscriptions start at the committed tail.
  • bounded(0) panics.

Structure

The backlog, the cursors, and the one receive poll step whose waker protocol is subtle enough to be worth a single copy move into a private common module. Each family keeps its own publish path and its own Recv future, so neither retention contract hides behind a shared abstraction. Ring, cursor, retention, and sequencing types stay private to broadcast::mpmc; nothing new was promoted to crate::internal beyond enabling the existing semaphore and wait set for the broadcast feature.

Payloads are held behind an Arc so T::clone and T::drop never run under the channel lock — both are arbitrary user code that may reenter the channel. Producers waiting on capacity park on the internal semaphore, and reclaimed slots are released before the receive touches the payload, so a panicking T::clone or T::drop cannot strand a producer on capacity it already freed.

Tests

27 integration tests plus 5 Miri-sized unit tests

@onenewcode
onenewcode marked this pull request as draft August 31, 2026 02:48
@onenewcode

onenewcode commented Aug 31, 2026

Copy link
Copy Markdown
Author

Benchmark results

40 runs per configuration; the ratio is taken within each run so drift cancels, then averaged. ± is the sd of the per-run ratio.

Apple M4 (10 cores), macOS 26.5.2, rustc 1.97.0-nightly. divan medians, 4096-message batches.

concurrent, vs async-broadcast

cap prod recv async-broadcast asyncband ratio
1024 1 8 4.26 ms 3.52 ms 0.83× ±0.05
64 8 32 36.9 ms 32.8 ms 0.89× ±0.03
64 1 8 9.42 ms 9.09 ms 0.97× ±0.03
1024 1 1 383 µs 392 µs 1.02× ±0.04
64 1 1 456 µs 525 µs 1.15× ±0.20
64 8 8 10.1 ms 12.5 ms 1.24× ±0.06
1024 8 8 3.05 ms 7.72 ms 2.53× ±0.17
1024 8 32 8.18 ms 20.7 ms 2.53× ±0.15
1024 8 1 941 µs 3.29 ms 3.50× ±0.33
64 1 32 35.9 ms 131 ms 3.65× ±0.14
64 8 1 1.33 ms 5.42 ms 4.10× ±0.26
1024 1 32 13.5 ms 65.3 ms 4.85× ±0.95

Not bounded-specific. Same method on the existing unbounded comparison — same direction, and ahead of tokio::sync::broadcast at 4 and 8 producers:

producers async-broadcast asyncband tokio asyncband ÷ ab
1 112 µs 163 µs 144 µs 1.47× ±0.13
2 187 µs 283 µs 234 µs 1.52× ±0.06
4 229 µs 345 µs 364 µs 1.51× ±0.08
8 314 µs 439 µs 487 µs 1.40× ±0.07

So the cost sits in the retention machinery #117 shipped and #213 asks this PR to reuse: one Arc per message, which keeps T::clone/T::drop — reentrant user code — off the channel lock, and one lock over backlog, cursors and wait set, which makes publish-and-drain a single critical section and the park path race-free. async-broadcast stores T inline and clones under its lock, guaranteeing neither.

A rejected optimisation. try_send calls Arc::new inside the lock, after the capacity check, so a rejected send never allocates. Moving it above the lock stops contending producers queueing behind each other's allocator call — but a rejected try_send then discards that allocation, and send allocates again on the waiting path, so every blocked send pays an extra malloc/free. Both variants, 40 runs each, back to back:

shape baseline hoisted delta
cap 64, 8 prod, 1 recv 5.3 ms 4.8 ms −8.6%
cap 1024, 8 prod, 1 recv 3.3 ms 3.0 ms −6.8%
cap 64, 8 prod, 8 recv 12.4 ms 11.9 ms −3.8%
cap 1024, 1 prod, 1 recv 420 µs 480 µs +14.2%
cap 1024, 1 prod, 8 recv 3.6 ms 4.0 ms +12.9%
cap 1024, 1 prod, 32 recv 64.3 ms 75.6 ms +17.6%

Net-neutral at best, so try_send keeps the deferred allocation; the code comment records this.

Closing the two bad regimes means revisiting the Arc or sharding the lock

@onenewcode
onenewcode marked this pull request as ready for review August 31, 2026 02:59
Add `broadcast::mpmc::bounded`, a lossless bounded broadcast channel.
Every accepted value stays readable by every subscription that was
active when it was accepted, so a receive never reports lag. Capacity
counts the shared backlog held by the slowest active subscription:
`send` waits on it and `try_send` reports `TrySendError::Full` without
taking the value. Nothing slides, overwrites, or is dropped to make
room.

Move the backlog, the cursors, and the one receive poll step both
families share into a private `common` module, while each keeps its own
publish path and `Recv` future so neither retention contract hides
behind a shared abstraction. Ring, cursor, retention, and sequencing
types stay private to `broadcast::mpmc`.

Signed-off-by: onenewcode <lovestudy@qq.com>
@onenewcode
onenewcode force-pushed the feat/broadcast-mpmc-bounded branch from 1d2bbb7 to 4ffb990 Compare August 31, 2026 08:31
@tisonkun tisonkun mentioned this pull request Aug 31, 2026
35 tasks
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

feat(broadcast): add a bounded lossless MPMC channel

1 participant