Skip to content

feat(broadcast): add unbounded MPMC channel - #117

Merged
tisonkun merged 8 commits into
apache:mainfrom
orthur2:feat/broadcast-unbounded
Aug 27, 2026
Merged

feat(broadcast): add unbounded MPMC channel#117
tisonkun merged 8 commits into
apache:mainfrom
orthur2:feat/broadcast-unbounded

Conversation

@orthur2

@orthur2 orthur2 commented Jun 6, 2026

Copy link
Copy Markdown
Contributor

Summary

  • add broadcast::mpmc::unbounded, a lossless broadcast channel that retains each message until every active receiver has consumed it or has been dropped
  • expose cloneable UnboundedSender handles for concurrent producers and non-cloneable UnboundedReceiver subscriptions with independent cursors
  • keep send synchronous and receive errors free of Lagged semantics
  • start new subscriptions at the committed tail through subscribe and resubscribe
  • expose UnboundedSender::buffer_len, UnboundedSender::receiver_count, UnboundedReceiver::len, and UnboundedReceiver::is_empty
  • benchmark the implementation directly and compare its common lossless hot paths with Tokio broadcast and async-broadcast

Design Notes

This branch is rebased onto current main. The public path and endpoint capabilities follow the taxonomy in #167: broadcast delivery, MPMC producer topology, and unbounded retention. Implementations live under the private channel::broadcast source family, while the public module is re-exported at asyncband::broadcast.

The buffer, receiver cursors, and receive waiters are protected by the same mutex. send appends the message and takes parked wakers in one critical section. Recv::poll likewise decides between receiving, reporting disconnection, and registering its waker under that state lock, so a receiver cannot observe an empty buffer and park after a publication is already visible.

Messages are stored as Arc<T> so cloning and dropping user values happen after the channel is unlocked. When a receive advances the slowest cursor, the reclaimed prefix transfers buffer ownership out of the critical section; a sole receiver can recover the payload without cloning through Arc::try_unwrap. The common one-message reclaim keeps its first value inline instead of allocating a temporary Vec.

The buffer retains capacity used by a steady fill-and-drain workload. Capacity grown for a one-off burst is released after a later cycle drains without needing it. Reclaiming a prefix scans the receiver arena's historical high-water mark, and the benchmark matrix keeps that cost visible.

The ecosystem comparison gives bounded peers enough capacity for the entire measured batch. It therefore compares the shared no-loss, non-blocking path; it does not treat Tokio's overwrite-and-lag policy or async-broadcast's backpressure policy as equivalent to this unbounded channel.

Validation

  • cargo x build
  • cargo x check
  • cargo x test
  • cargo x lint
  • cargo +1.86.0 check -p asyncband --no-default-features --features broadcast
  • cargo +1.86.0 check -p benchmarks --bench ecosystem
  • cargo bench -p benchmarks --bench benchmarks -- broadcast::mpmc::unbounded
  • cargo bench -p benchmarks --bench ecosystem -- broadcast::mpmc::unbounded

Refs #213. Contract baseline: #167. Related prior work: #143 and #145.

@tisonkun
tisonkun force-pushed the feat/broadcast-unbounded branch from db690ce to 40f4e6c Compare August 21, 2026 01:47
@orthur2
orthur2 marked this pull request as ready for review August 22, 2026 15:29
@orthur2
orthur2 force-pushed the feat/broadcast-unbounded branch 2 times, most recently from a116888 to 8cbea5b Compare August 22, 2026 15:34
@orthur2

orthur2 commented Aug 22, 2026

Copy link
Copy Markdown
Contributor Author

Thanks @tisonkun for introducing Arena in #143 , and for moving this branch from Slab to Arena and adapting it to the current WaitRegistration APIs. And I've pushed two follow-up commits to refine the implementation.

When you have a chance, could you take another look and let me know what you think?

@tisonkun
tisonkun force-pushed the feat/broadcast-unbounded branch from 8cbea5b to 889b891 Compare August 26, 2026 17:34
@tisonkun tisonkun changed the title feat(broadcast): add unbounded policy feat(broadcast): add unbounded MPMC channel Aug 26, 2026
@tisonkun

Copy link
Copy Markdown
Member

Benchmark note from an Apple M4 Max (rustc 1.99.0-nightly), using Divan medians and 4,096-message batches:

Workload Asyncband Tokio broadcast async-broadcast
try round trip 37–42 ns 15–19 ns 22–26 ns
1 producer, then drain 186–190 µs 153–157 µs 122–141 µs
8 producers, then drain 411–451 µs 420–451 µs 289–320 µs
fanout to 1 receiver 181–187 µs 154–162 µs 117–126 µs
fanout to 8 receivers 1.39–1.43 ms 0.20–0.23 ms 0.81–0.86 ms
fanout to 32 receivers 4.44–4.74 ms 0.44–0.61 ms 2.32–2.37 ms

Tokio and async-broadcast are bounded, so the benchmark gives each enough capacity for the full batch and never exercises lag, overflow, or backpressure. This is a comparison of the common no-loss hot path, not of equivalent retention policies.

The one-message reclaim change reduced Asyncband's ready round trip from roughly 54 ns to roughly 38–42 ns and improved the single-producer batch by about 25%. Producer contention is now in the same range as Tokio by 4–8 producers and remains within roughly 1.3–1.6x of async-broadcast.

High fanout remains the visible scaling limit. Asyncband stays around 1.9x behind the closer lock-based async-broadcast peer, while Tokio's preallocated ring and per-slot bookkeeping can be 7–10x faster at 32 receivers. Closing that Tokio gap would require changing the receiver/reclamation algorithm rather than another local cleanup, so I have left that tradeoff visible for review instead of hiding it behind a larger redesign in this PR.

@tisonkun tisonkun left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

API looks good.

Leave performance improvement for further items.

@tisonkun
tisonkun merged commit cbe49ea into apache:main Aug 27, 2026
9 checks passed
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.

2 participants