//! Unbounded MPSC channels. //! //! 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). //! //! ## Why `RawMutex` (Channel class), not `std::sync::Mutex` //! //! 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. //! //! 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. use crate::pid::Pid; use crate::raw_mutex::RawMutex; use std::collections::VecDeque; use std::sync::Arc; pub fn channel() -> (Sender, Receiver) { let inner = Arc::new(RawMutex::new_channel(Inner { queue: VecDeque::new(), parked_receiver: None, senders: 1, receiver_alive: true, })); (Sender { inner: inner.clone() }, Receiver { inner }) } 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. parked_receiver: Option<(Pid, u32)>, senders: usize, receiver_alive: bool, } pub struct Sender { inner: Arc>>, } pub struct Receiver { inner: Arc>>, } #[derive(Debug, PartialEq, Eq)] pub struct SendError(pub T); #[derive(Debug, PartialEq, Eq, Clone, Copy)] pub struct RecvError; impl std::fmt::Display for RecvError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "channel closed") } } impl std::error::Error for RecvError {} /// Returned by [`Receiver::recv_timeout`]. #[derive(Debug, PartialEq, Eq, Clone, Copy)] pub enum RecvTimeoutError { /// The deadline passed with no message available. Timeout, /// All senders dropped with no message available — the bounded analogue /// of [`RecvError`]. Disconnected, } impl std::fmt::Display for RecvTimeoutError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { RecvTimeoutError::Timeout => write!(f, "recv timed out"), RecvTimeoutError::Disconnected => write!(f, "channel closed"), } } } impl std::error::Error for RecvTimeoutError {} impl Clone for Sender { fn clone(&self) -> Self { self.inner.lock().senders += 1; Sender { inner: self.inner.clone() } } } impl Drop for Sender { fn drop(&mut self) { let unpark = { let mut g = self.inner.lock(); g.senders -= 1; // 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 // return Err rather than sleep forever. if g.senders == 0 { g.parked_receiver.take() } else { None } }; if let Some((pid, epoch)) = unpark { crate::scheduler::unpark_at(pid, epoch); } } } impl Drop for Receiver { fn drop(&mut self) { self.inner.lock().receiver_alive = false; } } impl Sender { pub fn send(&self, value: T) -> Result<(), SendError> { let unpark = { let mut g = self.inner.lock(); if !g.receiver_alive { return Err(SendError(value)); } g.queue.push_back(value); g.parked_receiver.take() }; if let Some((pid, epoch)) = unpark { crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: Some(pid) }); crate::scheduler::unpark_at(pid, epoch); } else { crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: None }); } Ok(()) } } impl Receiver { pub fn recv(&self) -> Result { loop { { let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { return Ok(v); } if g.senders == 0 { return Err(RecvError); } let me = crate::actor::current_pid() .expect("recv() called outside an actor"); debug_assert!( 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; // 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. crate::scheduler::park_current(); // Woken up — record it before looping to check the queue. crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); } } /// Bounded receive: like [`recv`](Self::recv), but gives up once /// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`]. /// /// 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`. /// /// 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. pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result where T: Send + 'static, { let me = crate::actor::current_pid() .expect("recv_timeout() called outside an actor"); // Fast path + wait registration, one critical section. let epoch; { let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { return Ok(v); } if g.senders == 0 { return Err(RecvTimeoutError::Disconnected); } debug_assert!( g.parked_receiver.is_none_or(|(p, _)| p == me), "channel has more than one receiver" ); epoch = crate::scheduler::begin_wait(); g.parked_receiver = Some((me, epoch)); crate::te!(crate::trace::Event::RecvPark(me)); } // 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 // 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(); crate::scheduler::insert_wait_timer(deadline, me, target, epoch); crate::scheduler::park_current(); crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { return Ok(v); } if g.senders == 0 { return Err(RecvTimeoutError::Disconnected); } 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. /// /// `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. pub fn recv_match(&self, pred: F) -> Result where F: Fn(&T) -> bool, { loop { { let mut g = self.inner.lock(); if let Some(i) = g.queue.iter().position(|v| pred(v)) { // position() found it, so remove() returns Some. return Ok(g.queue.remove(i).unwrap()); } if g.senders == 0 { // Closed and nothing queued can ever match. return Err(RecvError); } let me = crate::actor::current_pid() .expect("recv_match() called outside an actor"); debug_assert!( g.parked_receiver.is_none_or(|(p, _)| p == me), "channel has more than one 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. crate::scheduler::park_current(); crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); } } /// 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). pub fn try_recv_match(&self, pred: F) -> Result, RecvError> where F: Fn(&T) -> bool, { let mut g = self.inner.lock(); if let Some(i) = g.queue.iter().position(|v| pred(v)) { return Ok(Some(g.queue.remove(i).unwrap())); } if g.senders == 0 { return Err(RecvError); } 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. pub fn try_recv(&self) -> Result, RecvError> { let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { return Ok(Some(v)); } if g.senders == 0 { return Err(RecvError); } Ok(None) } } // --------------------------------------------------------------------------- // 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 // 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.) let unpark = { let mut g = self.lock(); if g.parked_receiver == Some((pid, epoch)) { g.parked_receiver = None; true } else { false } }; // 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); } } } // --------------------------------------------------------------------------- // select — ready-index wait over multiple receivers // --------------------------------------------------------------------------- pub(crate) mod sealed { pub trait 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. /// /// Contract (all under the arm's own lock): `sel_register` checks-or- /// 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 /// the arm could not register at all (only fd arms can fail; channel /// registration is infallible) — 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. #[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. #[doc(hidden)] fn sel_eager_cleanup(&self) -> bool { false } } impl Selectable for Receiver { fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result { let mut g = self.inner.lock(); if !g.queue.is_empty() || g.senders == 0 { return Ok(false); } debug_assert!( g.parked_receiver.is_none_or(|(p, _)| p == pid), "channel has more than one receiver" ); g.parked_receiver = Some((pid, epoch)); Ok(true) } fn sel_ready(&self) -> bool { let g = self.inner.lock(); !g.queue.is_empty() || g.senders == 0 } } /// Park on every arm at once; return the index of the first ready one. /// /// "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. /// /// 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. /// /// 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. /// /// 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. /// /// 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). pub fn select(arms: &[&dyn Selectable]) -> usize { try_select(arms).expect("select(): fd arm failed to register (use try_select)") } /// [`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 /// registration is left behind: every arm registered before the failing /// one has been unregistered. pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result { assert!(!arms.is_empty(), "select() on an empty arm list"); let me = crate::actor::current_pid().expect("select() called outside an actor"); loop { let epoch = crate::scheduler::begin_wait(); if let Some(i) = register_arms(me, epoch, arms)? { return Ok(i); } // 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. let eager = arms.iter().any(|a| a.sel_eager_cleanup()); let mut guard = UnregisterGuard { arms, me, epoch, armed: eager }; crate::scheduler::park_current(); if eager { unregister_arms(arms, me, epoch); } guard.armed = false; 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 // 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. 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). } } /// Eager-cleanup sweep: remove every fd arm's registration that is still /// ours. No-op per channel arm (one virtual call); one io-lock visit per /// fd arm. fn unregister_arms(arms: &[&dyn Selectable], me: Pid, epoch: u32) { for arm in arms { if arm.sel_eager_cleanup() { arm.sel_unregister(me, epoch); } } } /// 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. struct UnregisterGuard<'a> { arms: &'a [&'a dyn Selectable], me: Pid, epoch: u32, armed: bool, } impl Drop for UnregisterGuard<'_> { fn drop(&mut self) { if self.armed { unregister_arms(self.arms, self.me, self.epoch); } } } /// 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. fn register_arms( me: Pid, epoch: u32, arms: &[&dyn Selectable], ) -> std::io::Result> { for (i, arm) in arms.iter().enumerate() { let registered = match arm.sel_register(me, epoch) { Ok(r) => r, Err(e) => { unregister_arms(&arms[..i], me, epoch); crate::scheduler::retire_wait(); return Err(e); } }; if !registered { unregister_arms(&arms[..i], me, epoch); crate::scheduler::retire_wait(); return Ok(Some(i)); } } 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`). struct SelectTimeout; impl crate::timer::TimerTarget for SelectTimeout { fn on_timeout(&self, pid: Pid, epoch: u32) { crate::scheduler::unpark_at(pid, epoch); } } /// [`select`] with a deadline: returns `Some(index)` like `select`, or /// `None` once `timeout` elapses with no arm ready. /// /// 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. /// /// 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 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). pub fn select_timeout( arms: &[&dyn Selectable], timeout: std::time::Duration, ) -> Option { try_select_timeout(arms, timeout) .expect("select_timeout(): fd arm failed to register (use try_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. pub fn try_select_timeout( arms: &[&dyn Selectable], timeout: std::time::Duration, ) -> std::io::Result> { assert!(!arms.is_empty(), "select_timeout() on an empty arm list"); let me = crate::actor::current_pid().expect("select_timeout() called outside an actor"); let epoch = crate::scheduler::begin_wait(); if let Some(i) = register_arms(me, epoch, arms)? { 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). 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. let eager = arms.iter().any(|a| a.sel_eager_cleanup()); let mut guard = UnregisterGuard { arms, me, epoch, armed: eager }; crate::scheduler::park_current(); if eager { unregister_arms(arms, me, epoch); } guard.armed = false; drop(guard); // Woken precisely: an arm (ready below) or the timer (nothing ready). Ok(arms.iter().position(|arm| arm.sel_ready())) }