diff --git a/src/actor.rs b/src/actor.rs index 38bb7f7..96c9e9f 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -93,8 +93,10 @@ pub fn take_last_outcome() -> Option { /// unwinding to cross the boundary, but `catch_unwind` here means unwinding /// never actually does. pub extern "C-unwind" fn trampoline() { - let b = CURRENT_ACTOR_BOX.with(|c| c.borrow_mut().take()) - .expect("trampoline entered without a closure set"); + let b = match CURRENT_ACTOR_BOX.with(|c| c.borrow_mut().take()) { + Some(b) => b, + None => panic!("smarm: trampoline entered without a closure set (core corrupt)"), + }; let outcome = match panic::catch_unwind(panic::AssertUnwindSafe(b)) { Ok(()) => Outcome::Exit, diff --git a/src/context.rs b/src/context.rs index cf0ea86..cefc245 100644 --- a/src/context.rs +++ b/src/context.rs @@ -94,10 +94,28 @@ unsafe extern "C" fn switch_to_actor_asm() { } /// Resume the actor whose sp is in `ACTOR_SP`. Returns when the actor yields. +/// +/// # Safety +/// +/// The caller must be running on a scheduler thread with a valid actor stack +/// pointer installed in `ACTOR_SP` — either by `init_actor_stack` (first +/// resume) or by a prior `switch_to_scheduler` (subsequent resumes). Resuming +/// with an unset or stale `ACTOR_SP` transfers control to an arbitrary address. +/// Must not be called from within an actor (only the scheduler side may resume). pub unsafe fn switch_to_actor() { unsafe { switch_to_actor_asm() }; } +/// Yield from the running actor back to its scheduler thread. Returns when the +/// actor is next resumed via [`switch_to_actor`]. +/// +/// # Safety +/// +/// The caller must be running on an actor stack that was entered through +/// [`switch_to_actor`], so that `SCHEDULER_SP` holds the live saved stack +/// pointer of the scheduler side. Calling this from the scheduler thread, or +/// before any actor has been resumed, transfers control to an arbitrary +/// address. #[unsafe(naked)] pub unsafe extern "C" fn switch_to_scheduler() { core::arch::naked_asm!( diff --git a/src/io.rs b/src/io.rs index d4513cd..4e8a82a 100644 --- a/src/io.rs +++ b/src/io.rs @@ -240,15 +240,18 @@ impl IoThread { self.outstanding += 1; // Send can only fail if the pool has hung up, which only happens // on shutdown. submit during shutdown is a bug. - self.tx - .send(Request { pid, epoch, work }) - .expect("io pool hung up unexpectedly"); + if self.tx.send(Request { pid, epoch, work }).is_err() { + panic!("smarm: io pool hung up unexpectedly (submit during shutdown)"); + } } /// Drain every available completion. Caller (the scheduler) routes the /// results and updates `outstanding` / `waiters` accordingly. pub fn drain_completions(&mut self) -> Vec { - let mut q = self.completions.lock().unwrap(); + let mut q = match self.completions.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: io completions lock poisoned (core corrupt): {e}"), + }; let mut out = Vec::with_capacity(q.len()); while let Some(c) = q.pop_front() { out.push(c); @@ -389,10 +392,10 @@ fn pool_loop( Ok(r) => r, Err(payload) => Err(payload), }; - completions - .lock() - .unwrap() - .push_back(Completion::Blocking { pid, epoch, result }); + match completions.lock() { + Ok(mut g) => g.push_back(Completion::Blocking { pid, epoch, result }), + Err(e) => panic!("smarm: io completions lock poisoned (core corrupt): {e}"), + } wake_scheduler(wake_write); } } @@ -435,7 +438,10 @@ fn epoll_loop( let mut shutdown_requested = false; let mut pushed_any = false; { - let mut q = completions.lock().unwrap(); + let mut q = match completions.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: io completions lock poisoned (core corrupt): {e}"), + }; for ev in events.iter().take(n as usize) { if ev.u64 == SHUTDOWN_EPOLL_TOKEN { shutdown_requested = true; diff --git a/src/link.rs b/src/link.rs index 7fed90c..c4242d2 100644 --- a/src/link.rs +++ b/src/link.rs @@ -135,7 +135,10 @@ pub fn link(target: Pid) { if registered_on_target { with_runtime(|inner| { - let slot = inner.slot_at(me).expect("link: own slot vanished"); + let slot = match inner.slot_at(me) { + Some(s) => s, + None => panic!("smarm: link own slot vanished (core corrupt)"), + }; let mut cold = slot.cold.lock(); if !cold.links.contains(&target) { cold.links.push(target); diff --git a/src/mutex.rs b/src/mutex.rs index e5784fa..ea41872 100644 --- a/src/mutex.rs +++ b/src/mutex.rs @@ -64,7 +64,10 @@ impl MutexCore { impl TimerTarget for MutexCore { fn on_timeout(&self, pid: Pid, epoch: u32) { let unpark = { - let mut st = self.state.lock().unwrap(); + let mut st = match self.state.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; // Remove from waiters only if still there with matching epoch. // If the lock was already granted (holder == Some(pid)), the // timer fired after the grant — treat as no-op; the actor @@ -72,12 +75,12 @@ impl TimerTarget for MutexCore { if st.holder == Some(pid) { return; } - let pos = st.waiters.iter().position(|w| w.pid == pid && w.epoch == epoch); - if pos.is_some() { - st.waiters.remove(pos.unwrap()); - true - } else { - false + match st.waiters.iter().position(|w| w.pid == pid && w.epoch == epoch) { + Some(pos) => { + st.waiters.remove(pos); + true + } + None => false, } }; if unpark { @@ -105,11 +108,17 @@ impl Mutex { } pub fn set_default_timeout(&self, timeout: Duration) { - self.core.state.lock().unwrap().default_timeout = timeout; + match self.core.state.lock() { + Ok(mut st) => st.default_timeout = timeout, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + } } pub fn lock(&self) -> Result, LockTimeout> { - let timeout = self.core.state.lock().unwrap().default_timeout; + let timeout = match self.core.state.lock() { + Ok(st) => st.default_timeout, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; self.lock_timeout(timeout) } @@ -122,12 +131,21 @@ impl Mutex { // Fast path: nobody holds it. { - let mut st = self.core.state.lock().unwrap(); + let mut st = match self.core.state.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; if st.holder.is_none() { st.holder = Some(me); drop(st); - let value = self.value.lock().unwrap().take() - .expect("Mutex: value missing on free fast path"); + let taken = match self.value.lock() { + Ok(mut g) => g.take(), + Err(e) => panic!("smarm: mutex value lock poisoned (core corrupt): {e}"), + }; + let value = match taken { + Some(v) => v, + None => panic!("smarm: Mutex value missing on free fast path (core corrupt)"), + }; return Ok(MutexGuard { mutex: self, value: Some(value) }); } } @@ -135,7 +153,10 @@ impl Mutex { // Slow path: register as a waiter, set timeout, park. let _np = scheduler::NoPreempt::enter(); let epoch = { - let mut st = self.core.state.lock().unwrap(); + let mut st = match self.core.state.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; // begin_wait is lock-free — legal under the state lock; this // makes the epoch atomic with the registration's visibility to // grants and timeouts. @@ -153,10 +174,19 @@ impl Mutex { // wait (both epoch-stamped; a stop wake unwinds out of // park_current). The one-shot interpretation below is therefore // exhaustive. Are we the holder? - let is_holder = self.core.state.lock().unwrap().holder == Some(me); + let is_holder = match self.core.state.lock() { + Ok(st) => st.holder == Some(me), + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; if is_holder { - let value = self.value.lock().unwrap().take() - .expect("Mutex: value missing after grant"); + let taken = match self.value.lock() { + Ok(mut g) => g.take(), + Err(e) => panic!("smarm: mutex value lock poisoned (core corrupt): {e}"), + }; + let value = match taken { + Some(v) => v, + None => panic!("smarm: Mutex value missing after grant (core corrupt)"), + }; Ok(MutexGuard { mutex: self, value: Some(value) }) } else { Err(LockTimeout) @@ -165,14 +195,23 @@ impl Mutex { pub fn try_lock(&self) -> Option> { let me = crate::actor::current_pid()?; - let mut st = self.core.state.lock().unwrap(); + let mut st = match self.core.state.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; if st.holder.is_some() { return None; } st.holder = Some(me); drop(st); - let value = self.value.lock().unwrap().take() - .expect("Mutex: value missing on try_lock free path"); + let taken = match self.value.lock() { + Ok(mut g) => g.take(), + Err(e) => panic!("smarm: mutex value lock poisoned (core corrupt): {e}"), + }; + let value = match taken { + Some(v) => v, + None => panic!("smarm: Mutex value missing on try_lock free path (core corrupt)"), + }; Some(MutexGuard { mutex: self, value: Some(value) }) } @@ -183,7 +222,10 @@ impl Mutex { // tracking and just grab the value mutex directly. This is safe because // outside the runtime there are no green threads competing. let value = loop { - let v = self.value.lock().unwrap().take(); + let v = match self.value.lock() { + Ok(mut g) => g.take(), + Err(e) => panic!("smarm: mutex value lock poisoned (core corrupt): {e}"), + }; if let Some(v) = v { break v; } std::thread::yield_now(); }; @@ -212,30 +254,51 @@ pub struct MutexGuard<'a, T> { impl std::ops::Deref for MutexGuard<'_, T> { type Target = T; - fn deref(&self) -> &T { self.value.as_ref().expect("MutexGuard: value missing") } + fn deref(&self) -> &T { + match self.value.as_ref() { + Some(v) => v, + None => panic!("smarm: MutexGuard value missing (core corrupt)"), + } + } } impl std::ops::DerefMut for MutexGuard<'_, T> { fn deref_mut(&mut self) -> &mut T { - self.value.as_mut().expect("MutexGuard: value missing") + match self.value.as_mut() { + Some(v) => v, + None => panic!("smarm: MutexGuard value missing (core corrupt)"), + } } } impl std::fmt::Debug for MutexGuard<'_, T> { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let value = match self.value.as_ref() { + Some(v) => v, + None => panic!("smarm: MutexGuard value missing (core corrupt)"), + }; f.debug_tuple("MutexGuard") - .field(self.value.as_ref().expect("MutexGuard: value missing")) + .field(value) .finish() } } impl Drop for MutexGuard<'_, T> { fn drop(&mut self) { - let v = self.value.take().expect("MutexGuard: double drop"); - *self.mutex.value.lock().unwrap() = Some(v); + let v = match self.value.take() { + Some(v) => v, + None => panic!("smarm: MutexGuard double drop (core corrupt)"), + }; + match self.mutex.value.lock() { + Ok(mut g) => *g = Some(v), + Err(e) => panic!("smarm: mutex value lock poisoned (core corrupt): {e}"), + } let next = { - let mut st = self.mutex.core.state.lock().unwrap(); + let mut st = match self.mutex.core.state.lock() { + Ok(g) => g, + Err(e) => panic!("smarm: mutex state lock poisoned (core corrupt): {e}"), + }; match st.waiters.pop_front() { Some(w) => { st.holder = Some(w.pid); diff --git a/src/registry.rs b/src/registry.rs index 47cf40e..ab47e97 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -193,11 +193,12 @@ impl Mailbox { /// legal under a Leaf (Leaf -> Channel). fn clone_sender(&self) -> Option> { let ch = self.channels.get(&TypeId::of::())?; - let tx = ch - .sender - .as_any() - .downcast_ref::>() - .expect("channel keyed by TypeId but downcast to its own type failed — smarm bug"); + let tx = match ch.sender.as_any().downcast_ref::>() { + Some(tx) => tx, + None => panic!( + "smarm: channel keyed by TypeId but downcast to its own type failed (core corrupt)" + ), + }; debug_assert_eq!(ch.msg_type, type_name::(), "msg_type / TypeId disagree"); Some(tx.clone()) } diff --git a/src/run_queue.rs b/src/run_queue.rs index 18a4e1e..4df4bc8 100644 --- a/src/run_queue.rs +++ b/src/run_queue.rs @@ -113,16 +113,32 @@ impl MutexQueue { pub fn push(&self, pid: Pid) { assert_no_preempt(); - self.q.lock().unwrap().push_back(pid); + match self.q.lock() { + Ok(mut g) => g.push_back(pid), + Err(e) => panic!("smarm: run-queue q lock poisoned (core corrupt): {e}"), + } } pub fn pop(&self) -> Option { assert_no_preempt(); - self.q.lock().unwrap().pop_front() + match self.q.lock() { + Ok(mut g) => g.pop_front(), + Err(e) => panic!("smarm: run-queue q lock poisoned (core corrupt): {e}"), + } } pub fn len(&self) -> u64 { - self.q.lock().unwrap().len() as u64 + match self.q.lock() { + Ok(g) => g.len() as u64, + Err(e) => panic!("smarm: run-queue q lock poisoned (core corrupt): {e}"), + } + } + + pub fn is_empty(&self) -> bool { + match self.q.lock() { + Ok(g) => g.is_empty(), + Err(e) => panic!("smarm: run-queue q lock poisoned (core corrupt): {e}"), + } } } @@ -262,6 +278,10 @@ impl MpmcRing { let d = self.dequeue_pos.0.load(Ordering::Relaxed); e.saturating_sub(d) as u64 } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } } // --------------------------------------------------------------------------- @@ -341,6 +361,10 @@ impl StripedRing { pub fn len(&self) -> u64 { self.stripes.iter().map(|s| s.len()).sum() } + + pub fn is_empty(&self) -> bool { + self.stripes.iter().all(|s| s.is_empty()) + } } // --------------------------------------------------------------------------- diff --git a/src/runtime.rs b/src/runtime.rs index b49f076..c4a3891 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -655,6 +655,9 @@ pub(crate) struct RuntimeInner { } impl RuntimeInner { + // Private constructor taking the parsed Config fields one-for-one; a params + // struct would only move the same 8 values across the call boundary. + #[allow(clippy::too_many_arguments)] fn new( thread_count: usize, alloc_interval: u32, @@ -762,7 +765,10 @@ impl RuntimeInner { /// park-epoch and return it. See slot_state.rs for the rules. #[must_use] pub(crate) fn begin_wait(&self, pid: Pid) -> u32 { - let slot = self.slot_at(pid).expect("begin_wait: own slot vanished"); + let slot = match self.slot_at(pid) { + Some(slot) => slot, + None => panic!("begin_wait: own slot vanished: {:?}", pid), + }; slot.word.begin_wait(pid.generation()) } @@ -771,7 +777,10 @@ impl RuntimeInner { /// then eat a notification that already landed. The caller MUST /// re-check its stop flag afterwards — see `StateWord::clear_notify`. pub(crate) fn retire_wait(&self, pid: Pid) { - let slot = self.slot_at(pid).expect("retire_wait: own slot vanished"); + let slot = match self.slot_at(pid) { + Some(slot) => slot, + None => panic!("retire_wait: own slot vanished: {:?}", pid), + }; let _ = slot.word.begin_wait(pid.generation()); slot.word.clear_notify(pid.generation()); } @@ -932,7 +941,14 @@ impl Runtime { self.inner.live_actors.load(Ordering::Acquire), 0, "run() called while previous run still active" ); - *self.inner.io.lock().unwrap() = Some(IoThread::start().expect("failed to start IO thread")); + let io_thread = match IoThread::start() { + Ok(io) => io, + Err(e) => panic!("failed to start IO thread: {e}"), + }; + match self.inner.io.lock() { + Ok(mut io) => *io = Some(io_thread), + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + } // RFC 005: slot counters reset at the START of a run (not the end), // so `stats()` read after `run()` returns reports that run's totals. @@ -977,8 +993,14 @@ impl Runtime { drop(initial_handle); // Tear down IO and clean up for the next run() call. - drop(self.inner.io.lock().unwrap().take()); // joins IO threads - self.inner.timers.lock().unwrap().clear(); + match self.inner.io.lock() { + Ok(mut io) => drop(io.take()), // joins IO threads + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + } + match self.inner.timers.lock() { + Ok(mut timers) => timers.clear(), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } self.inner.next_monitor_id.store(0, Ordering::Relaxed); // Every slot must have come back: any leak here is a runtime bug // (a JoinHandle held across run() is decremented just above). @@ -1156,10 +1178,16 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { Outcome::Stopped => (Outcome::Stopped, Signal::Stopped(pid), DownReason::Stopped), }; - let slot = inner.slot_at(pid).expect("finalize_actor: pid out of range"); + let slot = match inner.slot_at(pid) { + Some(slot) => slot, + None => panic!("finalize_actor: pid out of range: {:?}", pid), + }; let (waiters, monitors, links, actor) = { let mut cold = slot.cold.lock(); - let actor = cold.actor.take().expect("finalize_actor: actor vanished"); + let actor = match cold.actor.take() { + Some(actor) => actor, + None => panic!("finalize_actor: actor vanished"), + }; cold.outcome = Some(joiner_outcome); slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release); // Done is published under the cold lock, so join's @@ -1301,17 +1329,23 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // just to discover there is nothing to drain. The clock is read // only when the timer heap is non-empty. let due = { - let mut t = inner.timers.lock().unwrap(); + let mut t = match inner.timers.lock() { + Ok(t) => t, + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + }; if t.is_empty() { Vec::new() } else { t.pop_due(std::time::Instant::now()) } }; - let completions = inner.io.lock().unwrap() - .as_mut() - .map(|io| io.drain_completions()) - .unwrap_or_default(); + let completions = match inner.io.lock() { + Ok(mut io) => io + .as_mut() + .map(|io| io.drain_completions()) + .unwrap_or_default(), + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; for entry in due { match entry.reason { // A sleep expiry is just an unpark: the protocol handles @@ -1340,8 +1374,15 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { for completion in completions { match completion { crate::io::Completion::Blocking { pid, epoch, result } => { - if let Some(io) = inner.io.lock().unwrap().as_mut() { - io.outstanding = io.outstanding.saturating_sub(1); + match inner.io.lock() { + Ok(mut io) => { + if let Some(io) = io.as_mut() { + io.outstanding = io.outstanding.saturating_sub(1); + } + } + Err(e) => { + panic!("smarm: io lock poisoned (core corrupt): {e}") + } } // Stash the result under the cold lock, then unpark. // The protocol also covers the submit→park window @@ -1363,11 +1404,16 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { crate::io::Completion::FdReady { fd, events: _ } => { // Resolve the parked pid under the io lock, then wake // through the protocol. Lock order: io before all. - let parked = inner.io.lock().unwrap().as_mut().and_then(|io| { - let entry = io.waiters.remove(&fd); - io.epoll_deregister(fd); - entry - }); + let parked = match inner.io.lock() { + Ok(mut io) => io.as_mut().and_then(|io| { + let entry = io.waiters.remove(&fd); + io.epoll_deregister(fd); + entry + }), + Err(e) => { + panic!("smarm: io lock poisoned (core corrupt): {e}") + } + }; if let Some((pid, epoch)) = parked { inner.unpark_at(pid, epoch); } @@ -1411,9 +1457,15 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // Read IO liveness BEFORE the queue lock (phase-1 ordering: a // completion resurrects an actor only via the drain path, whose // enqueue would be visible under the queue lock we take next). - 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 (io_out, io_fd) = { + let io = match inner.io.lock() { + Ok(io) => io, + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; + match io.as_ref() { + Some(io) => (io.outstanding + io.waiters.len() as u32, Some(io.wake_fd())), + None => (0, None), + } }; stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed); @@ -1457,7 +1509,10 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // Remaining timer entries are orphaned (no live actor can be // woken by them — e.g. a sleeper cancelled out of its sleep); // they must not keep the runtime alive. Drop them on the way out. - inner.timers.lock().unwrap().clear(); + match inner.timers.lock() { + Ok(mut timers) => timers.clear(), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } // Terminal wake: a sibling scheduler may be blocked in its // idle wait on a snapshot that is now terminally stale — an // orphaned long deadline (it would sleep it out in full) or @@ -1466,8 +1521,13 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // no completion, so nothing else writes the wake pipe). // One byte wakes every poller; each re-runs the verdict, // reaches AllDone itself, and re-wakes — idempotent. - if let Some(io) = inner.io.lock().unwrap().as_ref() { - io.wake(); + match inner.io.lock() { + Ok(io) => { + if let Some(io) = io.as_ref() { + io.wake(); + } + } + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), } return; } @@ -1481,7 +1541,10 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { Pop::Idle { io_outstanding, wake_fd } => { // Something is still in flight. Sleep on the appropriate // source to avoid hammering the queue mutex; retry on wake. - let next_deadline = inner.timers.lock().unwrap().peek_deadline(); + let next_deadline = match inner.timers.lock() { + Ok(timers) => timers.peek_deadline(), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + }; match (next_deadline, wake_fd) { (Some(deadline), fd_opt) => { let now = std::time::Instant::now(); diff --git a/src/scheduler.rs b/src/scheduler.rs index b782c11..cb5ab32 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -39,7 +39,10 @@ pub(crate) fn with_runtime(f: impl FnOnce(&Arc) -> R) -> R { let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); let result = RUNTIME.with(|r| { let b = r.borrow(); - let inner = b.as_ref().expect("smarm: not inside Runtime::run()"); + let inner = match b.as_ref() { + Some(inner) => inner, + None => panic!("smarm: not inside Runtime::run()"), + }; f(inner) }); crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev)); @@ -51,7 +54,7 @@ pub(crate) fn with_runtime(f: impl FnOnce(&Arc) -> R) -> R { /// Same preemption gate as [`with_runtime`]; same reasons. pub(crate) fn try_with_runtime(f: impl FnOnce(&Arc) -> R) -> Option { let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); - let result = RUNTIME.with(|r| r.borrow().as_ref().map(|inner| f(inner))); + let result = RUNTIME.with(|r| r.borrow().as_ref().map(f)); crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev)); result } @@ -76,15 +79,20 @@ impl JoinHandle { pub fn join(mut self) -> Result<(), JoinError> { use crate::actor::Outcome; - let me = current_pid().expect("join() called outside an actor"); + let me = match current_pid() { + Some(pid) => pid, + None => panic!("join() called outside an actor"), + }; loop { // Check-Done-or-register-waiter is atomic under the target's cold // lock; finalize publishes Done and takes the waiter list under // the same lock, so we either see the outcome or are woken. let outcome = with_runtime(|inner| { - let slot = inner.slot_at(self.pid) - .expect("join: pid index out of range"); + let slot = match inner.slot_at(self.pid) { + Some(slot) => slot, + None => panic!("join: pid index out of range: {:?}", self.pid), + }; let mut cold = slot.cold.lock(); match slot.status_for(self.pid) { // Our outstanding handle pins the slot: it cannot be @@ -93,7 +101,10 @@ impl JoinHandle { panic!("join: target slot has been reused") } crate::slot_state::Status::Done => { - Some(cold.outcome.take().expect("Done slot must have outcome")) + Some(match cold.outcome.take() { + Some(outcome) => outcome, + None => panic!("Done slot must have outcome"), + }) } crate::slot_state::Status::Live => { // begin_wait is lock-free, legal under the cold lock; @@ -181,8 +192,10 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> // syscall and no allocation ever stalls another scheduler thread. let stack = with_runtime(|inner| inner.stack_pool.lock().pop()) .unwrap_or_else(|| { - crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE) - .expect("stack allocation failed") + match crate::stack::Stack::new(crate::runtime::ACTOR_STACK_SIZE) { + Ok(stack) => stack, + Err(e) => panic!("stack allocation failed: {e}"), + } }); let sp = init_actor_stack(stack.top(), crate::actor::trampoline); let closure: crate::runtime::Closure = Box::new(f); @@ -225,7 +238,10 @@ pub fn spawn_addr( use crate::context::init_actor_stack; pub fn self_pid() -> Pid { - current_pid().expect("self_pid() called outside an actor") + match current_pid() { + Some(pid) => pid, + None => panic!("self_pid() called outside an actor"), + } } // --------------------------------------------------------------------------- @@ -282,7 +298,10 @@ pub(crate) fn unpark_at(pid: Pid, epoch: u32) { /// waker. Lock-free (one CAS on the own slot word), so it is legal under /// any lock, including a Channel-class lock. pub(crate) fn begin_wait() -> u32 { - let me = current_pid().expect("begin_wait() called outside an actor"); + let me = match current_pid() { + Some(pid) => pid, + None => panic!("begin_wait() called outside an actor"), + }; with_runtime(|inner| inner.begin_wait(me)) } @@ -294,7 +313,10 @@ pub(crate) fn begin_wait() -> u32 { /// self-clean at their wakers' failed CAS; nothing can fault the actor's /// next one-shot park. pub(crate) fn retire_wait() { - let me = current_pid().expect("retire_wait() called outside an actor"); + let me = match current_pid() { + Some(pid) => pid, + None => panic!("retire_wait() called outside an actor"), + }; with_runtime(|inner| inner.retire_wait(me)); // A request_stop that fired before the clear had its notification // eaten, but it set the stop flag first — observe it here and unwind. @@ -365,11 +387,19 @@ impl Drop for NoPreempt { // --------------------------------------------------------------------------- pub fn sleep(duration: std::time::Duration) { - let me = current_pid().expect("sleep() called outside an actor"); + let me = match current_pid() { + Some(pid) => pid, + None => panic!("sleep() called outside an actor"), + }; let _np = NoPreempt::enter(); let epoch = begin_wait(); let deadline = crate::timer::deadline_from_now(duration); - with_runtime(|inner| inner.timers.lock().unwrap().insert_sleep(deadline, me, epoch)); + with_runtime(|inner| { + match inner.timers.lock() { + Ok(mut timers) => timers.insert_sleep(deadline, me, epoch), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } + }); park_current(); } @@ -380,11 +410,14 @@ pub fn insert_wait_timer( epoch: u32, ) { with_runtime(|inner| { - inner.timers.lock().unwrap().insert( - deadline, - pid, - crate::timer::Reason::WaitTimeout { target, epoch }, - ); + match inner.timers.lock() { + Ok(mut timers) => timers.insert( + deadline, + pid, + crate::timer::Reason::WaitTimeout { target, epoch }, + ), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } }); } @@ -412,7 +445,12 @@ pub fn send_after( let fire = Box::new(move || { let _ = crate::registry::send_to(dest, msg); }); - with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, dest.erase(), fire)) + with_runtime(|inner| { + match inner.timers.lock() { + Ok(mut timers) => timers.insert_send(deadline, dest.erase(), fire), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } + }) } /// Deliver `msg` to whichever actor holds the name `dest` at fire time @@ -429,7 +467,12 @@ pub fn send_after_named( let fire = Box::new(move || { let _ = crate::registry::send(dest, msg); }); - with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, armed_by, fire)) + with_runtime(|inner| { + match inner.timers.lock() { + Ok(mut timers) => timers.insert_send(deadline, armed_by, fire), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } + }) } /// Deliver `msg` onto a channel the caller owns after `after`, rather than to a @@ -453,14 +496,24 @@ pub(crate) fn send_after_to( let fire = Box::new(move || { let _ = tx.send(msg); }); - with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, armed_by, fire)) + with_runtime(|inner| { + match inner.timers.lock() { + Ok(mut timers) => timers.insert_send(deadline, armed_by, fire), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } + }) } /// Cancel a timer armed by [`send_after`] / [`send_after_named`]. Returns /// `true` if it was still pending (delivery now prevented), `false` if it had /// already fired or been cancelled. pub fn cancel_timer(id: crate::timer::TimerId) -> bool { - with_runtime(|inner| inner.timers.lock().unwrap().cancel(id)) + with_runtime(|inner| { + match inner.timers.lock() { + Ok(mut timers) => timers.cancel(id), + Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"), + } + }) } // --------------------------------------------------------------------------- @@ -472,7 +525,10 @@ where F: FnOnce() -> T + Send + 'static, T: Send + 'static, { - let me = current_pid().expect("block_on_io() called outside an actor"); + let me = match current_pid() { + Some(pid) => pid, + None => panic!("block_on_io() called outside an actor"), + }; let work: Box crate::io::IoResult + Send> = Box::new(move || { let v: T = f(); Ok(Box::new(v) as Box) @@ -481,24 +537,37 @@ where let _np = NoPreempt::enter(); let epoch = begin_wait(); with_runtime(|inner| { - let mut io = inner.io.lock().unwrap(); - io.as_mut().expect("io thread not started").submit(me, epoch, work); + let mut io = match inner.io.lock() { + Ok(io) => io, + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; + match io.as_mut() { + Some(io) => io.submit(me, epoch, work), + None => panic!("io thread not started"), + } }); park_current(); } let result = with_runtime(|inner| { - let slot = inner.slot_at(me).expect("block_on_io: own slot vanished"); + let slot = match inner.slot_at(me) { + Some(slot) => slot, + None => panic!("block_on_io: own slot vanished"), + }; let mut cold = slot.cold.lock(); debug_assert_eq!( slot.generation(), me.generation(), "block_on_io: own slot reused mid-park" ); - cold.pending_io_result - .take() - .expect("block_on_io: resumed without a result") + match cold.pending_io_result.take() { + Some(result) => result, + None => panic!("block_on_io: resumed without a result"), + } }); match result { - Ok(any) => *any.downcast::().expect("block_on_io: type mismatch"), + Ok(any) => match any.downcast::() { + Ok(typed) => *typed, + Err(_) => panic!("block_on_io: type mismatch"), + }, Err(payload) => std::panic::resume_unwind(payload), } } @@ -512,12 +581,21 @@ 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 me = match current_pid() { + Some(pid) => pid, + None => panic!("wait_*() called outside an actor"), + }; let _np = NoPreempt::enter(); let epoch = begin_wait(); with_runtime(|inner| { - let mut io = inner.io.lock().unwrap(); - io.as_mut().expect("io thread not started").epoll_register(fd, me, epoch, readable, writable) + let mut io = match inner.io.lock() { + Ok(io) => io, + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; + match io.as_mut() { + Some(io) => io.epoll_register(fd, me, epoch, readable, writable), + None => panic!("io thread not started"), + } })?; // If a terminal stop unwinds us out of the park below, the registration @@ -535,7 +613,10 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R impl Drop for Dereg { fn drop(&mut self) { with_runtime(|inner| { - let mut io = inner.io.lock().unwrap(); + let mut io = match inner.io.lock() { + Ok(io) => io, + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; if let Some(io) = io.as_mut() { if io.waiters.get(&self.fd) == Some(&(self.me, self.epoch)) { io.waiters.remove(&self.fd); @@ -597,10 +678,16 @@ impl crate::channel::Selectable for FdArm { return Ok(false); } with_runtime(|inner| { - let mut io = inner.io.lock().unwrap(); - io.as_mut() - .expect("io thread not started") - .epoll_register(self.fd, pid, epoch, self.readable, self.writable) + let mut io = match inner.io.lock() { + Ok(io) => io, + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; + match io.as_mut() { + Some(io) => { + io.epoll_register(self.fd, pid, epoch, self.readable, self.writable) + } + None => panic!("io thread not started"), + } })?; Ok(true) } @@ -621,7 +708,10 @@ impl crate::channel::Selectable for FdArm { /// actor's fresh registration; in that case touch nothing. fn sel_unregister(&self, pid: Pid, epoch: u32) { with_runtime(|inner| { - let mut io = inner.io.lock().unwrap(); + let mut io = match inner.io.lock() { + Ok(io) => io, + Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"), + }; if let Some(io) = io.as_mut() { if io.waiters.get(&self.fd) == Some(&(pid, epoch)) { io.waiters.remove(&self.fd); diff --git a/src/supervisor.rs b/src/supervisor.rs index 41ca911..0f4d142 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -253,7 +253,7 @@ impl OneForOne { .collect(), }; // Stop survivors in reverse start order (highest child index first). - to_stop.sort_unstable_by(|a, b| b.1.cmp(&a.1)); + to_stop.sort_unstable_by_key(|x| std::cmp::Reverse(x.1)); // The set we will restart: the failed child plus every sibling we // are about to stop, restarted in start (ascending index) order. @@ -299,7 +299,7 @@ impl OneForOne { // and this is a no-op; on a cap-trip or mailbox-closed break it tears // the remaining children down deterministically instead of leaking them. let mut survivors: Vec<(Pid, usize)> = by_pid.iter().map(|(p, i)| (*p, *i)).collect(); - survivors.sort_unstable_by(|a, b| b.1.cmp(&a.1)); + survivors.sort_unstable_by_key(|x| std::cmp::Reverse(x.1)); let mut awaiting: Vec = Vec::with_capacity(survivors.len()); for (pid, _) in &survivors { crate::scheduler::request_stop(*pid); diff --git a/src/timer.rs b/src/timer.rs index 1d1c47d..de2ddb2 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -214,7 +214,10 @@ impl Timers { let mut out = Vec::new(); while let Some(r) = self.heap.peek() { if r.0.deadline <= now { - let entry = self.heap.pop().unwrap().0; + let entry = match self.heap.pop() { + Some(e) => e.0, + None => panic!("smarm: timer heap pop after peek returned None (core corrupt)"), + }; if matches!(entry.reason, Reason::Send { .. }) && !self.armed.remove(&entry.seq) { // Cancelled before it came due: discard, do not deliver. continue;