From 8ff6cf4afdb80ac3741bc9d6fe2821c5238071ba Mon Sep 17 00:00:00 2001 From: smarm Date: Tue, 9 Jun 2026 22:51:37 +0000 Subject: [PATCH] feat(channel): migrate Inner to RawMutex; two-class lock order (Leaf -> Channel) Phase 2 surfaced the hole this closes: channel guards were held with preemption enabled, so a timeslice switch inside a channel critical section could release the pthread mutex from a different OS thread (UB). RawMutex disables preemption for the guard span and is cross-thread-release sound by construction; poison goes away with it. The strict single-class leaf rule cannot survive the migration: finalize clones the supervisor/trap senders and monitor() clones the Down sender, all under a cold lock, and those senders live in the slot - the nesting is structural. The leaf check is therefore generalized to two classes: Leaf (cold locks, free list, stack pool) and Channel. Order is Leaf -> Channel, at most one of each; both directions of violation are debug-asserted at the acquisition site. Channel critical sections call only the lock-free unpark protocol, so the order is acyclic. Tests: class-ordering unit tests in raw_mutex (allowed nesting + all three rejected shapes), plus a multi-scheduler integration test driving channels through monitor churn and actor death so any ordering regression trips the debug assert instead of deadlocking. --- src/channel.rs | 41 +++++++++---- src/monitor.rs | 9 +-- src/raw_mutex.rs | 150 +++++++++++++++++++++++++++++++++++++++-------- tests/channel.rs | 48 +++++++++++++++ 4 files changed, 206 insertions(+), 42 deletions(-) diff --git a/src/channel.rs b/src/channel.rs index 8d59348..b3a1f7e 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -1,9 +1,23 @@ //! Unbounded MPSC channels. //! -//! Inner state is `Arc>>` so channels can be sent across OS +//! 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. @@ -15,11 +29,12 @@ //! parked, the receiver is unparked. use crate::pid::Pid; +use crate::raw_mutex::RawMutex; use std::collections::VecDeque; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; pub fn channel() -> (Sender, Receiver) { - let inner = Arc::new(Mutex::new(Inner { + let inner = Arc::new(RawMutex::new_channel(Inner { queue: VecDeque::new(), parked_receiver: None, senders: 1, @@ -36,11 +51,11 @@ struct Inner { } pub struct Sender { - inner: Arc>>, + inner: Arc>>, } pub struct Receiver { - inner: Arc>>, + inner: Arc>>, } #[derive(Debug, PartialEq, Eq)] @@ -59,7 +74,7 @@ impl std::error::Error for RecvError {} impl Clone for Sender { fn clone(&self) -> Self { - self.inner.lock().unwrap().senders += 1; + self.inner.lock().senders += 1; Sender { inner: self.inner.clone() } } } @@ -67,7 +82,7 @@ impl Clone for Sender { impl Drop for Sender { fn drop(&mut self) { let unpark = { - let mut g = self.inner.lock().unwrap(); + 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 @@ -89,14 +104,14 @@ impl Drop for Sender { impl Drop for Receiver { fn drop(&mut self) { - self.inner.lock().unwrap().receiver_alive = false; + self.inner.lock().receiver_alive = false; } } impl Sender { pub fn send(&self, value: T) -> Result<(), SendError> { let unpark = { - let mut g = self.inner.lock().unwrap(); + let mut g = self.inner.lock(); if !g.receiver_alive { return Err(SendError(value)); } @@ -117,7 +132,7 @@ impl Receiver { pub fn recv(&self) -> Result { loop { { - let mut g = self.inner.lock().unwrap(); + let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { return Ok(v); } @@ -156,7 +171,7 @@ impl Receiver { { loop { { - let mut g = self.inner.lock().unwrap(); + 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()); @@ -188,7 +203,7 @@ impl Receiver { where F: Fn(&T) -> bool, { - let mut g = self.inner.lock().unwrap(); + 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())); } @@ -201,7 +216,7 @@ impl Receiver { /// 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().unwrap(); + let mut g = self.inner.lock(); if let Some(v) = g.queue.pop_front() { return Ok(Some(v)); } diff --git a/src/monitor.rs b/src/monitor.rs index d6eb158..b299a61 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -109,10 +109,11 @@ pub struct Monitor { pub fn monitor(target: Pid) -> Monitor { let (tx, rx) = channel::(); - // Register under the target's cold lock. `tx.clone()` only touches the - // channel's own mutex, never any runtime lock, so it is safe here — but - // we must not *send* under the lock, as `Sender::send` can unpark a - // parked receiver, and there's no reason to nest that. + // Register under the target's cold lock. `tx.clone()` takes the channel's + // own lock — a Channel-class RawMutex, explicitly permitted *under* a Leaf + // (cold) lock by the lock order (see raw_mutex.rs). We must still not + // *send* under the lock, as `Sender::send` can unpark a parked receiver, + // and there's no reason to nest that. let (id, registered) = with_runtime(|inner| { let id = inner.alloc_monitor_id(); let registered = match inner.slot_at(target) { diff --git a/src/raw_mutex.rs b/src/raw_mutex.rs index 3152a56..628c898 100644 --- a/src/raw_mutex.rs +++ b/src/raw_mutex.rs @@ -25,12 +25,20 @@ //! Tricky", mutex3). 0 = unlocked, 1 = locked, 2 = locked with (possible) //! waiters. Uncontended lock/unlock is one CAS / one swap, no syscall. //! -//! Lock-order position: slot cold locks, the free list, and the stack pool -//! are all `RawMutex`es and all *leaves among themselves* — never hold two at -//! once. Holding a `RawMutex` while pushing to the run queue is permitted -//! (e.g. unpark from inside a cold section); the reverse — taking any -//! `RawMutex` from inside a run-queue op — cannot arise (queue ops call -//! nothing). +//! Lock-order position — two classes (see [`LockClass`]): +//! +//! - **Leaf**: slot cold locks, the free list, the stack pool, the name +//! registry. Mutual leaves — never hold two at once. +//! - **Channel**: a channel's internal lock. May be acquired *under* a Leaf +//! (finalize clones the supervisor/trap senders, and `monitor()` clones the +//! Down sender, all under a cold lock — structural, the sender lives in the +//! slot), but nothing may be acquired under a Channel lock: channel +//! critical sections call only the lock-free unpark protocol. +//! +//! So the total order is Leaf → Channel, one of each at most. Holding either +//! while pushing to the run queue is permitted (unpark from inside a cold or +//! channel section); the reverse — taking any `RawMutex` from inside a +//! run-queue op — cannot arise (queue ops call nothing). use std::cell::UnsafeCell; use std::ops::{Deref, DerefMut}; @@ -45,37 +53,78 @@ const CONTENDED: u32 = 2; /// sender), so a short spin almost always avoids the syscall. const SPIN_LIMIT: u32 = 64; -// The leaf rule, mechanically enforced (debug builds): slot cold locks, the -// free list, and the stack pool — i.e. every RawMutex — are mutual leaves. -// Holding two at once is a deadlock waiting for the right interleaving, so -// fail at the acquisition that violates it, not in the eventual hang. +/// Which rung of the two-rung lock order a `RawMutex` occupies. Debug builds +/// enforce the order mechanically (see the module docs); release builds carry +/// no state and no checks. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum LockClass { + /// Runtime cold data: slot cold locks, free list, stack pool, registry. + /// Mutual leaves among themselves; a Channel lock may be taken under one. + Leaf, + /// A channel's internal lock. One at a time, nothing acquired under it; + /// may itself be acquired under a Leaf. + Channel, +} + +// The ordering rules, mechanically enforced (debug builds): a deadlock from a +// violated order is a hang waiting for the right interleaving, so fail at the +// acquisition that violates it, not in the eventual hang. #[cfg(debug_assertions)] thread_local! { - static RAW_MUTEXES_HELD: std::cell::Cell = const { std::cell::Cell::new(0) }; + static LEAVES_HELD: std::cell::Cell = const { std::cell::Cell::new(0) }; + static CHANNELS_HELD: std::cell::Cell = const { std::cell::Cell::new(0) }; } #[inline] -fn leaf_check_acquire() { +fn order_check_acquire(class: LockClass) { #[cfg(debug_assertions)] - RAW_MUTEXES_HELD.with(|c| { - debug_assert_eq!( - c.get(), - 0, - "leaf rule violated: acquiring a RawMutex while already holding one \ - (cold locks / free list / stack pool are mutual leaves)" - ); - c.set(c.get() + 1); - }); + match class { + LockClass::Leaf => LEAVES_HELD.with(|l| { + debug_assert_eq!( + l.get(), + 0, + "lock order violated: acquiring a Leaf RawMutex while already \ + holding one (cold locks / free list / stack pool / registry \ + are mutual leaves)" + ); + CHANNELS_HELD.with(|c| { + debug_assert_eq!( + c.get(), + 0, + "lock order violated: acquiring a Leaf RawMutex under a \ + channel lock (order is Leaf -> Channel, never the reverse)" + ); + }); + l.set(l.get() + 1); + }), + LockClass::Channel => CHANNELS_HELD.with(|c| { + debug_assert_eq!( + c.get(), + 0, + "lock order violated: acquiring a channel lock while already \ + holding one (channel locks are mutual leaves)" + ); + c.set(c.get() + 1); + }), + } + #[cfg(not(debug_assertions))] + let _ = class; } #[inline] -fn leaf_check_release() { +fn order_check_release(class: LockClass) { #[cfg(debug_assertions)] - RAW_MUTEXES_HELD.with(|c| c.set(c.get() - 1)); + match class { + LockClass::Leaf => LEAVES_HELD.with(|c| c.set(c.get() - 1)), + LockClass::Channel => CHANNELS_HELD.with(|c| c.set(c.get() - 1)), + } + #[cfg(not(debug_assertions))] + let _ = class; } pub(crate) struct RawMutex { state: AtomicU32, + class: LockClass, data: UnsafeCell, } @@ -86,9 +135,20 @@ unsafe impl Send for RawMutex {} unsafe impl Sync for RawMutex {} impl RawMutex { + /// A Leaf-class mutex — the default for runtime cold data. pub(crate) const fn new(data: T) -> Self { + Self::with_class(data, LockClass::Leaf) + } + + /// A Channel-class mutex — for channel internals only. + pub(crate) const fn new_channel(data: T) -> Self { + Self::with_class(data, LockClass::Channel) + } + + pub(crate) const fn with_class(data: T, class: LockClass) -> Self { Self { state: AtomicU32::new(UNLOCKED), + class, data: UnsafeCell::new(data), } } @@ -98,7 +158,7 @@ impl RawMutex { // Enter NoPreempt *before* acquiring, so a preemption can't fire // between acquisition and guard construction. let prev_preempt = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); - leaf_check_acquire(); + order_check_acquire(self.class); if self .state .compare_exchange(UNLOCKED, LOCKED, Ordering::Acquire, Ordering::Relaxed) @@ -172,7 +232,7 @@ impl Drop for RawMutexGuard<'_, T> { #[inline] fn drop(&mut self) { self.m.unlock(); - leaf_check_release(); + order_check_release(self.m.class); // Restore preemption only after the lock is released. crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(self.prev_preempt)); } @@ -257,4 +317,44 @@ mod tests { *m.lock() += 1; assert_eq!(*m.lock(), 1); } + + #[test] + fn channel_lock_nests_under_leaf() { + // The permitted ordering: Leaf -> Channel (finalize/monitor clone a + // sender under a cold lock). Must not trip the order check. + let leaf = RawMutex::new(0u64); + let chan = RawMutex::new_channel(0u64); + let _l = leaf.lock(); + let _c = chan.lock(); + } + + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "lock order violated")] + fn leaf_under_channel_is_rejected() { + let leaf = RawMutex::new(0u64); + let chan = RawMutex::new_channel(0u64); + let _c = chan.lock(); + let _l = leaf.lock(); // Channel -> Leaf: forbidden + } + + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "lock order violated")] + fn two_leaves_are_rejected() { + let a = RawMutex::new(0u64); + let b = RawMutex::new(0u64); + let _ga = a.lock(); + let _gb = b.lock(); // leaves are mutual: forbidden + } + + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "lock order violated")] + fn two_channel_locks_are_rejected() { + let a = RawMutex::new_channel(0u64); + let b = RawMutex::new_channel(0u64); + let _ga = a.lock(); + let _gb = b.lock(); // channel locks are mutual leaves: forbidden + } } diff --git a/tests/channel.rs b/tests/channel.rs index ed87c8d..41d9575 100644 --- a/tests/channel.rs +++ b/tests/channel.rs @@ -108,3 +108,51 @@ fn recv_returns_err_when_all_senders_dropped() { assert!(saw_err.load(std::sync::atomic::Ordering::SeqCst)); } + +#[test] +fn channel_ops_interleaved_with_monitor_churn_multi_thread() { + // Regression for the RawMutex migration: monitor registration clones the + // Down sender under the target's cold (Leaf) lock, which now nests a + // Channel-class lock under it. Debug builds enforce the Leaf -> Channel + // ordering on every acquisition, so driving channels, monitors, and actor + // death concurrently across schedulers makes any ordering regression + // panic here rather than deadlock in the field. + use std::sync::atomic::{AtomicI64, Ordering}; + use std::sync::Arc; + + let total = Arc::new(AtomicI64::new(0)); + let total2 = total.clone(); + smarm::init(smarm::Config::exact(4)).run(move || { + let (tx, rx) = channel::(); + let consumer = spawn(move || { + let mut sum = 0; + while let Ok(v) = rx.recv() { + sum += v; + } + OUT.with(|c| c.set(sum)); // not asserted cross-thread; see total + total2.fetch_add(sum, Ordering::Relaxed); + }); + + let mut handles = Vec::new(); + for i in 0..32i64 { + let tx = tx.clone(); + handles.push(spawn(move || { + // Short-lived target whose death fires the monitor below. + let t = spawn(move || { + tx.send(i).unwrap(); + }); + let m = smarm::monitor(t.pid()); + t.join().unwrap(); + // Down delivery exercises send-from-finalize. + let d = m.rx.recv().unwrap(); + assert_eq!(d.reason, smarm::DownReason::Exit); + })); + } + drop(tx); + for h in handles { + h.join().unwrap(); + } + consumer.join().unwrap(); + }); + assert_eq!(total.load(std::sync::atomic::Ordering::Relaxed), (0..32).sum::()); +}