shuttle-tokio: Implement the mpsc reservation APIs - #340
Merged
Conversation
`Sender::reserve` and `Sender::reserve_owned` panicked with `unimplemented!()`,
`try_reserve`/`try_reserve_owned` were missing, and `Permit`/`OwnedPermit` had
no methods at all, so there was no way to use a reservation.
The permit accounting this needs already existed: `send` acquires one permit
from `send_semaphore` before pushing a message, and the receiver releases one
back once it has popped a message. A `Permit` is just that acquire without the
push, so `reserve` is the first half of `send`, `Permit::send` is the second
half, and dropping an unused permit returns the capacity. Because the receiver
only releases capacity for messages it actually popped, there is no risk of
double-releasing.
`OwnedPermit` additionally holds the moved-in sender's slot in `known_senders`.
`reserve_owned` cannot simply move the `Arc` out of the `SenderInternal`, since
`SenderInternal::drop` would then decrement that count and could close the
channel out from under the permit; instead it claims a second slot up front and
lets the consumed sender give its own slot back, which keeps the count from
dipping to zero across the `await`. `OwnedPermit::{send, release}` pass the slot
on to the `Sender` they return, and its `Drop` gives the slot back to the
channel after returning the capacity.
`Permit` gains a lifetime parameter to match tokio's `Permit<'a, T>`. That is a
breaking change in principle, but the only way to obtain a `Permit` used to
panic, so nothing can be relying on the old shape.
Also factors the send-capacity accounting into `Channel::{acquire_capacity,
try_acquire_capacity, release_capacity}` so the rule that capacity is released
exactly once, by whoever consumed it, lives in one place.
`reserve_many`/`try_reserve_many` and `PermitIterator` are still unimplemented.
Fixes #339
sarsko
force-pushed
the
impl-mpsc-reserve
branch
from
September 5, 2026 03:06
0976c71 to
2c08e6b
Compare
Contributor
Author
|
Rebased onto Both conflicts were in code #319 rewrote, and both resolved in favour of #319's structure:
No change to the reservation implementation itself. Re-verified after the rebase: all 44 tests in |
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
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.
Fixes #339.
Sender::reserveandSender::reserve_ownedpanicked withunimplemented!(),try_reserve/try_reserve_ownedwere missing, andPermit/OwnedPermithad no methods at all, so there was no way to use a reservation under Shuttle.Approach
The permit accounting this needs already existed.
sendacquires one permit fromsend_semaphorebefore pushing a message, and the receiver releases one back once it has popped a message. APermitis exactly that acquire without the push, so:reserveis the first half ofsend, with the acquired capacity parked in thePermitPermit::sendis the second halfPermit::dropreturns the capacity — the "returning capacity when an unused permit is dropped" the issue asks forThere is no double-release risk, because the receiver only releases capacity for messages it actually popped, so capacity that never became a message is solely the permit's to return.
OwnedPermitadditionally holds the moved-in sender's slot inChannelState::known_senders.reserve_ownedcan't just move theArcout of theSenderInternal, sinceSenderInternal::dropwould then decrement that count and could close the channel out from under the permit. Instead it claims a second slot up front and lets the consumed sender give its own slot back, which keeps the count from dipping to zero across theawait.OwnedPermit::{send, release}pass the slot on to theSenderthey return, andOwnedPermit::dropreturns the capacity and then gives the slot back.Also factors the send-capacity accounting into
Channel::{acquire_capacity, try_acquire_capacity, release_capacity}, so the rule that capacity is released exactly once by whoever consumed it lives in one place rather than being spelled out at eachis_bounded()call site.API
Signatures match tokio 1.x exactly:
Sender::reserveasync fn(&self) -> Result<Permit<'_, T>, SendError<()>>Sender::try_reservefn(&self) -> Result<Permit<'_, T>, TrySendError<()>>Sender::reserve_ownedasync fn(self) -> Result<OwnedPermit<T>, SendError<()>>Sender::try_reserve_ownedfn(self) -> Result<OwnedPermit<T>, TrySendError<Self>>Permit::sendfn(self, value: T)OwnedPermit::sendfn(self, value: T) -> Sender<T>OwnedPermit::releasefn(self) -> Sender<T>OwnedPermit::same_channelfn(&self, other: &Self) -> boolOwnedPermit::same_channel_as_senderfn(&self, sender: &Sender<T>) -> boolPermitgains a lifetime parameter (Permit<T>→Permit<'a, T>) to match tokio'sPermit<'a, T>. That's a breaking change in principle, but the only way to obtain aPermitwasreserve, which panicked, so nothing can be relying on the old shape. Doing it now is what makes the type drop-in compatible;SemaphorePermit<'a>in the same crate already borrows this way.reserve_many/try_reserve_manyandPermitIteratorare still unimplemented — out of scope for #339.Tests
15 new tests in
tests/mpsc.rs, allcheck_dfsexcept the two that fan out over many senders. They cover: reserve→send delivery; capacity held while a permit lives and restored on drop; a dropped permit unblocking asendthat was waiting on a full channel (this deadlocks without theDropimpl);reserveblocking when full;try_reservefull/closed; reserve after the receiver dropped;Permit::sendafter the receiver dropped not panicking;reserve_ownedsend/release round-tripping theSender; anOwnedPermitkeeping the channel open and closing it exactly once on drop, including waking a waiting receiver;try_reserve_ownedhanding theSenderback on failure; reservations and plainsends contending for the same capacity; andsame_channel/same_channel_as_sender.Verified locally: full
tests/mpsc.rssuite (40 tests) passes,cargo clippy --all-targets -- -D clippy::allclean,cargo fmt --checkclean,cargo doc --no-depsadds no new warnings, andshuttle-tokiobuilds with--features shuttle,full.