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.
This commit is contained in:
smarm
2026-06-09 22:51:37 +00:00
parent d789d301e0
commit 8ff6cf4afd
4 changed files with 206 additions and 42 deletions
+28 -13
View File
@@ -1,9 +1,23 @@
//! Unbounded MPSC channels.
//!
//! Inner state is `Arc<Mutex<Inner<T>>>` so channels can be sent across OS
//! Inner state is `Arc<RawMutex<Inner<T>>>` 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<T>() -> (Sender<T>, Receiver<T>) {
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<T> {
}
pub struct Sender<T> {
inner: Arc<Mutex<Inner<T>>>,
inner: Arc<RawMutex<Inner<T>>>,
}
pub struct Receiver<T> {
inner: Arc<Mutex<Inner<T>>>,
inner: Arc<RawMutex<Inner<T>>>,
}
#[derive(Debug, PartialEq, Eq)]
@@ -59,7 +74,7 @@ impl std::error::Error for RecvError {}
impl<T> Clone for Sender<T> {
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<T> Clone for Sender<T> {
impl<T> Drop for Sender<T> {
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<T> Drop for Sender<T> {
impl<T> Drop for Receiver<T> {
fn drop(&mut self) {
self.inner.lock().unwrap().receiver_alive = false;
self.inner.lock().receiver_alive = false;
}
}
impl<T> Sender<T> {
pub fn send(&self, value: T) -> Result<(), SendError<T>> {
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<T> Receiver<T> {
pub fn recv(&self) -> Result<T, RecvError> {
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<T> Receiver<T> {
{
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<T> Receiver<T> {
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<T> Receiver<T> {
/// 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<Option<T>, 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));
}
+5 -4
View File
@@ -109,10 +109,11 @@ pub struct Monitor {
pub fn monitor(target: Pid) -> Monitor {
let (tx, rx) = channel::<Down>();
// 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) {
+120 -20
View File
@@ -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<u32> = const { std::cell::Cell::new(0) };
static LEAVES_HELD: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
static CHANNELS_HELD: std::cell::Cell<u32> = 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| {
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,
"leaf rule violated: acquiring a RawMutex while already holding one \
(cold locks / free list / stack pool are mutual leaves)"
"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<T> {
state: AtomicU32,
class: LockClass,
data: UnsafeCell<T>,
}
@@ -86,9 +135,20 @@ unsafe impl<T: Send> Send for RawMutex<T> {}
unsafe impl<T: Send> Sync for RawMutex<T> {}
impl<T> RawMutex<T> {
/// 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<T> RawMutex<T> {
// 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<T> 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
}
}
+48
View File
@@ -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::<i64>();
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::<i64>());
}