Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(channel): user-facing rewrite of the MPSC channel primitive

Browse Source 
Lead with what a channel is and how to use it (compiling doctest for
channel()/send/recv/close), before any internal rationale. Document
every previously-undocumented public item (channel(), Sender, Receiver,
SendError, RecvError). Move the RawMutex-vs-std::sync::Mutex rationale
and lock-class discipline into an Implementation notes section. Drop
em-dashes throughout.
This commit is contained in:
Markk116
2026-07-24 08:40:26 +02:00
parent 8625ae4c35
commit feda6517e5
1 changed files with 275 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/channel.rs
+275 -205
View File
@@ -1,38 +1,102 @@
//! Unbounded MPSC channels. //! Unbounded multi-producer, single-consumer channels: how actors talk to
//! each other.
//! //!
//! Inner state is `Arc<RawMutex<Inner<T>>>` so channels can be sent across OS //! A channel is a queue with a typed [`Sender`] on one end and a typed
//! threads (required for the multi-scheduler runtime where a sender and //! [`Receiver`] on the other. Any number of actors can hold a clone of the
//! receiver may run on different scheduler threads simultaneously). //! `Sender` and push messages onto the same queue; exactly one [`Receiver`]
//! reads them back out, in the order they arrived. This is the basic wiring
//! smarm's other actor primitives (`gen_server`, `pg`, the registry) are all
//! built out of, and it is directly usable on its own for a worker that just
//! needs an inbox.
//! //!
//! ## Why `RawMutex` (Channel class), not `std::sync::Mutex` //! ## A first channel
//! //!
//! An actor holding a guard with preemption *enabled* can be timesliced //! ```
//! inside the critical section and resume on a different OS thread — the //! use smarm::{channel, run, spawn};
//! pthread mutex would then be released from a thread that didn't lock it,
//! which is UB (Linux futexes happen to tolerate it, but it's not
//! guaranteed). `RawMutex` disables preemption for the guard's span and is
//! cross-thread-release sound by construction, closing the hole. It also
//! cannot poison. Channel locks form their own [`LockClass::Channel`]
//! (raw_mutex.rs): they may be taken under a cold (Leaf) lock — finalize and
//! `monitor()` clone senders that live in slots — but nothing may be locked
//! under them, which the debug build enforces. `recv_match` runs its user
//! predicate under this lock: keep it cheap, pure, and channel-free.
//! //!
//! Semantics: //! run(|| {
//! - Senders are clonable; the last sender drop closes the channel. //! let (tx, rx) = channel::<u64>();
//! - `Receiver::recv` on an empty open channel parks the receiver. //!
//! - `Receiver::recv` on an empty closed channel returns `Err(RecvError)`. //! let worker = spawn(move || {
//! - `Sender::send` on an open channel always succeeds. //! // Blocks until a message arrives.
//! - `Sender::send` on a closed channel (receiver dropped) returns //! let n = rx.recv().unwrap();
//! `Err(SendError(value))`. //! assert_eq!(n, 42);
//! - When a send pushes to a previously empty queue and a receiver is //!
//! parked, the receiver is unparked. //! // Once every Sender is dropped, recv() reports the channel closed
//! // instead of blocking forever.
//! assert!(rx.recv().is_err());
//! });
//!
//! tx.send(42).unwrap();
//! drop(tx); // last sender gone: the channel is now closed
//! worker.join().unwrap();
//! });
//! ```
//!
//! ## Sending
//!
//! [`Sender`] is cheaply clonable: hand a clone to every actor that needs to
//! push messages into this queue. The channel stays open as long as at least
//! one clone exists; [`Sender::send`] never blocks and always succeeds while
//! the channel is open, since the queue is unbounded. Once the [`Receiver`]
//! has been dropped, `send` returns the message back to you in
//! [`SendError`] instead of delivering it.
//!
//! ## Receiving
//!
//! There is exactly one [`Receiver`] per channel (it is not clonable).
//! [`Receiver::recv`] returns the next message in arrival order, parking the
//! calling actor if the queue is currently empty. Once every `Sender` has
//! been dropped and the queue has been drained, `recv` stops parking and
//! returns [`RecvError`] instead, so a receiver never blocks forever waiting
//! on senders that are never coming back.
//!
//! Beyond plain `recv`, three variants cover the common needs:
//!
//! - [`Receiver::try_recv`]: never parks: reports an empty-but-open channel
//! as `Ok(None)` instead of waiting.
//! - [`Receiver::recv_timeout`]: parks, but gives up and returns
//! [`RecvTimeoutError::Timeout`] if no message arrives before a deadline.
//! - [`Receiver::recv_match`] / [`Receiver::try_recv_match`]: selective
//! receive. Instead of taking whatever is at the front of the queue, pick
//! out the first message matching a predicate, leaving the rest queued in
//! order. Handy for an actor that wants to prioritise one kind of message
//! over others already waiting.
//!
//! ## Waiting on several channels: `select`
//!
//! [`select`] parks an actor across several receivers at once and reports
//! the index of the first one that is ready (has a message queued, or has
//! been closed). [`select_timeout`] adds a deadline, the way `recv_timeout`
//! does for a single channel. See their docs for the full contract,
//! including the priority-order and no-fairness guarantee.
//!
//! ## Implementation notes
//!
//! The queue and its bookkeeping live behind `Arc<RawMutex<Inner<T>>>`
//! rather than a `std::sync::Mutex`, so that a channel can be freely shared
//! and sent across the OS threads backing the multi-scheduler runtime.
//! `RawMutex` matters here for a subtler reason too: an ordinary pthread
//! mutex can be released from a different OS thread than the one that took
//! it (smarm's preemption can migrate a timesliced actor between scheduler
//! threads mid-critical-section), and doing that to a `std::sync::Mutex` is
//! undefined behavior. `RawMutex` disables preemption for the guard's short
//! lifetime instead, so the release always happens on the thread that
//! acquired it, and it has no poisoning to worry about besides. Channel
//! locks are cheap and are never held across another lock acquisition or a
//! blocking call; the predicate passed to `recv_match` runs under this lock,
//! which is why it needs to stay cheap, pure, and must not call back into
//! the same channel.
use crate::pid::Pid; use crate::pid::Pid;
use crate::raw_mutex::RawMutex; use crate::raw_mutex::RawMutex;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::sync::Arc; use std::sync::Arc;
/// Create a new channel and return its `(Sender, Receiver)` halves.
///
/// The channel is unbounded (no capacity limit) and open until every
/// `Sender` has been dropped.
pub fn channel<T>() -> (Sender<T>, Receiver<T>) { pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
let inner = Arc::new(RawMutex::new_channel(Inner { let inner = Arc::new(RawMutex::new_channel(Inner {
queue: VecDeque::new(), queue: VecDeque::new(),
@@ -45,29 +109,40 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
struct Inner<T> { struct Inner<T> {
queue: VecDeque<T>, queue: VecDeque<T>,
/// The parked receiver's `(pid, park-epoch)`. The epoch is the slot /// The parked receiver's `(pid, park-epoch)`, if one is currently
/// word's runtime-wide wait identity (see slot_state.rs): wakers call /// waiting. The epoch identifies exactly which wait this is, so a waker
/// `unpark_at(pid, epoch)`, so an entry left over from an already-woken /// left over from a wait that already ended (a losing `select` arm, a
/// wait — a `select` loser arm, a satisfied `recv_timeout`'s timer — is /// `recv_timeout` whose timer fired after it was already satisfied) is
/// inert: the wake fails the word's epoch CAS and no-ops. This replaces /// inert and does nothing when it fires.
/// the old per-channel `cur_wait`/`next_wait_seq`/`timed_out` trio: wait
/// identity now exists exactly once, in the slot word.
parked_receiver: Option<(Pid, u32)>, parked_receiver: Option<(Pid, u32)>,
senders: usize, senders: usize,
receiver_alive: bool, receiver_alive: bool,
} }
/// The sending half of a channel, created by [`channel`]. Clonable: every
/// clone pushes onto the same queue, and the channel stays open as long as
/// any clone is alive. Dropping the last `Sender` closes the channel, which
/// wakes a parked [`Receiver`] so it can observe the closure.
pub struct Sender<T> { pub struct Sender<T> {
inner: Arc<RawMutex<Inner<T>>>, inner: Arc<RawMutex<Inner<T>>>,
} }
/// The receiving half of a channel, created by [`channel`]. Not clonable:
/// a channel has exactly one receiver. Reads messages in the order they
/// were sent, via [`recv`](Receiver::recv) and its variants.
pub struct Receiver<T> { pub struct Receiver<T> {
inner: Arc<RawMutex<Inner<T>>>, inner: Arc<RawMutex<Inner<T>>>,
} }
/// Returned by [`Sender::send`] when the channel's [`Receiver`] has already
/// been dropped. Carries the message back so it is never silently lost;
/// recover it with `.0` or by matching.
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
pub struct SendError<T>(pub T); pub struct SendError<T>(pub T);
/// Returned by [`Receiver::recv`] (and the other receive methods, in their
/// own error types) when the channel is closed: every `Sender` has been
/// dropped and no message is left queued.
#[derive(Debug, PartialEq, Eq, Clone, Copy)] #[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub struct RecvError; pub struct RecvError;
@@ -84,8 +159,8 @@ impl std::error::Error for RecvError {}
pub enum RecvTimeoutError { pub enum RecvTimeoutError {
/// The deadline passed with no message available. /// The deadline passed with no message available.
Timeout, Timeout,
/// All senders dropped with no message available — the bounded analogue /// Every sender was dropped with no message available. The
/// of [`RecvError`]. /// timeout-aware counterpart of plain [`RecvError`].
Disconnected, Disconnected,
} }
@@ -115,8 +190,8 @@ impl<T> Drop for Sender<T> {
// Wake the parked receiver on the last sender drop regardless of // Wake the parked receiver on the last sender drop regardless of
// whether the queue is empty. A plain `recv` only ever parks on an // whether the queue is empty. A plain `recv` only ever parks on an
// empty queue (so this is unchanged for it), but a selective // empty queue (so this is unchanged for it), but a selective
// `recv_match` may be parked on a *non-empty* queue holding only // `recv_match` may be parked on a non-empty queue holding only
// non-matching messages — it must wake to observe closure and // non-matching messages. It must wake to observe closure and
// return Err rather than sleep forever. // return Err rather than sleep forever.
if g.senders == 0 { if g.senders == 0 {
g.parked_receiver.take() g.parked_receiver.take()
@@ -133,14 +208,15 @@ impl<T> Drop for Sender<T> {
impl<T> Drop for Receiver<T> { impl<T> Drop for Receiver<T> {
fn drop(&mut self) { fn drop(&mut self) {
// The only consumer is gone: queued messages can never be delivered. // The only consumer is gone: queued messages can never be delivered.
// Drop them now instead of stranding them until the last Sender goes // Drop them now instead of leaving them queued until the last Sender
// away (a registry entry under lazy prune can keep a Sender — and thus // happens to go away, which can be long after this receiver's owner
// the Arc<Inner> — alive long after the server exits). Dropping a queued // has exited if some other part of the runtime is still holding a
// Envelope::Call drops its reply_tx, waking any caller parked in `call` // clone of the Sender. Draining runs each queued message's own drop
// with ServerDown, so the documented guarantee holds on *every* teardown // glue, which matters for a gen_server call: dropping a queued call
// path, not only the all-senders-drop one. Drain under the lock, then // envelope drops its reply channel too, which wakes the caller with
// run item destructors after releasing it (a reply_tx drop reaches into // an error instead of leaving it parked forever. Drain under the
// a *different* channel's lock + the scheduler, so it must not nest). // lock, then run the drops after releasing it, since a message's
// drop glue may itself touch a different channel or the scheduler.
let drained = { let drained = {
let mut g = self.inner.lock(); let mut g = self.inner.lock();
g.receiver_alive = false; g.receiver_alive = false;
@@ -151,14 +227,17 @@ impl<T> Drop for Receiver<T> {
} }
impl<T> Sender<T> { impl<T> Sender<T> {
/// Number of messages currently queued behind this channel. Introspection /// Number of messages currently queued and not yet received. For
/// only (RFC 016 mailbox depth); takes the channel lock, so callers reach /// introspection and monitoring; takes the channel's internal lock, so
/// it under the registry Leaf (Leaf → Channel) via the erased probe in /// avoid calling it from a hot path.
/// `registry.rs`, never on a hot path.
pub(crate) fn queued_len(&self) -> usize { pub(crate) fn queued_len(&self) -> usize {
self.inner.lock().queue.len() self.inner.lock().queue.len()
} }
/// Push `value` onto the channel. Succeeds unconditionally as long as
/// the [`Receiver`] is still alive: the queue has no capacity limit, so
/// this never blocks and never fails except when the channel is closed,
/// in which case `value` comes back in [`SendError`].
pub fn send(&self, value: T) -> Result<(), SendError<T>> { pub fn send(&self, value: T) -> Result<(), SendError<T>> {
let unpark = { let unpark = {
let mut g = self.inner.lock(); let mut g = self.inner.lock();
@@ -179,6 +258,10 @@ impl<T> Sender<T> {
} }
impl<T> Receiver<T> { impl<T> Receiver<T> {
/// Block until a message is available and return it. Messages come back
/// in the order they were sent. If the queue is empty and every
/// [`Sender`] has already been dropped, returns [`RecvError`] instead of
/// blocking forever.
pub fn recv(&self) -> Result<T, RecvError> { pub fn recv(&self) -> Result<T, RecvError> {
loop { loop {
{ {
@@ -198,15 +281,15 @@ impl<T> Receiver<T> {
g.parked_receiver.is_none_or(|(p, _)| p == me), g.parked_receiver.is_none_or(|(p, _)| p == me),
"channel has more than one receiver" "channel has more than one receiver"
); );
// begin_wait is lock-free — legal under the Channel lock; // begin_wait is lock-free, so it's legal under the Channel lock;
// registering in the same critical section makes the epoch // registering in the same critical section makes the epoch
// atomic with the senders' view of the registration. // atomic with the senders' view of the registration.
g.parked_receiver = Some((me, crate::scheduler::begin_wait())); g.parked_receiver = Some((me, crate::scheduler::begin_wait()));
crate::te!(crate::trace::Event::RecvPark(me)); crate::te!(crate::trace::Event::RecvPark(me));
} }
// Release the lock before parking — the unparker will need it. // Release the lock before parking: the unparker will need it.
crate::scheduler::park_current(); crate::scheduler::park_current();
// Woken up — record it before looping to check the queue. // Woken up. Record it before looping to check the queue.
crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() { crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
Some(p) => p, Some(p) => p,
None => panic!("smarm: RecvWake outside an actor (core corrupt)"), None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
@@ -214,29 +297,18 @@ impl<T> Receiver<T> {
} }
} }
/// Bounded receive: like [`recv`](Self::recv), but gives up once /// Like [`recv`](Self::recv), but gives up and returns
/// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`]. /// [`RecvTimeoutError::Timeout`] if no message has arrived by the time
/// `timeout` elapses.
/// ///
/// Built on the same timer machinery as `Mutex::lock_timeout`: the wait /// If a message arrives at essentially the same moment the deadline
/// registers a `WaitTimeout` entry stamped with the wait's park-epoch; /// passes, the message wins: you get `Ok` rather than `Timeout`. If
/// on expiry the channel (as the /// every sender is dropped before a message arrives or the deadline
/// [`TimerTarget`](crate::timer::TimerTarget)) checks whether *this* /// passes, you get [`RecvTimeoutError::Disconnected`].
/// wait is still parked and, only then, cancels it. A wake that races
/// the deadline resolves message-first: if a message is available when
/// the receiver runs, it is delivered even if the timer had already
/// fired. A satisfied or abandoned wait leaves its timer entry to expire
/// as a no-op (registration gone; epoch consumed), per the
/// no-cancellation convention in `timer.rs`.
/// ///
/// The wake is classified from state alone — wakes are precise (the only /// `Duration::ZERO` is a valid timeout: it still gives any
/// stamped wakers of this wait are a send, the last-sender drop, and the /// already-queued message a chance to be returned, and only then
/// timer; a stop wake unwinds out of `park_current` and never reaches /// reports `Timeout`.
/// the classification), so: message queued → `Ok`; `senders == 0` →
/// `Disconnected`; neither → it was the timer → `Timeout`.
///
/// `Duration::ZERO` is a valid timeout: it parks until the immediately-
/// due timer is drained, then reports `Timeout` unless a message was
/// already queued.
pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<T, RecvTimeoutError> pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<T, RecvTimeoutError>
where where
T: Send + 'static, T: Send + 'static,
@@ -268,7 +340,7 @@ impl<T> Receiver<T> {
// Arm the timer after releasing the channel lock (insert takes the // Arm the timer after releasing the channel lock (insert takes the
// timers lock; never nest under a Channel lock). A send or even the // timers lock; never nest under a Channel lock). A send or even the
// timer itself may unpark us before we park — the RunningNotified // timer itself may unpark us before we park; the runtime's wake
// protocol makes the park below return immediately in that case. // protocol makes the park below return immediately in that case.
let deadline = crate::timer::deadline_from_now(timeout); let deadline = crate::timer::deadline_from_now(timeout);
let target: std::sync::Arc<dyn crate::timer::TimerTarget> = self.inner.clone(); let target: std::sync::Arc<dyn crate::timer::TimerTarget> = self.inner.clone();
@@ -290,16 +362,23 @@ impl<T> Receiver<T> {
Err(RecvTimeoutError::Timeout) Err(RecvTimeoutError::Timeout)
} }
/// Selective receive: remove and return the first queued message for which /// Selective receive: find and return the first queued message for
/// `pred` holds, leaving the rest in arrival order. If no queued message /// which `pred` returns `true`, leaving every other message in the
/// matches, parks and re-scans on every send (a selective receiver may park /// queue untouched and in order. Useful when an actor's inbox mixes
/// on a *non-empty* queue). Returns `Err(RecvError)` only once the channel /// message kinds and it wants to handle one kind out of turn, without
/// is closed and no queued message matches. /// discarding the rest.
/// ///
/// `pred` is run while the channel lock is held: keep it cheap and pure, /// If nothing queued matches, this blocks and re-checks every time a new
/// and do not call back into this channel from inside it. It is modelled as /// message arrives, the same way [`recv`](Self::recv) blocks on an empty
/// `Fn` (not `FnMut`) deliberately — it is re-run from scratch on every /// queue: a selective receiver can be waiting even while the queue holds
/// scan, so a stateful predicate would observe surprising re-counting. /// messages, just none that match yet. Returns [`RecvError`] only once
/// the channel is closed and still nothing matches.
///
/// `pred` runs while the channel is locked, so keep it cheap, side
/// effect free, and make sure it never calls back into this same
/// channel. It takes `&T` and is called fresh on every scan (not `FnMut`
/// with running state), so it should judge each message purely on its
/// own content.
pub fn recv_match<F>(&self, pred: F) -> Result<T, RecvError> pub fn recv_match<F>(&self, pred: F) -> Result<T, RecvError>
where where
F: Fn(&T) -> bool, F: Fn(&T) -> bool,
@@ -331,7 +410,7 @@ impl<T> Receiver<T> {
g.parked_receiver = Some((me, crate::scheduler::begin_wait())); g.parked_receiver = Some((me, crate::scheduler::begin_wait()));
crate::te!(crate::trace::Event::RecvPark(me)); crate::te!(crate::trace::Event::RecvPark(me));
} }
// Release the lock before parking — the unparker will need it. // Release the lock before parking: the unparker will need it.
crate::scheduler::park_current(); crate::scheduler::park_current();
crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() { crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
Some(p) => p, Some(p) => p,
@@ -340,10 +419,12 @@ impl<T> Receiver<T> {
} }
} }
/// Non-blocking selective receive. `Ok(Some(v))` if a queued message /// The non-blocking counterpart of [`recv_match`](Self::recv_match):
/// matched `pred` (removed, rest left in order), `Ok(None)` if the channel /// returns immediately either way. `Ok(Some(v))` if a queued message
/// is open but nothing matched, `Err(RecvError)` if closed and nothing /// matched `pred` (removed; the rest stay queued in order), `Ok(None)`
/// matched. Same predicate contract as [`recv_match`](Self::recv_match). /// if the channel is open but nothing currently matches, `Err(RecvError)`
/// if the channel is closed and nothing matches. Same predicate contract
/// as `recv_match`.
pub fn try_recv_match<F>(&self, pred: F) -> Result<Option<T>, RecvError> pub fn try_recv_match<F>(&self, pred: F) -> Result<Option<T>, RecvError>
where where
F: Fn(&T) -> bool, F: Fn(&T) -> bool,
@@ -363,8 +444,10 @@ impl<T> Receiver<T> {
Ok(None) Ok(None)
} }
/// Non-blocking. `Ok(Some(v))` if a message was available, `Ok(None)` if /// The non-blocking counterpart of [`recv`](Self::recv): returns
/// the channel is empty but open, `Err(RecvError)` if closed and drained. /// immediately either way. `Ok(Some(v))` if a message was queued,
/// `Ok(None)` if the channel is open but currently empty, `Err(RecvError)`
/// if the channel is closed and the queue is drained.
pub fn try_recv(&self) -> Result<Option<T>, RecvError> { pub fn try_recv(&self) -> Result<Option<T>, RecvError> {
let mut g = self.inner.lock(); let mut g = self.inner.lock();
if let Some(v) = g.queue.pop_front() { if let Some(v) = g.queue.pop_front() {
@@ -379,18 +462,18 @@ impl<T> Receiver<T> {
} }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// TimerTarget — the expiry half of recv_timeout // TimerTarget: the expiry half of recv_timeout
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> { impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
fn on_timeout(&self, pid: Pid, epoch: u32) { fn on_timeout(&self, pid: Pid, epoch: u32) {
// Cancel the wait only if THIS wait (epoch match) is still // Cancel the wait only if THIS wait (epoch match) is still
// registered. If a sender already took `parked_receiver`, the // registered. If a sender already took `parked_receiver`, the
// receiver is waking with a message — message wins, the timer // receiver is waking with a message: message wins, the timer
// no-ops. If a later wait by the same receiver is registered, the // no-ops. If a later wait by the same receiver is registered, the
// epoch mismatches — stale entry, no-op. (The unpark_at would fail // epoch mismatches: stale entry, no-op. (unpark_at would fail its
// its word CAS in either case anyway; checking under the lock keeps // internal check in either case anyway; checking under the lock
// the registration bookkeeping exact.) // keeps the registration bookkeeping exact.)
let unpark = { let unpark = {
let mut g = self.lock(); let mut g = self.lock();
if g.parked_receiver == Some((pid, epoch)) { if g.parked_receiver == Some((pid, epoch)) {
@@ -400,7 +483,7 @@ impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
false false
} }
}; };
// Unpark outside the channel lock — it may take the run-queue lock; // Unpark outside the channel lock: it may take the run-queue lock;
// legal under a Channel lock, but pointless to nest. // legal under a Channel lock, but pointless to nest.
if unpark { if unpark {
crate::scheduler::unpark_at(pid, epoch); crate::scheduler::unpark_at(pid, epoch);
@@ -409,7 +492,7 @@ impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
} }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// select — ready-index wait over multiple receivers // select: ready-index wait over multiple receivers
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
pub(crate) mod sealed { pub(crate) mod sealed {
@@ -417,33 +500,34 @@ pub(crate) mod sealed {
} }
impl<T> sealed::Sealed for Receiver<T> {} impl<T> sealed::Sealed for Receiver<T> {}
/// An arm of a [`select`]. Implemented by [`Receiver`]; sealed, because the /// An arm of a [`select`]: something you can wait on alongside other arms
/// registration contract below is part of the runtime's wake protocol. /// and be told when it becomes ready. Implemented by [`Receiver`]; sealed
/// (cannot be implemented outside this crate), since the registration
/// contract below is part of the runtime's internal wake protocol.
/// ///
/// Contract (all under the arm's own lock): `sel_register` checks-or- /// Contract (all under the arm's own lock): `sel_register` checks-or-
/// registers atomically — if the arm is ready it does NOT register and /// registers atomically. If the arm is ready it does not register and
/// returns `Ok(false)`; otherwise it publishes `(pid, epoch)` where its /// returns `Ok(false)`; otherwise it publishes `(pid, epoch)` where its
/// wakers will find it and returns `Ok(true)`. "Ready" means a receive /// wakers will find it and returns `Ok(true)`. "Ready" means a receive
/// would not park: a message is queued, or the arm is closed. `Err` means /// would not block: a message is queued, or the arm is closed. `Err` means
/// the arm could not register at all (only fd arms can fail; channel /// the arm could not register at all (only fd arms can fail; channel
/// registration is infallible) — the wait must be retired and earlier /// registration always succeeds), and the wait must be retired and earlier
/// eager-cleanup arms unregistered. /// eager-cleanup arms unregistered.
pub trait Selectable: sealed::Sealed { pub trait Selectable: sealed::Sealed {
#[doc(hidden)] #[doc(hidden)]
fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool>; fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result<bool>;
#[doc(hidden)] #[doc(hidden)]
fn sel_ready(&self) -> bool; fn sel_ready(&self) -> bool;
/// Remove this arm's `(pid, epoch)` registration if — and only if — it /// Remove this arm's `(pid, epoch)` registration if, and only if, it is
/// is still in place. Default no-op: a losing channel arm's stale /// still in place. Default no-op: a losing channel arm's stale
/// registration is inert (its wakers die at the epoch CAS; the next /// registration is harmless and self-cleans. Fd arms override this:
/// wait overwrites the slot). Fd arms override this: their staleness /// their staleness would otherwise leave the fd unusable for future
/// poisons the fd (waiters entry + kernel-side ONESHOT registration) /// selects, so they need an eager cleanup pass.
/// and needs an eager cleanup pass.
#[doc(hidden)] #[doc(hidden)]
fn sel_unregister(&self, _pid: Pid, _epoch: u32) {} fn sel_unregister(&self, _pid: Pid, _epoch: u32) {}
/// Whether this arm requires the eager cleanup pass at all. Gates the /// Whether this arm requires the eager cleanup pass at all. Gates the
/// post-wake `sel_unregister` sweep so channel-only selects keep /// post-wake `sel_unregister` sweep so channel-only selects keep their
/// today's zero-cancellation hot path. /// cheap, cleanup-free path.
#[doc(hidden)] #[doc(hidden)]
fn sel_eager_cleanup(&self) -> bool { fn sel_eager_cleanup(&self) -> bool {
false false
@@ -470,38 +554,35 @@ impl<T> Selectable for Receiver<T> {
} }
} }
/// Park on every arm at once; return the index of the first ready one. /// Wait on several channels at once and return the index of the first one
/// that is ready, instead of blocking on just one with [`Receiver::recv`].
/// ///
/// "Ready" means a receive on that arm would not park: a message is queued, /// "Ready" means a receive on that arm would not block: a message is
/// or the arm is **closed** (so the caller's `try_recv` observes the /// queued, or the arm is closed (so the caller's own `try_recv` observes
/// disconnect — a dead arm is an event, not a hang). The caller consumes the /// the disconnect: a dead arm is something to react to, not something to
/// arm itself, typically via [`Receiver::try_recv`]; single-receiver /// hang on). `select` only tells you which arm is ready; read the actual
/// channels guarantee nothing can steal the message in between. /// message yourself, typically with [`Receiver::try_recv`] on that arm.
/// ///
/// A closed arm stays ready *forever*: once its disconnect has been /// A closed arm stays ready forever. Once you have observed its disconnect,
/// observed, drop it from the arm set — under priority order it would /// drop it from the arm set you pass in next time: otherwise, under the
/// otherwise win every subsequent call and starve every higher-indexed arm. /// priority order below, it would win every subsequent call and starve
/// every arm listed after it.
/// ///
/// Arms are scanned **in order**: index 0 is the highest priority, both on /// Arms are checked **in order**: index 0 is the highest priority, both
/// the immediate-ready path and after a wake. This is a documented /// when checking immediately and after being woken. This is a deliberate,
/// guarantee (compose like BEAM receive clauses: put control channels /// documented guarantee, not an accident of implementation: put a control
/// first), not an accident — and therefore there is NO fairness promise; a /// or shutdown channel first so it is always noticed promptly. The
/// saturated arm 0 starves arm 1 by design. /// flip side is that there is **no fairness guarantee**: a busy arm 0 can
/// starve arm 1 indefinitely by design.
/// ///
/// One actor may select on a channel and later `recv` on it (or select on /// One actor can `select` on a channel and later plain `recv` on it (or
/// overlapping sets) freely. What stays illegal is what was always illegal: /// `select` again on an overlapping set of arms) with no restriction. What
/// two *different* actors receiving on one channel. /// stays illegal is what was always illegal for a channel: two *different*
/// actors receiving on the same one.
/// ///
/// Built on the consuming-wake protocol (see slot_state.rs): all arms are /// Panics if `arms` is empty, if called outside an actor, or if an fd arm
/// registered under one wait epoch; the winning wake consumes it, so losing /// fails to register (see [`try_select`] for the fallible form; a
/// arms' registrations are inert and need no cancellation pass — they /// channel-only `select` can never fail).
/// self-clean at their wakers' failed CAS, or get overwritten by this
/// receiver's next wait on that channel.
///
/// Panics if `arms` is empty, when called outside an actor, or if an fd
/// arm fails to register (EBADF, EMFILE, a second waiter on one fd —
/// see [`try_select`] for the fallible form; channel-only selects cannot
/// fail).
pub fn select(arms: &[&dyn Selectable]) -> usize { pub fn select(arms: &[&dyn Selectable]) -> usize {
match try_select(arms) { match try_select(arms) {
Ok(i) => i, Ok(i) => i,
@@ -509,9 +590,10 @@ pub fn select(arms: &[&dyn Selectable]) -> usize {
} }
} }
/// [`select`], fallible: `Err` when an arm fails to register (only fd /// The fallible form of [`select`]: `Err` when an arm fails to register.
/// arms can — EBADF, EMFILE on the epoll set, or a second waiter on an /// Only fd arms can fail this way (for example, the file descriptor is
/// fd that already has one). On `Err` the wait is fully retired and no /// invalid, or something else is already waiting on it); a channel-only
/// select can never fail. On `Err` the wait is fully retired and no
/// registration is left behind: every arm registered before the failing /// registration is left behind: every arm registered before the failing
/// one has been unregistered. /// one has been unregistered.
pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> { pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> {
@@ -527,13 +609,12 @@ pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> {
} }
// Stale fd registrations are not harmless (a losing fd arm's // Stale fd registrations are not harmless (a losing fd arm's
// waiters entry poisons the fd with AlreadyExists and its // leftover registration can make the fd unusable for the next
// kernel-side ONESHOT registration can fire arbitrarily late), so // select until a kernel event happens to clear it), so selects
// selects containing fd arms run an eager cleanup pass after the // containing fd arms run an eager cleanup pass after the park,
// park — including when a terminal stop unwinds out of it, via // including when a terminal stop unwinds out of it, via the guard.
// the guard. Channel-only selects skip all of it: `eager` is // Channel-only selects skip all of it: `eager` is false, the guard
// false, the guard is disarmed, and the loser-arm self-cleaning // is disarmed, and the loser-arm self-cleaning story is unchanged.
// story is unchanged.
let eager = arms.iter().any(|a| a.sel_eager_cleanup()); let eager = arms.iter().any(|a| a.sel_eager_cleanup());
let mut guard = UnregisterGuard { arms, me, epoch, armed: eager }; let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };
@@ -546,22 +627,22 @@ pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> {
drop(guard); drop(guard);
// Woken precisely: an arm's send (message) or last-sender drop // Woken precisely: an arm's send (message) or last-sender drop
// (closure) consumed our epoch, and both leave their arm ready — // (closure) is what woke us, and both leave their arm ready.
// return the first one, in priority order (which may be a // Return the first ready one, in priority order (which may be a
// different, higher-priority arm than the one that woke us; its // different, higher-priority arm than the one that woke us; its
// message stays queued and re-reports ready on the next call). // message stays queued and re-reports ready on the next call).
// Fd arms classify by a fresh zero-timeout poll, so they too are // Fd arms classify by a fresh zero-timeout poll, so they too are
// a pure function of state — independent of the registration the // a pure function of current state, independent of the
// cleanup pass just removed. // registration the cleanup pass just removed.
for (i, arm) in arms.iter().enumerate() { for (i, arm) in arms.iter().enumerate() {
if arm.sel_ready() { if arm.sel_ready() {
return Ok(i); return Ok(i);
} }
} }
// Unreachable by protocol (a stop wake unwinds out of // Unreachable in practice (a stop wake unwinds out of
// park_current). Defensive: re-open the wait and re-register — // park_current before we get here). Defensive: re-open the wait
// stale own-registrations are overwritten (channels) or were // and re-register; stale own-registrations are overwritten
// removed by the cleanup pass above (fds). // (channels) or were removed by the cleanup pass above (fds).
} }
} }
@@ -576,11 +657,11 @@ fn unregister_arms(arms: &[&dyn Selectable], me: Pid, epoch: u32) {
} }
} }
/// Stop-unwind twin of the explicit cleanup pass: a terminal stop unwinds // Stop-unwind twin of the explicit cleanup pass: a terminal stop unwinds
/// out of `park_current`, and a registered fd arm must not outlive its // out of `park_current`, and a registered fd arm must not outlive its
/// actor (the generalization of `wait_fd`'s `Dereg`). Disarmed on the // actor. Disarmed on the normal path after the explicit pass runs; never
/// normal path after the explicit pass runs; never armed when no fd arm // armed when no fd arm is registered, keeping the channel-only path
/// registered, keeping the channel-only path guard-free in effect. // guard-free in effect.
struct UnregisterGuard<'a> { struct UnregisterGuard<'a> {
arms: &'a [&'a dyn Selectable], arms: &'a [&'a dyn Selectable],
me: Pid, me: Pid,
@@ -596,20 +677,16 @@ impl Drop for UnregisterGuard<'_> {
} }
} }
/// The registration pass shared by [`select`] and [`select_timeout`]: // The registration pass shared by `select` and `select_timeout`: check-or-
/// check-or-register each arm, in priority order, each atomically under its // register each arm, in priority order, each atomically under its own lock.
/// own lock. Cross-arm atomicity is unnecessary: an arm becoming ready // Cross-arm atomicity is unnecessary: an arm becoming ready right after its
/// right after its registration wakes the caller through the protocol (the // registration still wakes the caller through the normal wake path.
/// prep-to-park window is closed by RunningNotified). //
/// // `Ok(Some(i))` = arm `i` was already ready, the pass stopped, and the wait
/// `Ok(Some(i))` = arm `i` was ready, the pass stopped, and the wait has // has been fully retired (no park may follow): earlier fd arms are
/// been RETIRED (no park may follow): earlier arms hold live-epoch // unregistered eagerly so none are left dangling. `Err` = an arm failed to
/// registrations, so earlier *fd* arms are unregistered eagerly, then the // register; same unwind (earlier fd arms unregistered, wait retired).
/// epoch is bumped, a landed notification eaten, and a pending stop // `Ok(None)` = every arm registered successfully; the caller parks.
/// re-observed — without which a stale arm wake could fault a later
/// one-shot park. `Err` = an arm failed to register; identical unwind
/// (earlier fd arms unregistered, wait retired). `Ok(None)` = every arm
/// registered; the caller parks.
fn register_arms( fn register_arms(
me: Pid, me: Pid,
epoch: u32, epoch: u32,
@@ -633,10 +710,10 @@ fn register_arms(
Ok(None) Ok(None)
} }
/// The [`select_timeout`] timer target: stateless, because precise wakes // The `select_timeout` timer target: stateless, because a wake's cause can
/// make classification a pure function of channel state. The entry is // always be read back off plain channel state (an arm ready, or not). If
/// stamped with the select's epoch; if an arm already won, this unpark dies // an arm already won before the deadline, this timer's fire is simply
/// at the word's epoch CAS (the no-cancellation convention in `timer.rs`). // ignored, the way any other stale wakeup is.
struct SelectTimeout; struct SelectTimeout;
impl crate::timer::TimerTarget for SelectTimeout { impl crate::timer::TimerTarget for SelectTimeout {
fn on_timeout(&self, pid: Pid, epoch: u32) { fn on_timeout(&self, pid: Pid, epoch: u32) {
@@ -644,27 +721,21 @@ impl crate::timer::TimerTarget for SelectTimeout {
} }
} }
/// [`select`] with a deadline: returns `Some(index)` like `select`, or /// Like [`select`], but gives up and returns `None` if no arm becomes
/// `None` once `timeout` elapses with no arm ready. /// ready before `timeout` elapses.
/// ///
/// All of `select`'s semantics carry over (priority order, closed arms /// All of `select`'s semantics carry over: arms are still checked in
/// permanently ready, no fairness promise). The timeout is one more stamped /// priority order, a closed arm is still permanently ready, and there is
/// waker on the same wait epoch — nothing is registered in any arm for it, /// still no fairness guarantee across arms. A message that arrives at
/// so there is nothing to cancel or leak: an arm winning leaves the timer /// essentially the same moment the deadline passes still wins, the same
/// entry to expire as a stale-epoch no-op; the timer winning leaves the /// way [`Receiver::recv_timeout`] resolves that race.
/// arms' registrations to self-clean exactly as a `select` loser's would.
/// ///
/// The wake is classified from state alone (wakes are precise): some arm /// `Duration::ZERO` is a valid timeout: it still gives an already-ready arm
/// ready → `Some` of the first, in priority order; none ready → the timer /// a chance to be reported before falling through to `None`.
/// was the only remaining stamped waker → `None`. A message that races the
/// deadline resolves message-first, as `recv_timeout` does.
/// ///
/// `Duration::ZERO` is a valid timeout: it parks until the immediately-due /// Panics if `arms` is empty, if called outside an actor, or if an fd arm
/// timer is drained, then reports `None` unless an arm was already ready. /// fails to register (see [`try_select_timeout`] for the fallible form; a
/// /// channel-only select can never fail).
/// Panics if `arms` is empty, when called outside an actor, or if an fd
/// arm fails to register (see [`try_select_timeout`] for the fallible
/// form; channel-only selects cannot fail).
pub fn select_timeout( pub fn select_timeout(
arms: &[&dyn Selectable], arms: &[&dyn Selectable],
timeout: std::time::Duration, timeout: std::time::Duration,
@@ -677,9 +748,9 @@ pub fn select_timeout(
} }
} }
/// [`select_timeout`], fallible: `Err` when an arm fails to register /// The fallible form of [`select_timeout`]: `Err` when an arm fails to
/// (only fd arms can). On `Err` the wait is fully retired and no /// register (only fd arms can). On `Err` the wait is fully retired and no
/// registration — arm-side or kernel-side — is left behind. /// registration is left behind on any arm.
pub fn try_select_timeout( pub fn try_select_timeout(
arms: &[&dyn Selectable], arms: &[&dyn Selectable],
timeout: std::time::Duration, timeout: std::time::Duration,
@@ -694,17 +765,16 @@ pub fn try_select_timeout(
return Ok(Some(i)); // ready now: the timer was never armed return Ok(Some(i)); // ready now: the timer was never armed
} }
// Arm the timer after the registration pass, outside every Channel // Arm the timer after the registration pass, outside every channel
// lock (insert takes the timers lock). // lock (inserting a timer takes the timers lock).
let deadline = crate::timer::deadline_from_now(timeout); let deadline = crate::timer::deadline_from_now(timeout);
let target: std::sync::Arc<dyn crate::timer::TimerTarget> = std::sync::Arc::new(SelectTimeout); let target: std::sync::Arc<dyn crate::timer::TimerTarget> = std::sync::Arc::new(SelectTimeout);
crate::scheduler::insert_wait_timer(deadline, me, target, epoch); crate::scheduler::insert_wait_timer(deadline, me, target, epoch);
// Same eager-cleanup story as `try_select`: the timer arm needs none // Same eager-cleanup story as `try_select`: a timer win in particular
// (stateless, stale entries die at the epoch CAS), channel arms need // leaves every fd arm's registration behind, which without this pass
// none, fd arms do — and a timer win in particular leaves every fd // would leave those fds unusable until a kernel event happened to
// arm's registration behind, which without this pass would poison // clear them.
// those fds until a kernel event happened to fire.
let eager = arms.iter().any(|a| a.sel_eager_cleanup()); let eager = arms.iter().any(|a| a.sel_eager_cleanup());
let mut guard = UnregisterGuard { arms, me, epoch, armed: eager }; let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };