diff --git a/src/channel.rs b/src/channel.rs index d0aa0f6..6ded9cb 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -1,38 +1,102 @@ -//! Unbounded MPSC channels. +//! Unbounded multi-producer, single-consumer channels: how actors talk to +//! each other. //! -//! Inner state is `Arc>>` so channels can be sent across OS -//! threads (required for the multi-scheduler runtime where a sender and -//! receiver may run on different scheduler threads simultaneously). +//! A channel is a queue with a typed [`Sender`] on one end and a typed +//! [`Receiver`] on the other. Any number of actors can hold a clone of the +//! `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 -//! 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. +//! ``` +//! use smarm::{channel, run, spawn}; //! -//! Semantics: -//! - Senders are clonable; the last sender drop closes the channel. -//! - `Receiver::recv` on an empty open channel parks the receiver. -//! - `Receiver::recv` on an empty closed channel returns `Err(RecvError)`. -//! - `Sender::send` on an open channel always succeeds. -//! - `Sender::send` on a closed channel (receiver dropped) returns -//! `Err(SendError(value))`. -//! - When a send pushes to a previously empty queue and a receiver is -//! parked, the receiver is unparked. +//! run(|| { +//! let (tx, rx) = channel::(); +//! +//! let worker = spawn(move || { +//! // Blocks until a message arrives. +//! let n = rx.recv().unwrap(); +//! assert_eq!(n, 42); +//! +//! // 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>>` +//! 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::raw_mutex::RawMutex; use std::collections::VecDeque; 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() -> (Sender, Receiver) { let inner = Arc::new(RawMutex::new_channel(Inner { queue: VecDeque::new(), @@ -45,29 +109,40 @@ pub fn channel() -> (Sender, Receiver) { struct Inner { queue: VecDeque, - /// The parked receiver's `(pid, park-epoch)`. The epoch is the slot - /// word's runtime-wide wait identity (see slot_state.rs): wakers call - /// `unpark_at(pid, epoch)`, so an entry left over from an already-woken - /// wait — a `select` loser arm, a satisfied `recv_timeout`'s timer — is - /// inert: the wake fails the word's epoch CAS and no-ops. This replaces - /// the old per-channel `cur_wait`/`next_wait_seq`/`timed_out` trio: wait - /// identity now exists exactly once, in the slot word. + /// The parked receiver's `(pid, park-epoch)`, if one is currently + /// waiting. The epoch identifies exactly which wait this is, so a waker + /// left over from a wait that already ended (a losing `select` arm, a + /// `recv_timeout` whose timer fired after it was already satisfied) is + /// inert and does nothing when it fires. parked_receiver: Option<(Pid, u32)>, senders: usize, 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 { inner: Arc>>, } +/// 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 { inner: Arc>>, } +/// 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)] pub struct SendError(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)] pub struct RecvError; @@ -84,8 +159,8 @@ impl std::error::Error for RecvError {} pub enum RecvTimeoutError { /// The deadline passed with no message available. Timeout, - /// All senders dropped with no message available — the bounded analogue - /// of [`RecvError`]. + /// Every sender was dropped with no message available. The + /// timeout-aware counterpart of plain [`RecvError`]. Disconnected, } @@ -115,8 +190,8 @@ impl Drop for Sender { // Wake the parked receiver on the last sender drop regardless of // whether the queue is empty. A plain `recv` only ever parks on an // empty queue (so this is unchanged for it), but a selective - // `recv_match` may be parked on a *non-empty* queue holding only - // non-matching messages — it must wake to observe closure and + // `recv_match` may be parked on a non-empty queue holding only + // non-matching messages. It must wake to observe closure and // return Err rather than sleep forever. if g.senders == 0 { g.parked_receiver.take() @@ -133,14 +208,15 @@ impl Drop for Sender { impl Drop for Receiver { fn drop(&mut self) { // The only consumer is gone: queued messages can never be delivered. - // Drop them now instead of stranding them until the last Sender goes - // away (a registry entry under lazy prune can keep a Sender — and thus - // the Arc — alive long after the server exits). Dropping a queued - // Envelope::Call drops its reply_tx, waking any caller parked in `call` - // with ServerDown, so the documented guarantee holds on *every* teardown - // path, not only the all-senders-drop one. Drain under the lock, then - // run item destructors after releasing it (a reply_tx drop reaches into - // a *different* channel's lock + the scheduler, so it must not nest). + // Drop them now instead of leaving them queued until the last Sender + // happens to go away, which can be long after this receiver's owner + // has exited if some other part of the runtime is still holding a + // clone of the Sender. Draining runs each queued message's own drop + // glue, which matters for a gen_server call: dropping a queued call + // envelope drops its reply channel too, which wakes the caller with + // an error instead of leaving it parked forever. Drain under the + // 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 mut g = self.inner.lock(); g.receiver_alive = false; @@ -151,14 +227,17 @@ impl Drop for Receiver { } impl Sender { - /// Number of messages currently queued behind this channel. Introspection - /// only (RFC 016 mailbox depth); takes the channel lock, so callers reach - /// it under the registry Leaf (Leaf → Channel) via the erased probe in - /// `registry.rs`, never on a hot path. + /// Number of messages currently queued and not yet received. For + /// introspection and monitoring; takes the channel's internal lock, so + /// avoid calling it from a hot path. pub(crate) fn queued_len(&self) -> usize { 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> { let unpark = { let mut g = self.inner.lock(); @@ -179,6 +258,10 @@ impl Sender { } impl Receiver { + /// 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 { loop { { @@ -198,15 +281,15 @@ impl Receiver { g.parked_receiver.is_none_or(|(p, _)| p == me), "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 // atomic with the senders' view of the registration. g.parked_receiver = Some((me, crate::scheduler::begin_wait())); 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(); - // 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() { Some(p) => p, None => panic!("smarm: RecvWake outside an actor (core corrupt)"), @@ -214,29 +297,18 @@ impl Receiver { } } - /// Bounded receive: like [`recv`](Self::recv), but gives up once - /// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`]. + /// Like [`recv`](Self::recv), but gives up and returns + /// [`RecvTimeoutError::Timeout`] if no message has arrived by the time + /// `timeout` elapses. /// - /// Built on the same timer machinery as `Mutex::lock_timeout`: the wait - /// registers a `WaitTimeout` entry stamped with the wait's park-epoch; - /// on expiry the channel (as the - /// [`TimerTarget`](crate::timer::TimerTarget)) checks whether *this* - /// 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`. + /// If a message arrives at essentially the same moment the deadline + /// passes, the message wins: you get `Ok` rather than `Timeout`. If + /// every sender is dropped before a message arrives or the deadline + /// passes, you get [`RecvTimeoutError::Disconnected`]. /// - /// The wake is classified from state alone — wakes are precise (the only - /// stamped wakers of this wait are a send, the last-sender drop, and the - /// timer; a stop wake unwinds out of `park_current` and never reaches - /// 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. + /// `Duration::ZERO` is a valid timeout: it still gives any + /// already-queued message a chance to be returned, and only then + /// reports `Timeout`. pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result where T: Send + 'static, @@ -268,7 +340,7 @@ impl Receiver { // Arm the timer after releasing the channel lock (insert takes 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. let deadline = crate::timer::deadline_from_now(timeout); let target: std::sync::Arc = self.inner.clone(); @@ -290,16 +362,23 @@ impl Receiver { Err(RecvTimeoutError::Timeout) } - /// Selective receive: remove and return the first queued message for which - /// `pred` holds, leaving the rest in arrival order. If no queued message - /// matches, parks and re-scans on every send (a selective receiver may park - /// on a *non-empty* queue). Returns `Err(RecvError)` only once the channel - /// is closed and no queued message matches. + /// Selective receive: find and return the first queued message for + /// which `pred` returns `true`, leaving every other message in the + /// queue untouched and in order. Useful when an actor's inbox mixes + /// message kinds and it wants to handle one kind out of turn, without + /// discarding the rest. /// - /// `pred` is run while the channel lock is held: keep it cheap and pure, - /// and do not call back into this channel from inside it. It is modelled as - /// `Fn` (not `FnMut`) deliberately — it is re-run from scratch on every - /// scan, so a stateful predicate would observe surprising re-counting. + /// If nothing queued matches, this blocks and re-checks every time a new + /// message arrives, the same way [`recv`](Self::recv) blocks on an empty + /// queue: a selective receiver can be waiting even while the queue holds + /// 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(&self, pred: F) -> Result where F: Fn(&T) -> bool, @@ -331,7 +410,7 @@ impl Receiver { g.parked_receiver = Some((me, crate::scheduler::begin_wait())); 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::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() { Some(p) => p, @@ -340,10 +419,12 @@ impl Receiver { } } - /// Non-blocking selective receive. `Ok(Some(v))` if a queued message - /// matched `pred` (removed, rest left in order), `Ok(None)` if the channel - /// is open but nothing matched, `Err(RecvError)` if closed and nothing - /// matched. Same predicate contract as [`recv_match`](Self::recv_match). + /// The non-blocking counterpart of [`recv_match`](Self::recv_match): + /// returns immediately either way. `Ok(Some(v))` if a queued message + /// matched `pred` (removed; the rest stay queued in order), `Ok(None)` + /// 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(&self, pred: F) -> Result, RecvError> where F: Fn(&T) -> bool, @@ -363,8 +444,10 @@ impl Receiver { Ok(None) } - /// Non-blocking. `Ok(Some(v))` if a message was available, `Ok(None)` if - /// the channel is empty but open, `Err(RecvError)` if closed and drained. + /// The non-blocking counterpart of [`recv`](Self::recv): returns + /// 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, RecvError> { let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { @@ -379,18 +462,18 @@ impl Receiver { } // --------------------------------------------------------------------------- -// TimerTarget — the expiry half of recv_timeout +// TimerTarget: the expiry half of recv_timeout // --------------------------------------------------------------------------- impl crate::timer::TimerTarget for RawMutex> { fn on_timeout(&self, pid: Pid, epoch: u32) { // Cancel the wait only if THIS wait (epoch match) is still // 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 - // epoch mismatches — stale entry, no-op. (The unpark_at would fail - // its word CAS in either case anyway; checking under the lock keeps - // the registration bookkeeping exact.) + // epoch mismatches: stale entry, no-op. (unpark_at would fail its + // internal check in either case anyway; checking under the lock + // keeps the registration bookkeeping exact.) let unpark = { let mut g = self.lock(); if g.parked_receiver == Some((pid, epoch)) { @@ -400,7 +483,7 @@ impl crate::timer::TimerTarget for RawMutex> { 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. if unpark { crate::scheduler::unpark_at(pid, epoch); @@ -409,7 +492,7 @@ impl crate::timer::TimerTarget for RawMutex> { } // --------------------------------------------------------------------------- -// select — ready-index wait over multiple receivers +// select: ready-index wait over multiple receivers // --------------------------------------------------------------------------- pub(crate) mod sealed { @@ -417,33 +500,34 @@ pub(crate) mod sealed { } impl sealed::Sealed for Receiver {} -/// An arm of a [`select`]. Implemented by [`Receiver`]; sealed, because the -/// registration contract below is part of the runtime's wake protocol. +/// An arm of a [`select`]: something you can wait on alongside other arms +/// 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- -/// 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 /// 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 -/// 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. pub trait Selectable: sealed::Sealed { #[doc(hidden)] fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result; #[doc(hidden)] fn sel_ready(&self) -> bool; - /// Remove this arm's `(pid, epoch)` registration if — and only if — it - /// is still in place. Default no-op: a losing channel arm's stale - /// registration is inert (its wakers die at the epoch CAS; the next - /// wait overwrites the slot). Fd arms override this: their staleness - /// poisons the fd (waiters entry + kernel-side ONESHOT registration) - /// and needs an eager cleanup pass. + /// Remove this arm's `(pid, epoch)` registration if, and only if, it is + /// still in place. Default no-op: a losing channel arm's stale + /// registration is harmless and self-cleans. Fd arms override this: + /// their staleness would otherwise leave the fd unusable for future + /// selects, so they need an eager cleanup pass. #[doc(hidden)] fn sel_unregister(&self, _pid: Pid, _epoch: u32) {} /// Whether this arm requires the eager cleanup pass at all. Gates the - /// post-wake `sel_unregister` sweep so channel-only selects keep - /// today's zero-cancellation hot path. + /// post-wake `sel_unregister` sweep so channel-only selects keep their + /// cheap, cleanup-free path. #[doc(hidden)] fn sel_eager_cleanup(&self) -> bool { false @@ -470,38 +554,35 @@ impl Selectable for Receiver { } } -/// 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, -/// or the arm is **closed** (so the caller's `try_recv` observes the -/// disconnect — a dead arm is an event, not a hang). The caller consumes the -/// arm itself, typically via [`Receiver::try_recv`]; single-receiver -/// channels guarantee nothing can steal the message in between. +/// "Ready" means a receive on that arm would not block: a message is +/// queued, or the arm is closed (so the caller's own `try_recv` observes +/// the disconnect: a dead arm is something to react to, not something to +/// hang on). `select` only tells you which arm is ready; read the actual +/// message yourself, typically with [`Receiver::try_recv`] on that arm. /// -/// A closed arm stays ready *forever*: once its disconnect has been -/// observed, drop it from the arm set — under priority order it would -/// otherwise win every subsequent call and starve every higher-indexed arm. +/// A closed arm stays ready forever. Once you have observed its disconnect, +/// drop it from the arm set you pass in next time: otherwise, under the +/// 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 -/// the immediate-ready path and after a wake. This is a documented -/// guarantee (compose like BEAM receive clauses: put control channels -/// first), not an accident — and therefore there is NO fairness promise; a -/// saturated arm 0 starves arm 1 by design. +/// Arms are checked **in order**: index 0 is the highest priority, both +/// when checking immediately and after being woken. This is a deliberate, +/// documented guarantee, not an accident of implementation: put a control +/// or shutdown channel first so it is always noticed promptly. The +/// 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 -/// overlapping sets) freely. What stays illegal is what was always illegal: -/// two *different* actors receiving on one channel. +/// One actor can `select` on a channel and later plain `recv` on it (or +/// `select` again on an overlapping set of arms) with no restriction. What +/// 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 -/// registered under one wait epoch; the winning wake consumes it, so losing -/// arms' registrations are inert and need no cancellation pass — they -/// 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). +/// Panics if `arms` is empty, if called outside an actor, or if an fd arm +/// fails to register (see [`try_select`] for the fallible form; a +/// channel-only `select` can never fail). pub fn select(arms: &[&dyn Selectable]) -> usize { match try_select(arms) { 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 -/// arms can — EBADF, EMFILE on the epoll set, or a second waiter on an -/// fd that already has one). On `Err` the wait is fully retired and no +/// The fallible form of [`select`]: `Err` when an arm fails to register. +/// Only fd arms can fail this way (for example, the file descriptor is +/// 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 /// one has been unregistered. pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result { @@ -527,13 +609,12 @@ pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result { } // Stale fd registrations are not harmless (a losing fd arm's - // waiters entry poisons the fd with AlreadyExists and its - // kernel-side ONESHOT registration can fire arbitrarily late), so - // selects containing fd arms run an eager cleanup pass after the - // park — including when a terminal stop unwinds out of it, via - // the guard. Channel-only selects skip all of it: `eager` is - // false, the guard is disarmed, and the loser-arm self-cleaning - // story is unchanged. + // leftover registration can make the fd unusable for the next + // select until a kernel event happens to clear it), so selects + // containing fd arms run an eager cleanup pass after the park, + // including when a terminal stop unwinds out of it, via the guard. + // Channel-only selects skip all of it: `eager` is false, the guard + // is disarmed, and the loser-arm self-cleaning story is unchanged. let eager = arms.iter().any(|a| a.sel_eager_cleanup()); let mut guard = UnregisterGuard { arms, me, epoch, armed: eager }; @@ -546,22 +627,22 @@ pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result { drop(guard); // Woken precisely: an arm's send (message) or last-sender drop - // (closure) consumed our epoch, and both leave their arm ready — - // return the first one, in priority order (which may be a + // (closure) is what woke us, and both leave their arm ready. + // Return the first ready one, in priority order (which may be a // different, higher-priority arm than the one that woke us; its // message stays queued and re-reports ready on the next call). // Fd arms classify by a fresh zero-timeout poll, so they too are - // a pure function of state — independent of the registration the - // cleanup pass just removed. + // a pure function of current state, independent of the + // registration the cleanup pass just removed. for (i, arm) in arms.iter().enumerate() { if arm.sel_ready() { return Ok(i); } } - // Unreachable by protocol (a stop wake unwinds out of - // park_current). Defensive: re-open the wait and re-register — - // stale own-registrations are overwritten (channels) or were - // removed by the cleanup pass above (fds). + // Unreachable in practice (a stop wake unwinds out of + // park_current before we get here). Defensive: re-open the wait + // and re-register; stale own-registrations are overwritten + // (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 -/// out of `park_current`, and a registered fd arm must not outlive its -/// actor (the generalization of `wait_fd`'s `Dereg`). Disarmed on the -/// normal path after the explicit pass runs; never armed when no fd arm -/// registered, keeping the channel-only path guard-free in effect. +// 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 +// actor. Disarmed on the normal path after the explicit pass runs; never +// armed when no fd arm is registered, keeping the channel-only path +// guard-free in effect. struct UnregisterGuard<'a> { arms: &'a [&'a dyn Selectable], me: Pid, @@ -596,20 +677,16 @@ impl Drop for UnregisterGuard<'_> { } } -/// The registration pass shared by [`select`] and [`select_timeout`]: -/// check-or-register each arm, in priority order, each atomically under its -/// own lock. Cross-arm atomicity is unnecessary: an arm becoming ready -/// right after its registration wakes the caller through the protocol (the -/// prep-to-park window is closed by RunningNotified). -/// -/// `Ok(Some(i))` = arm `i` was ready, the pass stopped, and the wait has -/// been RETIRED (no park may follow): earlier arms hold live-epoch -/// registrations, so earlier *fd* arms are unregistered eagerly, then the -/// epoch is bumped, a landed notification eaten, and a pending stop -/// 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. +// The registration pass shared by `select` and `select_timeout`: check-or- +// register each arm, in priority order, each atomically under its own lock. +// Cross-arm atomicity is unnecessary: an arm becoming ready right after its +// registration still wakes the caller through the normal wake path. +// +// `Ok(Some(i))` = arm `i` was already ready, the pass stopped, and the wait +// has been fully retired (no park may follow): earlier fd arms are +// unregistered eagerly so none are left dangling. `Err` = an arm failed to +// register; same unwind (earlier fd arms unregistered, wait retired). +// `Ok(None)` = every arm registered successfully; the caller parks. fn register_arms( me: Pid, epoch: u32, @@ -633,10 +710,10 @@ fn register_arms( Ok(None) } -/// The [`select_timeout`] timer target: stateless, because precise wakes -/// make classification a pure function of channel state. The entry is -/// stamped with the select's epoch; if an arm already won, this unpark dies -/// at the word's epoch CAS (the no-cancellation convention in `timer.rs`). +// The `select_timeout` timer target: stateless, because a wake's cause can +// always be read back off plain channel state (an arm ready, or not). If +// an arm already won before the deadline, this timer's fire is simply +// ignored, the way any other stale wakeup is. struct SelectTimeout; impl crate::timer::TimerTarget for SelectTimeout { 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 -/// `None` once `timeout` elapses with no arm ready. +/// Like [`select`], but gives up and returns `None` if no arm becomes +/// ready before `timeout` elapses. /// -/// All of `select`'s semantics carry over (priority order, closed arms -/// permanently ready, no fairness promise). The timeout is one more stamped -/// waker on the same wait epoch — nothing is registered in any arm for it, -/// so there is nothing to cancel or leak: an arm winning leaves the timer -/// entry to expire as a stale-epoch no-op; the timer winning leaves the -/// arms' registrations to self-clean exactly as a `select` loser's would. +/// All of `select`'s semantics carry over: arms are still checked in +/// priority order, a closed arm is still permanently ready, and there is +/// still no fairness guarantee across arms. A message that arrives at +/// essentially the same moment the deadline passes still wins, the same +/// way [`Receiver::recv_timeout`] resolves that race. /// -/// The wake is classified from state alone (wakes are precise): some arm -/// ready → `Some` of the first, in priority order; none ready → the timer -/// 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 still gives an already-ready arm +/// a chance to be reported before falling through to `None`. /// -/// `Duration::ZERO` is a valid timeout: it parks until the immediately-due -/// timer is drained, then reports `None` unless an arm was already ready. -/// -/// 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). +/// Panics if `arms` is empty, if called outside an actor, or if an fd arm +/// fails to register (see [`try_select_timeout`] for the fallible form; a +/// channel-only select can never fail). pub fn select_timeout( arms: &[&dyn Selectable], timeout: std::time::Duration, @@ -677,9 +748,9 @@ pub fn select_timeout( } } -/// [`select_timeout`], fallible: `Err` when an arm fails to register -/// (only fd arms can). On `Err` the wait is fully retired and no -/// registration — arm-side or kernel-side — is left behind. +/// The fallible form of [`select_timeout`]: `Err` when an arm fails to +/// register (only fd arms can). On `Err` the wait is fully retired and no +/// registration is left behind on any arm. pub fn try_select_timeout( arms: &[&dyn Selectable], timeout: std::time::Duration, @@ -694,17 +765,16 @@ pub fn try_select_timeout( return Ok(Some(i)); // ready now: the timer was never armed } - // Arm the timer after the registration pass, outside every Channel - // lock (insert takes the timers lock). + // Arm the timer after the registration pass, outside every channel + // lock (inserting a timer takes the timers lock). let deadline = crate::timer::deadline_from_now(timeout); let target: std::sync::Arc = std::sync::Arc::new(SelectTimeout); crate::scheduler::insert_wait_timer(deadline, me, target, epoch); - // Same eager-cleanup story as `try_select`: the timer arm needs none - // (stateless, stale entries die at the epoch CAS), channel arms need - // none, fd arms do — and a timer win in particular leaves every fd - // arm's registration behind, which without this pass would poison - // those fds until a kernel event happened to fire. + // Same eager-cleanup story as `try_select`: a timer win in particular + // leaves every fd arm's registration behind, which without this pass + // would leave those fds unusable until a kernel event happened to + // clear them. let eager = arms.iter().any(|a| a.sel_eager_cleanup()); let mut guard = UnregisterGuard { arms, me, epoch, armed: eager };