feat(mpsc): add fair capacity reservations to bounded channels - #291
Merged
Conversation
Reserve capacity independently from the FIFO ticket. Claim a ticket only inside synchronous send, then initialize and publish the slot without holding the capacity wait queue lock. Keep publication and close ownership in a per-slot state. Cover cursor wrap, paused publishers, held permits, and payload cleanup with native and Miri tests. Restore the unmodified AtomicWaker with its attribution.
Run producer and receiver tasks on the same executor by default, and measure a receiver on the block_on caller thread as a separate workload. Use the same reusable-task harness and start protocol for usize and inline 1 KiB messages, including capacity-one and reservation cases.
Remove the Kanal adapter and dependency because its pending-operation cancellation semantics differ from the bounded MPSC contract.
This reverts commit 04d2f08.
Rounded-up slot storage overflows a power of two and the shared permit counter packs channel state into two flag bits, so capacity is now bounded at usize::MAX >> 2 with an explicit panic message, replacing opaque arithmetic or allocation failures. Zero-sized messages allocate no slots and remain limited only by the permit counter.
The permit counter, closed flag, and a may-have-waiters flag now share a single atomic state. Acquire and release are each one lock-free operation when no sender waits; releases only take the wait-queue lock to hand capacity directly to the oldest waiter, preserving registration order. A registration sets the waiting flag before its final capacity recheck so a racing release switches to the locked path and no permit is stranded without a wake.
Publishing a message previously ran the receiver waker's two read-modify-write operations even when no receiver was parked. A publishable parked flag now gates the wake: both flag accesses are SeqCst, so a skipped wake implies the receiver's post-store recheck observes the published message and no wake is lost.
The zero-sized path packs its message count and a closed flag into one atomic word, replacing the last mutex inside the bounded buffer. Close and publication stay atomic: a publication ahead of the flag is counted in the drain, and every later one observes the flag and fails.
The parked flag is unsound in the C++ memory model: slot publication is only a Release operation, so the SeqCst total order on the flag does not bind it. Miri finds a store-buffering interleaving where the producer observes the flag unset (skipping the wake) while the receiver's recheck still observes the pre-publication slot, stranding a parked receiver (concurrency::publication_racing_with_receiver_registration_cannot_lose_wakeup, concurrency::last_sender_drop_racing_with_receiver_registration_cannot_lose_wakeup). Closing the hole needs a SeqCst fence per publish, which costs as much as the wake it replaces. Keep the unconditional wake; Tokio does the same.
Divan's auto-tuning starts at one iteration per sample and stops once a sample exceeds 100x timer precision. A cold first call (lazy initialization, cache misses) can cross that threshold immediately, ending tuning at sample_size = 1. Every sample then quantizes to one timer tick (41 ns on this machine): try_round_trip reported 41.74 ns medians for Tokio and AsyncChannel, and cancel_reserved_capacity reported 40.75 ns for Asyncband, with means dominated by cold outliers. Which bench hits this varies from run to run, so pin sample_size on all nanosecond-scale benches, in line with the sizes auto-tuning converges to on healthy runs. Medians now agree with independent measurements (e.g. cancel_reserved_capacity: Asyncband 5.11 ns, Tokio 7.10 ns).
The pre-flight check duplicates what allocation already reports: collecting the slot array panics with "capacity overflow" when the layout exceeds the allocation limit, and larger requests abort like any other Rust allocation failure. Keep only the `usize::MAX >> 2` capacity bound, which protects the packed permit counter and the power-of-two rounding.
After a wake claims the WAKING bit, the state can only be WAKING: registration enters from WAITING and concurrent wakes keep the bit set, so the restoring swap always reads the value it replaces. A Release store publishes the emptied slot the same way and drops one read-modify-write from every message publication.
The packed queue length moves out of the slot ring into a ZeroSized storage value whose three operations own its invariants: publication fails once the closed flag is set, consumption relies on the count being a lower bound, and close transfers the remaining count to the drain. Dispatch on size_of::<T>() == 0 is const-folded, so generated code is unchanged.
WAITING and CLOSED become the two largest usize values instead of flag bits below a shifted permit counter. The usable range is no longer narrowed by the encoding, so the capacity ceiling rises from usize::MAX >> 2 to usize::MAX >> 1, now bound by the power-of-two slot rounding and the zero-sized queue's packed closed bit. Installing WAITING is a compare exchange over the exhausted counter: a permit that arrives first wins and is picked up by the registering sender's recheck, keeping the no-stranded-permit invariant. Hot-path operation counts are unchanged.
The receiver waker is written at most once per park cycle, while the permit counter and slot ticket are read-modify-written by every producer on every message. Padding the quiet word only spreads the shared allocation over more cache lines. Interleaved runs of the 8-producer sustained-capacity bench show padding nothing loses ~12% throughput, while padding only the two contended words is at parity with padding all three (slightly ahead in every round).
Split the 650-line bounded module along the channel's own structure: mod.rs wires up the channel, sender.rs and receiver.rs hold the two public endpoints, and the two internal mechanisms get one file each — semaphore.rs for capacity, buffer.rs for storage. The sender's wait machinery moves into the semaphore as Acquire, the in-flight counterpart of acquire(), so all sentinel and wait-queue interlocking lives in one place. Verb-named operations replace noun-shaped internals: Reservation becomes Acquire, and the receiver's single-attempt core try_pop becomes pull.
Give the slotted and zero-sized queues separate variants of one storage enum instead of sharing a flat struct with size checks. The zero-sized channel no longer carries a dead ticket, close flag, or slot pointer; the slotted variant is boxed so the enum does not reserve its room either way. Review experiment: the dispatch tag is nearly free, but the extra dependent load behind the box costs the single-threaded round trip. Numbers accompany the pull request discussion; drop this commit if the readability does not pay for them.
The file boundary bought nothing over a test module, and the split hid how little code the unsafe core actually has.
tisonkun
changed the title
Reuse try_send for ready sends and clone wakers before locking channel state. Remove redundant ZST lifecycle coverage while keeping the maximum-capacity boundary test.
tisonkun
marked this pull request as ready for review
September 9, 2026 10:36
tisonkun
enabled auto-merge (squash)
September 9, 2026 10:39
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Add
BoundedSender::reserve()andtry_reserve(), returning a borrowedPermitso producers can wait for capacity before constructing a message. Pending sends and reservations receive capacity in wait-queue order, and cancelling an unused reservation returns its capacity.Design Notes
VecDeque<T>storage. The bounded implementation no longer needs per-slot state, custom atomics, or unsafe storage.try_*methods do not wait for messages or capacity, but may contend briefly for the mutex.Validation:
cargo x test(501 tests),cargo x lint, and the MPSC suite under Miri with four scheduling seeds (36 passed and three ignored per seed).