diff --git a/ROADMAP_v0.5.md b/ROADMAP_v0.5.md new file mode 100644 index 0000000..fd07975 --- /dev/null +++ b/ROADMAP_v0.5.md @@ -0,0 +1,109 @@ +# smarm v0.5 — Runtime decomposition & run-path scaling + +Goal: dismantle the single `Mutex` along the run path so the +runtime scales to dozens of cores, with crash isolation hardened so a torn +write or a stray cancellation can never poison shared runtime state. + +Guiding rules established this cycle: +- **Lock count that matters is hot-path locks.** Cold lifecycle paths + (spawn/join/monitor/link/finalize) may keep a small per-slot lock; the run + path (yield/park/unpark/pop/resume) targets zero locks. +- **Lock ordering: io-before-shared, and slot locks are leaves** (never hold + two slot locks at once). Any new lock states its position in this order. +- **No unwind inside a runtime critical section.** The stop sentinel only + fires at lock-free observation points. + +--- + +## Phase 1 — State decomposition: easy peel-offs ✅ DONE +Split independent concerns out of `SharedState` so the global lock guards less. +- [x] `next_monitor_id` → `AtomicU64` on `RuntimeInner` (lock-free id minting). +- [x] `timers` → own `Mutex` (only drain-winner + blocking prims touch it). +- [x] `io` → own `Mutex>` (completion queue already self-locked). +- [x] `pending_closures: Vec` → folded into `Slot::pending_closure` (per-actor data). +- [x] Termination check split: read io liveness *before* `shared`; ordering + argument documented in `schedule_loop`. +- [x] **Poison fix:** gate `check_cancelled()` in `maybe_preempt` behind + `PREEMPTION_ENABLED`, so a cancellation sentinel can never unwind while a + `std::sync::Mutex` (shared/channel) is held. Regression test: + `tests/poison_stop.rs`. + +`SharedState` now holds only: `slots`, `free_list`, `run_queue`, `root_pid`. + +## Phase 2 — Slot table split (NEXT, the big correctness step) +Make slot lookup lock-free and per-slot state independently mutable. +- [ ] Fixed slab: `Box<[SlotCell; max_actors]>`, slots never move → stable + addresses, lock-free index. **Assert on exhaustion** with a panic message + naming `Config::max_actors(n)` as the fix. Default `max_actors = 16_384`. + (Mental note: must become unbounded or configurable-bounded later — see + "Deferred" below. Do not let the fixed cap calcify into an assumption.) +- [ ] Per-slot `generation: AtomicU32` outside the lock for cheap stale-PID + rejection. +- [ ] Per-slot CAS state machine replacing `state` + `pending_unpark`: + `Parked / Queued / Running / RunningNotified / Done` in one `AtomicU8`. + `unpark` = CAS Parked→Queued (then push) or Running→RunningNotified; + park-return = CAS Running→Parked, else re-queue. Eliminates the + lost-wakeup bool by turning the race window into a state. +- [ ] `sp` → relaxed `AtomicUsize` (happens-before carried by queue + release-push / acquire-pop). +- [ ] Per-slot raw non-poisoning mutex (spin-then-futex, Linux; embedded- + friendlier than `std::sync::Mutex`) for the cold collections: `waiters`, + `monitors`, `links`, `supervisor_channel`, `outcome`, `pending_io_result`, + `outstanding_handles`. Guard enters `NoPreempt`. +- [ ] Free list → its own structure (lock-free Treiber stack or striped). +- [ ] `finalize_actor` link cascade: lock peers one at a time (leaf rule); + write the acyclicity argument next to it. +- [ ] `live_actors` / `io_outstanding` → atomics; termination = both zero + + queue empty. Finalize decrements `live` strictly *after* all wakeups it + produced are enqueued — document this ordering as the correctness crux. + +## Phase 3 — Pluggable run queue + bench shootout +- [ ] Extract `RunQueue` behind a `type RunQueue = ...` alias selected at + compile time by mutually-exclusive features, with a `compile_error!` + guard if zero or >1 is set. No runtime dispatch. + - `rq-mutex` — `Mutex`, the control/baseline. + - `rq-mpmc` — single hand-rolled Vyukov bounded MPMC ring + (per-cell seq numbers). Strict FIFO; "one hot cache line". + - `rq-striped` — M Vyukov rings, fetch-add ticket distribution. Relaxed + FIFO, reordering bounded by ~M. Predicted winner @20c. +- [ ] Bounded rings are sound because the slab cap + at-most-once-enqueued + invariant bound occupancy ≤ `max_actors`; size Σcapacity > `max_actors` + so push is infallible (probe next stripe when full; terminates). +- [ ] All hand-rolled, dependency-free (portability incl. possible embedded). + +## Phase 4 — Bench harness +- [ ] Raw-structure microbench: N threads, push/pop throughput vs thread count, + sweeping producer:consumer ratios. Isolates the data structure. +- [ ] Runtime-level: yield-storm, ping-pong-pairs, spawn-storm, all sweeping + scheduler count. Reuses existing `benches/` harness style. +- [ ] Driver script rebuilding per `rq-*` feature to compare variants in one go. +- [ ] Sandbox validates correctness via oversubscribed `Config::exact(N)` on 1 + core; real contention numbers come from the 20-core box. + +## Phase 5 — Safety hardening & model checking +- [ ] `loom` (dev-dependency only, x86 Linux) model tests for the slot state + machine and each ring variant. +- [ ] `NoPreempt` / no-unwind-under-lock audit across all internal critical + sections; assert the invariant in debug builds where feasible. + +--- + +## Fast follow (post-v0.5, written down so it isn't lost) +- **Channel mutex migration.** `channel::Inner` is `Arc>` of the + same poison class as the old shared lock; `recv_match` even runs a user + predicate under it. The Phase-1 `check_cancelled` gating already removes the + unwind *source* globally, so channels are safe today — but migrating the + channel to the raw non-poisoning mutex (Phase 2) is a clean follow-up rather + than bundled into the run-path rework. + +## Deferred / explicitly out of scope for v0.5 +- **Unbounded / configurable-bounded actor count.** v0.5 ships a fixed slab + with a loud assert. Revisit with a segmented slab (array of + `AtomicPtr`, doubling segment sizes, append-only) once we actually + hit the cap or a workload demands it. +- **Idle-wakeup eventcount.** Idle scheduler threads keep the current 100µs + poll-sleep. A futex-based eventcount is a later optimization if benches show + idle latency matters. +- **User-facing safe data structures** (ArcSwap-style cells, structurally- + shared persistent maps). Context for the runtime work, not in scope here. +- **IO fd hygiene on actor death** (pre-existing v0.2 TODO in `io.rs`). diff --git a/src/monitor.rs b/src/monitor.rs index c5aee8a..ce473fd 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -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 { diff --git a/src/preempt.rs b/src/preempt.rs index e8b88c1..da03d1e 100644 --- a/src/preempt.rs +++ b/src/preempt.rs @@ -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 diff --git a/src/runtime.rs b/src/runtime.rs index ee484bf..8aa26f2 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -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>` 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, } 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, pub(crate) run_queue: VecDeque, pub(crate) root_pid: Option, - pub(crate) timers: Timers, - pub(crate) io: Option, - /// 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>, - /// 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 { - 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, + /// 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, + /// 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>, + /// 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, 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, 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, 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, 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, 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, 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)); diff --git a/src/scheduler.rs b/src/scheduler.rs index 9011fae..56e6147 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -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(()) } diff --git a/tests/poison_stop.rs b/tests/poison_stop.rs new file mode 100644 index 0000000..316730c --- /dev/null +++ b/tests/poison_stop.rs @@ -0,0 +1,42 @@ +//! Regression: request_stop racing an alloc-under-lock must not poison any +//! runtime mutex. Before the fix, check_cancelled() fired regardless of +//! PREEMPTION_ENABLED, so a sentinel unwind could trigger inside with_shared +//! (which allocates: run_queue.push_back, waiters.push, etc), poisoning the +//! shared mutex and cascading lock().unwrap() panics. + +use smarm::runtime::{init, Config}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +#[test] +fn stop_storm_does_not_poison_runtime() { + let rt = init(Config::exact(4)); + let completed = Arc::new(AtomicUsize::new(0)); + let c = completed.clone(); + rt.run(move || { + // Spawn many short actors that allocate + yield heavily (hammering + // with_shared), and request_stop each one mid-flight from siblings. + let mut handles = Vec::new(); + for _ in 0..200 { + let h = smarm::spawn(|| { + for _ in 0..50 { + let _v: Vec = Vec::with_capacity(64); // alloc → maybe_preempt + smarm::yield_now(); + } + }); + handles.push(h); + } + // Cancel half of them; the others run to completion. If any lock got + // poisoned the runtime would panic on a subsequent lock().unwrap(). + for (i, h) in handles.iter().enumerate() { + if i % 2 == 0 { + smarm::request_stop(h.pid()); + } + } + for h in handles { + let _ = h.join(); + } + c.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!(completed.load(Ordering::SeqCst), 1, "root completed cleanly"); +}