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

feat(runtime): phase 1 — peel timers/io/monitor-id out of SharedState; gate stop sentinel behind PREEMPTION_ENABLED

Browse Source 
- next_monitor_id -> AtomicU64 on RuntimeInner
- timers -> own Mutex<Timers>; io -> own Mutex<Option<IoThread>> (lock order: io-before-shared)
- pending_closures Vec folded into Slot::pending_closure
- termination check reads io liveness before shared; ordering argument documented
- poison fix: check_cancelled() no longer fires while preemption is disabled,
  so a cancellation unwind can never poison a runtime/channel mutex
- regression test: tests/poison_stop.rs
- ROADMAP_v0.5.md added
This commit is contained in:
Claude
2026-06-09 18:56:17 +00:00
parent d908eb3f95
commit 3c7e26bc98
6 changed files with 303 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/monitor.rs
+6 -6
View File
@@ -117,17 +117,17 @@ pub fn monitor(target: Pid) -> Monitor {
// we must not *send* here, as `Sender::send` can call back in to unpark a
// parked receiver and the shared mutex is not reentrant.
let (id, registered) = with_runtime(|inner| {
inner.with_shared(|s| {
let id = s.alloc_monitor_id();
let registered = match s.slot_mut(target) {
let id = inner.alloc_monitor_id();
let registered = inner.with_shared(|s| {
match s.slot_mut(target) {
Some(slot) if !matches!(slot.state, State::Done) => {
slot.monitors.push((id, tx.clone()));
true
}
_ => false,
};
(id, registered)
})
}
});
(id, registered)
});
if !registered {
src/preempt.rs
+15 -4
View File
@@ -171,11 +171,22 @@ pub fn maybe_preempt() {
let n = c.get();
if n == 0 {
c.set(CONFIGURED_ALLOC_INTERVAL.with(|i| i.get()));
// Cooperative cancellation shares the amortised cadence with the
// timeslice check. Observe a pending stop first: if we are being
// cancelled there is no point yielding, we unwind instead.
check_cancelled();
if PREEMPTION_ENABLED.with(|e| e.get()) {
// Cooperative cancellation shares the amortised cadence with
// the timeslice check, and shares its gate: while preemption
// is disabled (`NoPreempt`, `with_shared`, channel critical
// sections) the stop sentinel must NOT be raised, because an
// allocation-triggered unwind inside a region holding a
// `std::sync::Mutex` would poison it — one `request_stop` at
// the wrong moment would then cascade `lock().unwrap()`
// panics through every later user of that lock. Observation
// is merely deferred to the next enabled allocation or the
// wakeup side of the next park/yield, both of which are
// lock-free points by construction.
//
// Observe a pending stop first: if we are being cancelled
// there is no point yielding, we unwind instead.
check_cancelled();
let start = TIMESLICE_START.with(|s| s.get());
if rdtsc().saturating_sub(start) > CONFIGURED_TIMESLICE_CYCLES.with(|t| t.get()) {
// SAFETY: reachable only inside an actor (the scheduler
src/runtime.rs
+116 -76
View File
@@ -240,6 +240,12 @@ pub(crate) struct Slot {
/// The scheduler checks this after a Park yield and re-queues instead
/// of sleeping, closing the lost-wakeup window.
pub(crate) pending_unpark: bool,
/// This actor's closure awaiting its first resume. `Some` until the
/// scheduler pops it on first dequeue, `None` thereafter. Previously a
/// parallel `pending_closures: Vec<Option<Closure>>` indexed by slot
/// index; folded into the slot since it is per-actor data with the same
/// lifetime as the slot and no independent access pattern.
pub(crate) pending_closure: Option<Closure>,
}
impl Slot {
@@ -256,6 +262,7 @@ impl Slot {
outstanding_handles: 0,
pending_io_result: None,
pending_unpark: false,
pending_closure: None,
}
}
}
@@ -267,16 +274,6 @@ pub(crate) struct SharedState {
pub(crate) free_list: Vec<u32>,
pub(crate) run_queue: VecDeque<Pid>,
pub(crate) root_pid: Option<Pid>,
pub(crate) timers: Timers,
pub(crate) io: Option<IoThread>,
/// Closures awaiting their first resume, indexed by slot index.
/// `pending_closures[idx]` is `Some` iff the actor at slot `idx` has not
/// yet been resumed for the first time. Grows on demand; never shrinks.
pub(crate) pending_closures: Vec<Option<Closure>>,
/// Monotonic source of `MonitorId`s, bumped under the shared lock by
/// `alloc_monitor_id`. Never reused, so `demonitor` can name one of several
/// monitors on the same target unambiguously.
pub(crate) next_monitor_id: u64,
}
impl SharedState {
@@ -286,20 +283,9 @@ impl SharedState {
free_list: Vec::new(),
run_queue: VecDeque::new(),
root_pid: None,
timers: Timers::new(),
io: None,
pending_closures: Vec::new(), // indexed by slot index
next_monitor_id: 0,
}
}
/// Allocate the next process-unique `MonitorId`. Caller holds the shared
/// lock (this is only reached from `monitor()` under `with_shared`).
pub(crate) fn alloc_monitor_id(&mut self) -> MonitorId {
self.next_monitor_id += 1;
MonitorId(self.next_monitor_id)
}
pub(crate) fn allocate_slot(&mut self) -> (u32, u32) {
if let Some(idx) = self.free_list.pop() {
let gen = self.slots[idx as usize].generation;
@@ -320,12 +306,6 @@ impl SharedState {
let s = self.slots.get_mut(pid.index() as usize)?;
if s.generation == pid.generation() { Some(s) } else { None }
}
pub(crate) fn pop_pending_closure(&mut self, pid: Pid) -> Option<Closure> {
self.pending_closures
.get_mut(pid.index() as usize)
.and_then(|cell| cell.take())
}
}
// ---------------------------------------------------------------------------
@@ -334,6 +314,18 @@ impl SharedState {
pub(crate) struct RuntimeInner {
pub(crate) shared: Mutex<SharedState>,
/// Timer heap, peeled out of `shared` (RFC: state decomposition). Its own
/// mutex: only the drain-winner and the blocking primitives (sleep, wait)
/// touch it, never on the pure-compute hot path.
pub(crate) timers: Mutex<Timers>,
/// IO subsystem, peeled out of `shared`. `None` between runs; `Some` for
/// the duration of a `run()`. Holds its own completion mutex internally;
/// this outer mutex guards only the `Option`/`IoThread` bookkeeping
/// (submit, epoll register/deregister, completion drain).
pub(crate) io: Mutex<Option<IoThread>>,
/// Monotonic `MonitorId` source, peeled out of `shared`. Never reused, so
/// `demonitor` can name one of several monitors on a target unambiguously.
pub(crate) next_monitor_id: AtomicU64,
/// Try-lock: exactly one scheduler thread drains timers/IO per iteration.
drain_lock: Mutex<()>,
/// Per-thread stats, indexed by scheduler thread slot (0..N).
@@ -356,6 +348,9 @@ impl RuntimeInner {
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
Arc::new(Self {
shared: Mutex::new(SharedState::new()),
timers: Mutex::new(Timers::new()),
io: Mutex::new(None),
next_monitor_id: AtomicU64::new(0),
drain_lock: Mutex::new(()),
stats,
io_parked: AtomicU32::new(0),
@@ -378,6 +373,12 @@ impl RuntimeInner {
crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev));
result
}
/// Allocate the next process-unique `MonitorId`. Lock-free; monitors are a
/// cold path but there is no reason to serialize id minting under any lock.
pub(crate) fn alloc_monitor_id(&self) -> MonitorId {
MonitorId(self.next_monitor_id.fetch_add(1, Ordering::Relaxed) + 1)
}
}
// ---------------------------------------------------------------------------
@@ -439,8 +440,8 @@ impl Runtime {
let mut s = self.inner.shared.lock().unwrap();
assert!(s.run_queue.is_empty(), "run() called while previous run still active");
s.root_pid = Some(ROOT_PID);
s.io = Some(IoThread::start().expect("failed to start IO thread"));
}
*self.inner.io.lock().unwrap() = Some(IoThread::start().expect("failed to start IO thread"));
// Spawn the initial actor through the public spawn path (which
// requires a running runtime in the thread-local).
@@ -473,9 +474,9 @@ impl Runtime {
drop(initial_handle);
// Tear down IO and clean up shared state for the next run() call.
let mut s = self.inner.shared.lock().unwrap();
drop(s.io.take()); // joins IO threads
s.pending_closures.clear();
drop(self.inner.io.lock().unwrap().take()); // joins IO threads
self.inner.timers.lock().unwrap().clear();
self.inner.next_monitor_id.store(0, Ordering::Relaxed);
// Reset per-thread stats.
for stat in &self.inner.stats {
stat.current_pid_index.store(u32::MAX, Ordering::Relaxed);
@@ -547,6 +548,7 @@ pub(crate) fn reclaim_slot(s: &mut SharedState, pid: Pid) {
slot.outstanding_handles = 0;
slot.pending_unpark = false;
slot.pending_io_result = None;
slot.pending_closure = None;
s.free_list.push(idx);
}
@@ -670,31 +672,29 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
// losers skip immediately and proceed to step 2.
// ----------------------------------------------------------------
if let Ok(_drain_guard) = inner.drain_lock.try_lock() {
// Drain due timers and IO completions in a SINGLE shared-lock
// acquisition. Previously this took two separate `with_shared`
// calls (one for timers, one for IO) plus an unconditional
// `Instant::now()` every loop iteration. On the pure-yield /
// pure-compute hot path there are never any timers, so the clock
// read and the extra lock were pure per-yield overhead. We now
// read the clock only when the timer heap is non-empty and fold
// both drains into one lock hold.
// Drain due timers and IO completions. These now live behind their
// own mutexes (peeled out of `shared`), so the pure-yield /
// pure-compute hot path no longer contends the global lock just to
// discover there is nothing to drain. We still read the clock only
// when the timer heap is non-empty.
//
// IO completions are still drained unconditionally while the IO
// subsystem is present: the IO/pool threads push completions onto
// their own mutex (not `shared`), so the scheduler can't cheaply
// know in advance whether any arrived — it must look. That look is
// a single empty-VecDeque check on the empty path.
let (due, completions) = inner.with_shared(|s| {
let due = if s.timers.is_empty() {
// their own mutex, so the scheduler can't cheaply know in advance
// whether any arrived — it must look. That look is a single
// empty-VecDeque check on the empty path.
let due = {
let mut t = inner.timers.lock().unwrap();
if t.is_empty() {
Vec::new()
} else {
s.timers.pop_due(std::time::Instant::now())
};
let completions = s.io.as_mut()
.map(|io| io.drain_completions())
.unwrap_or_default();
(due, completions)
});
t.pop_due(std::time::Instant::now())
}
};
let completions = inner.io.lock().unwrap()
.as_mut()
.map(|io| io.drain_completions())
.unwrap_or_default();
for entry in due {
match entry.reason {
crate::timer::Reason::Sleep => {
@@ -731,10 +731,10 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
for completion in completions {
match completion {
crate::io::Completion::Blocking { pid, result } => {
if let Some(io) = inner.io.lock().unwrap().as_mut() {
io.outstanding = io.outstanding.saturating_sub(1);
}
inner.with_shared(|s| {
if let Some(io) = s.io.as_mut() {
io.outstanding = io.outstanding.saturating_sub(1);
}
if let Some(slot) = s.slot_mut(pid) {
slot.pending_io_result = Some(result);
if matches!(slot.state, State::Parked) {
@@ -746,12 +746,18 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
});
}
crate::io::Completion::FdReady { fd, events: _ } => {
// Resolve the parked pid under the io lock, then do the
// state transition under the shared lock. Lock order is
// always io-before-shared; never the reverse. The two
// are independent here (the waiters map is io-private,
// the slot table is shared), so taking them in sequence
// rather than nested is both correct and cheaper.
let parked_pid = inner.io.lock().unwrap().as_mut().and_then(|io| {
let pid = io.waiters.remove(&fd);
io.epoll_deregister(fd);
pid
});
inner.with_shared(|s| {
let parked_pid = s.io.as_mut().and_then(|io| {
let pid = io.waiters.remove(&fd);
io.epoll_deregister(fd);
pid
});
if let Some(pid) = parked_pid {
if let Some(slot) = s.slot_mut(pid) {
match slot.state {
@@ -804,6 +810,14 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
// thread immediately after this block (switch_to_actor / set_current_stop).
unsafe impl Send for PopResult {} // closure is Send; usize sp + stop ptr are plain data
// Read IO liveness BEFORE taking the shared lock. Lock order is
// io-before-shared everywhere, and reading it first is also what makes
// the split termination check correct: see the all_clear note below.
let (io_out, io_fd) = match inner.io.lock().unwrap().as_ref() {
Some(io) => (io.outstanding + io.waiters.len() as u32, Some(io.wake_fd())),
None => (0, None),
};
let pop = inner.with_shared(|s| {
let len = s.run_queue.len() as u64;
stats.run_queue_len.store(len, Ordering::Relaxed);
@@ -815,12 +829,14 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
// box stays alive and at a fixed address for as long as the
// actor is on-CPU (it can't be finalized until it yields back),
// so the scheduler's resume path pays no atomic ref-count cost.
let info = s.slot(pid)
.and_then(|slot| slot.actor.as_ref()
.map(|a| (a.sp, std::sync::Arc::as_ptr(&a.stop))));
let info = s.slot_mut(pid)
.and_then(|slot| {
let closure = slot.pending_closure.take();
slot.actor.as_ref()
.map(|a| (a.sp, std::sync::Arc::as_ptr(&a.stop), closure))
});
match info {
Some((sp, stop)) => {
let closure = s.pop_pending_closure(pid);
Some((sp, stop, closure)) => {
PopResult::Work(pid, sp, stop, closure)
}
None => {
@@ -839,32 +855,56 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot: usize) {
// 1. run queue is still empty
// 2. no live actors (nothing parked, nothing mid-finalize)
// 3. no outstanding IO
//
// `io_out` was read from the io mutex BEFORE we took `shared`.
// That ordering is what keeps the now-split check correct: an
// IO completion can only resurrect an actor by going through
// the drain path, which marks the actor Runnable under THIS
// shared lock. So if we read io first and saw `io_out == 0`,
// and then under `shared` we see the queue empty and no live
// actors, no completion that we missed could have made an actor
// live without also having taken `shared` after us — and it
// didn't, because we hold it now and the queue is empty. A
// completion landing after this point finds no live actor to
// wake. Reading io AFTER shared would be the buggy order: an op
// could complete in the gap and be lost.
//
// Pending timers are deliberately NOT a condition: a timer can
// only ever do something useful by waking a live actor, so once
// `live == 0` every remaining timer entry is orphaned (e.g. a
// sleeping actor that was cooperatively cancelled out of its
// sleep, leaving its deadline behind). Such entries must not
// keep the runtime alive until they fire — that could be an
// arbitrarily long hang. They are dropped here on the way out.
let next = s.timers.peek_deadline();
let (out, fd) = match s.io.as_ref() {
Some(io) => (
io.outstanding + io.waiters.len() as u32,
Some(io.wake_fd()),
),
None => (0, None),
};
// arbitrarily long hang. They are dropped on the way out.
let live = s.slots.iter().filter(|slot| slot.actor.is_some()).count();
let all_clear = s.run_queue.is_empty() && live == 0 && out == 0;
let all_clear = s.run_queue.is_empty() && live == 0 && io_out == 0;
if all_clear {
s.timers.clear();
PopResult::AllDone
} else {
PopResult::Idle { next_deadline: next, io_outstanding: out, wake_fd: fd }
PopResult::Idle {
next_deadline: None, // filled in below from the timers lock
io_outstanding: io_out,
wake_fd: io_fd,
}
}
}
});
// If we're going idle, consult the timer heap for the soonest deadline
// (separate lock, taken after shared is released). If we decided
// AllDone, drop any orphaned timers on the way out.
let pop = match pop {
PopResult::Idle { next_deadline: _, io_outstanding, wake_fd } => {
let next = inner.timers.lock().unwrap().peek_deadline();
PopResult::Idle { next_deadline: next, io_outstanding, wake_fd }
}
PopResult::AllDone => {
inner.timers.lock().unwrap().clear();
PopResult::AllDone
}
other => other,
};
let (pid, sp, stop_flag, first_closure) = match pop {
PopResult::Work(pid, sp, stop, closure) => {
crate::te!(crate::trace::Event::Dequeue(pid));
src/scheduler.rs
+15 -22
View File
@@ -168,13 +168,8 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa
slot.links.clear();
slot.pending_unpark = false;
slot.pending_io_result = None;
slot.pending_closure = Some(Box::new(f) as crate::runtime::Closure);
s.run_queue.push_back(pid);
// Grow the closures vec to cover this slot index, then store.
let idx = idx as usize;
if s.pending_closures.len() <= idx {
s.pending_closures.resize_with(idx + 1, || None);
}
s.pending_closures[idx] = Some(Box::new(f) as crate::runtime::Closure);
crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid });
crate::te!(crate::trace::Event::Enqueue(pid));
pid
@@ -297,7 +292,7 @@ pub fn sleep(duration: std::time::Duration) {
let me = current_pid().expect("sleep() called outside an actor");
let _np = NoPreempt::enter();
let deadline = crate::timer::deadline_from_now(duration);
with_runtime(|inner| inner.with_shared(|s| s.timers.insert_sleep(deadline, me)));
with_runtime(|inner| inner.timers.lock().unwrap().insert_sleep(deadline, me));
park_current();
}
@@ -308,13 +303,11 @@ pub fn insert_wait_timer(
wait_seq: u64,
) {
with_runtime(|inner| {
inner.with_shared(|s| {
s.timers.insert(
deadline,
pid,
crate::timer::Reason::WaitTimeout { target, wait_seq },
);
})
inner.timers.lock().unwrap().insert(
deadline,
pid,
crate::timer::Reason::WaitTimeout { target, wait_seq },
);
});
}
@@ -334,10 +327,10 @@ where
});
{
let _np = NoPreempt::enter();
with_runtime(|inner| inner.with_shared(|s| {
let io = s.io.as_mut().expect("io thread not started");
io.submit(me, work);
}));
with_runtime(|inner| {
let mut io = inner.io.lock().unwrap();
io.as_mut().expect("io thread not started").submit(me, work);
});
park_current();
}
let result = with_runtime(|inner| inner.with_shared(|s| {
@@ -364,10 +357,10 @@ pub fn wait_writable(fd: std::os::fd::RawFd) -> std::io::Result<()> {
fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::Result<()> {
let me = current_pid().expect("wait_*() called outside an actor");
let _np = NoPreempt::enter();
with_runtime(|inner| inner.with_shared(|s| {
let io = s.io.as_mut().expect("io thread not started");
io.epoll_register(fd, me, readable, writable)
}))?;
with_runtime(|inner| {
let mut io = inner.io.lock().unwrap();
io.as_mut().expect("io thread not started").epoll_register(fd, me, readable, writable)
})?;
park_current();
Ok(())
}