feat(broadcast): add unbounded MPMC channel - #117
Conversation
db690ce to
40f4e6c
Compare
a116888 to
8cbea5b
Compare
|
Thanks @tisonkun for introducing Arena in #143 , and for moving this branch from When you have a chance, could you take another look and let me know what you think? |
8cbea5b to
889b891
Compare
|
Benchmark note from an Apple M4 Max (
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
left a comment
There was a problem hiding this comment.
API looks good.
Leave performance improvement for further items.
Summary
broadcast::mpmc::unbounded, a lossless broadcast channel that retains each message until every active receiver has consumed it or has been droppedUnboundedSenderhandles for concurrent producers and non-cloneableUnboundedReceiversubscriptions with independent cursorssendsynchronous and receive errors free ofLaggedsemanticssubscribeandresubscribeUnboundedSender::buffer_len,UnboundedSender::receiver_count,UnboundedReceiver::len, andUnboundedReceiver::is_emptyDesign 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 privatechannel::broadcastsource family, while the public module is re-exported atasyncband::broadcast.The buffer, receiver cursors, and receive waiters are protected by the same mutex.
sendappends the message and takes parked wakers in one critical section.Recv::polllikewise 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 throughArc::try_unwrap. The common one-message reclaim keeps its first value inline instead of allocating a temporaryVec.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 buildcargo x checkcargo x testcargo x lintcargo +1.86.0 check -p asyncband --no-default-features --features broadcastcargo +1.86.0 check -p benchmarks --bench ecosystemcargo bench -p benchmarks --bench benchmarks -- broadcast::mpmc::unboundedcargo bench -p benchmarks --bench ecosystem -- broadcast::mpmc::unboundedRefs #213. Contract baseline: #167. Related prior work: #143 and #145.