Compare commits

..
8 Commits
Author SHA1 Message Date
smarm-agent 3e9c33377c gen_statem: disambiguate module/macro intra-doc links 2026-06-20 19:51:33 +00:00
smarm-agent e54c67c431 supervisor: rewrite docs for users; relocate internals to items 2026-06-20 19:51:32 +00:00
smarm-agent 3e321eaaf3 pg: rewrite docs for users; relocate internals to items
Reframe the module doc in the gen_server house style: lead with what a
process group is and when to reach for one, the auto-eviction-on-death
behaviour, groups-vs-registry, and a running-context note. Keep the
runnable example; add an ignored dispatch/worker-pool example.

Move implementation reasoning to where a maintainer stands: lock
discipline onto the ProcessGroups store, the eager-cleanup rationale onto
reap_group, the join finalize-race detail into join's body comment.

Drop all RFC references and the stray phase marker, and lead the public
read/select fn docs with what the caller gets. Demote the pub(crate)
assert_type intra-doc link in pick_as to plain code, clearing a
pre-existing broken-link warning.
2026-06-20 18:51:00 +00:00
smarm-agent c415f14dd0 ci: deny unwrap_used/expect_used on the library target
Add [lints.clippy] unwrap_used = "deny", expect_used = "deny" plus a tracked
pre-commit hook running `cargo clippy --lib -- -D warnings`. Library code may
not hide a panic behind unwrap/expect; panic!/unreachable! stay un-linted as
the explicit sanctioned form. Gate is the library target only — tests and
examples are not gated. A fresh clone must run
`git config core.hooksPath .githooks` to enable the hook.
2026-06-20 17:47:44 +00:00
smarm-agent 33177a0c48 library + trace: rewrite panic sites as explicit match+panic
Apply the same explicit match+panic shape to the library layer (channel,
gen_server, gen_statem). Extend it to the smarm-trace-gated code that the
default `cargo clippy --lib` does not see: the GLOBAL lock-poison sites in
trace.rs and the current_pid sites inside te!() in channel.rs. Keep the
current_pid match inside the te!() argument so non-trace builds evaluate
nothing extra on the recv-wake hot path. const-init the trace thread-local.
2026-06-20 17:47:39 +00:00
smarm-agent a875fa8285 core: rewrite panic sites as explicit match+panic
Replace implicit unwrap()/expect() in the lock-ordered core with explicit
match arms. Lock-poison sites use one uniform message
("smarm: <lock> lock poisoned (core corrupt): {e}"); invariant sites panic
with a descriptive message naming the violated invariant. No behaviour
change: each rewrite preserves the prior panic-on-bad-arm semantics. Also
clears the accompanying clippy hygiene in these files (redundant_closure,
len_without_is_empty, too_many_arguments, unnecessary_sort_by,
missing_safety_doc, nonminimal_bool/unnecessary_unwrap).
2026-06-20 17:47:33 +00:00
smarm-agent 531571bfa5 gen_statem: postpone events for replay after a transition
A `=> postpone` row (cast/call/info) defers the current event untouched,
to be replayed after the next real transition. `handle` is now two-phase:
a borrow-only postpone pre-pass that hands the event back as
`Step::Postponed(ev)`, then the existing consuming `match (state, event)`.
The loop owns a FIFO queue, drained in the new state ahead of further
intake; a replayed event may postpone again. A postponed `call` keeps its
Reply, so a later state answers it.

`handle` returns `Step` (Postponed / Transitioned / Stayed) so the loop
can see both deferral and transition without reading the state cell.
`Resolution::Postpone` is removed: postpone is a pre-dispatch routing
decision, not a consuming-dispatch outcome.
2026-06-20 13:02:15 +00:00
smarm-agent acf67fef06 gen_statem: state and named timeouts, info events
Add the two timeout flavours, both surfacing as ordinary events matched
in on-state arms:

- cx.state_timeout(d): fires a state_timeout event after d in the current
  state, auto-reset by the loop on every real transition.
- cx.timeout(name, d): fires a timeout(name) event after d, surviving state
  changes, keyed by name, with cx.cancel_timeout(name).

Both ride the existing timer min-heap via send_after_to onto a new per-loop
system channel, selected above the inbox so a fire can't be starved by inbox
traffic. A local-id stamp on each fire lets a reset/cancel that loses the race
discard a stale fire. The macro grows an info: clause and folds three internal
Ev variants (Info, StateTimeout, Timeout) alongside cast/call, with new row
keywords info / state_timeout / timeout. Unmatched info silently drops (the
gen_server default); state/named timeouts have no default, so a state that can
see one must handle it or the match is non-exhaustive.

Rename the hand-written expansion-target example fused -> expanded, retire the
deprecated Switch demo machine (its round-trip / enter / panic-down coverage
moves onto the timer machine), and refresh the macro docs to the door machine.

cargo build --all-targets warning-free; cargo test green.
2026-06-20 12:22:45 +00:00
21 changed files with 1785 additions and 451 deletions
+8
View File
@@ -0,0 +1,8 @@
#!/bin/sh
# smarm pre-commit gate: clippy the library (src/) with warnings as errors.
# unwrap_used / expect_used are denied (Cargo.toml [lints.clippy]): library
# code must not hide a panic behind unwrap/expect. Tests/examples are not gated.
set -eu
[ -f "$HOME/.cargo/env" ] && . "$HOME/.cargo/env"
cd "$(git rev-parse --show-toplevel)"
cargo clippy --lib -- -D warnings
+9
View File
@@ -7,6 +7,15 @@ rust-version = "1.95"
[lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = ["cfg(loom)"] }
[lints.clippy]
# Library code must never hide a panic behind unwrap/expect. Both are denied; an
# intentional panic is written explicitly as `match { Err(e) => panic!(..) }`.
# panic!/unreachable! are deliberately left un-linted as the blessed explicit
# form. Enforced on the library target only (`cargo clippy --lib`); tests and
# examples unwrap freely and are not gated.
unwrap_used = "deny"
expect_used = "deny"
[features]
default = ["rq-mutex"]
smarm-trace = []
@@ -31,12 +31,12 @@
//! Default build is clean and runs. A BREAK-CASE MENU at the bottom documents
//! how to make each of the four guarantees fire.
//!
//! Run: `cargo run --example gen_statem_fused`
//! Run: `cargo run --example gen_statem_expanded`
#![deny(dead_code, unreachable_patterns)]
use smarm::run;
use smarm::gen_statem::{spawn, Cx, Machine, Reply, Resolution, GenStatemRef};
use smarm::gen_statem::{spawn, Cx, Machine, Reply, Resolution, Step, GenStatemRef};
// === user types ============================================================
@@ -50,6 +50,7 @@ enum Door {
struct Data {
enters: u32, // total state entries (incl. initial)
pushes: u32, // times a push closed the door
knocks: u32, // knocks answered (a Locked knock is postponed, then counted)
}
enum Cast {
@@ -57,17 +58,24 @@ enum Cast {
Pull,
Lock,
Unlock(u32), // carries a key
Knock, // counted when the door is reachable; postponed while Locked
}
enum Call {
GetState(Reply<Door>),
GetEnters(Reply<u32>),
GetPushes(Reply<u32>),
GetKnocks(Reply<u32>),
}
enum Ev {
Cast(Cast),
Call(Call),
// The runtime's internal events. `Info` is out-of-band (here unused, so
// `()`); `StateTimeout` / `Timeout` are timer fires the loop feeds back in.
Info(()),
StateTimeout,
Timeout(&'static str),
}
const CODE: u32 = 1234;
@@ -115,34 +123,62 @@ impl DoorSm {
fn start(init: Door) -> GenStatemRef<DoorSm> {
spawn(DoorSm {
state: init,
data: Data { enters: 0, pushes: 0 },
data: Data { enters: 0, pushes: 0, knocks: 0 },
})
}
fn enter(&mut self, _cx: &mut Cx<Ev>) {
fn enter(&mut self, cx: &mut Cx<Ev>) {
self.data.enters += 1;
// A state-timeout: an Open door auto-closes after a quiet window. The
// loop auto-resets it on any transition, so it fires only if the door is
// still Open when it elapses.
if self.state == Door::Open {
cx.state_timeout(std::time::Duration::from_millis(5));
}
}
}
impl Machine for DoorSm {
type Ev = Ev;
fn state_timeout_ev() -> Ev {
Ev::StateTimeout
}
fn timeout_ev(name: &'static str) -> Ev {
Ev::Timeout(name)
}
fn on_start(&mut self, cx: &mut Cx<Ev>) {
self.enter(cx);
}
fn handle(&mut self, ev: Ev, cx: &mut Cx<Ev>) {
fn handle(&mut self, ev: Ev, cx: &mut Cx<Ev>) -> Step<Ev> {
let prev = self.state;
// ---- the transition table -----------------------------------------
// This match is total over (Door, Ev). Read it as the declared graph:
// each `=> To(x)` is an edge, each `=> Unhandled` an explicit refusal.
// ---- phase 1: postpone routing (borrow-only) ----------------------
// A deferred event is handed back untouched for the loop's postpone
// queue; no handler code runs on it. Here: a knock at a locked door.
// (The macro emits this as a `match (state, &ev)` yielding a bool; the
// hand-written form can just test directly.)
if let (Door::Locked, Ev::Cast(Cast::Knock)) = (prev, &ev) {
return Step::Postponed(ev);
}
// ---- phase 2: the consuming transition table ----------------------
// Total over (Door, Ev). Read it as the declared graph: each `=> To(x)`
// is an edge, each `=> Unhandled` an explicit refusal. The postponed
// pair above reappears as `unreachable!` so the match stays total.
let res: Resolution<Door> = match (self.state, ev) {
// --- Open -------------------------------------------------------
(Door::Open, Ev::Cast(Cast::Push)) => {
on_push(&mut self.data);
Resolution::To(Door::Closed)
}
(Door::Open, Ev::Cast(Cast::Knock)) => {
self.data.knocks += 1;
Resolution::To(prev)
}
(Door::Open, Ev::Cast(Cast::Pull | Cast::Lock | Cast::Unlock(_))) => {
Resolution::Unhandled
}
@@ -150,12 +186,20 @@ impl Machine for DoorSm {
// --- Closed -----------------------------------------------------
(Door::Closed, Ev::Cast(Cast::Pull)) => Resolution::To(Door::Open),
(Door::Closed, Ev::Cast(Cast::Lock)) => Resolution::To(Door::Locked),
(Door::Closed, Ev::Cast(Cast::Knock)) => {
self.data.knocks += 1;
Resolution::To(prev)
}
(Door::Closed, Ev::Cast(Cast::Push | Cast::Unlock(_))) => Resolution::Unhandled,
// --- Locked (branching row: handler picks within UnlockOutcome) -
(Door::Locked, Ev::Cast(Cast::Unlock(key))) => {
Resolution::To(on_unlock(key).into())
}
// Routed out in phase 1; listed only to keep this match total.
(Door::Locked, Ev::Cast(Cast::Knock)) => {
unreachable!("postponed event is replayed, not dispatched here")
}
(Door::Locked, Ev::Cast(Cast::Push | Cast::Pull | Cast::Lock)) => {
Resolution::Unhandled
}
@@ -173,17 +217,38 @@ impl Machine for DoorSm {
r.reply(self.data.pushes);
Resolution::To(prev)
}
(_, Ev::Call(Call::GetKnocks(r))) => {
r.reply(self.data.knocks);
Resolution::To(prev)
}
// --- timeouts: an Open door auto-closes; others have none armed --
(Door::Open, Ev::StateTimeout) => Resolution::To(Door::Closed),
(_, Ev::StateTimeout) => Resolution::Unhandled,
(_, Ev::Timeout(name)) => {
// No named timeout is armed in this run; a real handler would
// dispatch on `name`. Acknowledge it to exercise the field.
let _ = name;
Resolution::Unhandled
}
// --- out-of-band info: silent drop (the gen_server default) ------
(_, Ev::Info(_)) => Resolution::Unhandled,
};
// ---- apply the resolution -----------------------------------------
match res {
Resolution::To(s) if s == prev => {} // stay: no enter
Resolution::To(s) if s == prev => Step::Stayed, // stay: no enter
Resolution::To(s) => {
self.state = s; // sole writer of the state cell
cx.__reset_state_timeout(); // auto-reset across transitions
self.enter(cx);
Step::Transitioned
}
Resolution::Unhandled => {
cx.on_unhandled();
Step::Stayed
}
Resolution::Postpone => unreachable!("postpone is not generated yet"),
Resolution::Unhandled => cx.on_unhandled(),
}
}
}
@@ -193,20 +258,29 @@ fn main() {
let door = DoorSm::start(Door::Closed);
door.send(Ev::Cast(Cast::Lock)).unwrap(); // Closed -> Locked
door.send(Ev::Cast(Cast::Knock)).unwrap(); // Locked: postponed (not yet counted)
door.send(Ev::Cast(Cast::Push)).unwrap(); // Locked: Push invalid -> Unhandled
door.send(Ev::Cast(Cast::Unlock(0))).unwrap(); // Locked: wrong key -> stay
door.send(Ev::Cast(Cast::Unlock(CODE))).unwrap(); // Locked -> Closed
door.send(Ev::Cast(Cast::Pull)).unwrap(); // Closed -> Open
door.send(Ev::Cast(Cast::Push)).unwrap(); // Open -> Closed (pushes=1)
door.send(Ev::Cast(Cast::Unlock(0))).unwrap(); // Locked: wrong key -> stay (knock still deferred)
door.send(Ev::Cast(Cast::Unlock(CODE))).unwrap(); // Locked -> Closed; deferred Knock replays here
door.send(Ev::Cast(Cast::Push)).unwrap(); // Closed: Push invalid -> Unhandled
door.send(Ev::Cast(Cast::Pull)).unwrap(); // Closed -> Open (arms 5ms auto-close)
door.send(Ev::Info(())).unwrap(); // out-of-band: silently dropped
// Wait past the auto-close window: the state-timeout fires and the door
// closes itself, with no further input. (`smarm::sleep` parks the actor
// without blocking a worker thread, so the timer wheel keeps turning.)
smarm::sleep(std::time::Duration::from_millis(40));
let st = door.call(|r| Ev::Call(Call::GetState(r))).unwrap();
let enters = door.call(|r| Ev::Call(Call::GetEnters(r))).unwrap();
let pushes = door.call(|r| Ev::Call(Call::GetPushes(r))).unwrap();
let knocks = door.call(|r| Ev::Call(Call::GetKnocks(r))).unwrap();
println!("state={st:?} enters={enters} pushes={pushes}");
assert_eq!(st, Door::Closed);
println!("state={st:?} enters={enters} pushes={pushes} knocks={knocks}");
assert_eq!(st, Door::Closed); // auto-closed by the state-timeout
assert_eq!(enters, 5); // Closed(start) + Locked + Closed + Open + Closed
assert_eq!(pushes, 1);
assert_eq!(pushes, 0); // no push ever closed it this run
assert_eq!(knocks, 1); // the locked-door knock, replayed once unlocked
println!("ok");
});
}
+24 -8
View File
@@ -1,4 +1,4 @@
//! The **same** machine as `examples/gen_statem_fused.rs`, written through the
//! The **same** machine as `examples/gen_statem_expanded.rs`, written through the
//! `gen_statem!` macro. Diff this file against that one to see exactly what the
//! macro buys: every `// ===` section there that was boilerplate (the `Ev`
//! enum, the `DoorSm` struct, `start`, the whole `Machine` impl, the `enter`
@@ -21,7 +21,7 @@ use smarm::gen_statem;
use smarm::run;
use smarm::gen_statem::Reply;
// === user types (identical to gen_statem_fused.rs) =========================
// === user types (identical to gen_statem_expanded.rs) =========================
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum Door {
@@ -33,6 +33,7 @@ enum Door {
struct Data {
enters: u32, // total state entries (incl. initial)
pushes: u32, // times a push closed the door
knocks: u32, // knocks answered (a Locked knock is postponed, then counted)
}
enum Cast {
@@ -40,12 +41,14 @@ enum Cast {
Pull,
Lock,
Unlock(u32), // carries a key
Knock, // counted when the door is reachable; postponed while Locked
}
enum Call {
GetState(Reply<Door>),
GetEnters(Reply<u32>),
GetPushes(Reply<u32>),
GetKnocks(Reply<u32>),
}
const CODE: u32 = 1234;
@@ -87,7 +90,7 @@ fn on_unlock(key: u32) -> UnlockOutcome {
gen_statem! {
machine: DoorSm { state: Door, data: Data };
event: Ev { cast: Cast, call: Call };
event: Ev { cast: Cast, call: Call, info: () };
// You name the bindings the bodies use; the macro can't lend you its own
// `self`/`cx` across macro hygiene. `data` = &mut Data, `prev` = current
@@ -100,15 +103,20 @@ gen_statem! {
on Door::Open => {
cast Cast::Push => { on_push(data); Door::Closed },
cast Cast::Knock => { data.knocks += 1; prev },
cast Cast::Pull | Cast::Lock | Cast::Unlock(_) => unhandled,
}
on Door::Closed => {
cast Cast::Pull => Door::Open,
cast Cast::Lock => Door::Locked,
cast Cast::Knock => { data.knocks += 1; prev },
cast Cast::Push | Cast::Unlock(_) => unhandled,
}
on Door::Locked => {
cast Cast::Unlock(key) => on_unlock(key), // branch -> UnlockOutcome
// A knock at a locked door waits: defer it until the door is reachable,
// where the replay counts it.
cast Cast::Knock => postpone,
cast Cast::Push | Cast::Pull | Cast::Lock => unhandled,
}
@@ -117,35 +125,43 @@ gen_statem! {
call Call::GetState(r) => { r.reply(prev); prev },
call Call::GetEnters(r) => { r.reply(data.enters); prev },
call Call::GetPushes(r) => { r.reply(data.pushes); prev },
call Call::GetKnocks(r) => { r.reply(data.knocks); prev },
// This machine arms no timeouts, so refuse them everywhere. (Info has a
// built-in silent-drop default, so it needs no row.)
state_timeout => unhandled,
timeout _ => unhandled,
}
}
fn main() {
run(|| {
let door = DoorSm::start(Door::Closed, Data { enters: 0, pushes: 0 });
let door = DoorSm::start(Door::Closed, Data { enters: 0, pushes: 0, knocks: 0 });
door.send(Ev::Cast(Cast::Lock)).unwrap(); // Closed -> Locked
door.send(Ev::Cast(Cast::Knock)).unwrap(); // Locked: postponed (not yet counted)
door.send(Ev::Cast(Cast::Push)).unwrap(); // Locked: Push invalid -> Unhandled
door.send(Ev::Cast(Cast::Unlock(0))).unwrap(); // Locked: wrong key -> stay
door.send(Ev::Cast(Cast::Unlock(CODE))).unwrap(); // Locked -> Closed
door.send(Ev::Cast(Cast::Unlock(0))).unwrap(); // Locked: wrong key -> stay (knock still deferred)
door.send(Ev::Cast(Cast::Unlock(CODE))).unwrap(); // Locked -> Closed; deferred Knock replays here
door.send(Ev::Cast(Cast::Pull)).unwrap(); // Closed -> Open
door.send(Ev::Cast(Cast::Push)).unwrap(); // Open -> Closed (pushes=1)
let st = door.call(|r| Ev::Call(Call::GetState(r))).unwrap();
let enters = door.call(|r| Ev::Call(Call::GetEnters(r))).unwrap();
let pushes = door.call(|r| Ev::Call(Call::GetPushes(r))).unwrap();
let knocks = door.call(|r| Ev::Call(Call::GetKnocks(r))).unwrap();
println!("state={st:?} enters={enters} pushes={pushes}");
println!("state={st:?} enters={enters} pushes={pushes} knocks={knocks}");
assert_eq!(st, Door::Closed);
assert_eq!(enters, 5); // Closed(start) + Locked + Closed + Open + Closed
assert_eq!(pushes, 1);
assert_eq!(knocks, 1); // the locked-door knock, replayed once unlocked
println!("ok");
});
}
// ===========================================================================
// BREAK-CASE MENU — the four guarantees, through the macro. Each fires exactly
// as it does in the hand-written gen_statem_fused.rs.
// as it does in the hand-written gen_statem_expanded.rs.
//
// 1. ORPHAN HANDLER (dead_code -> error):
// add `fn on_slam(_d: &mut Data) {}` and don't reference it.
+4 -2
View File
@@ -93,8 +93,10 @@ pub fn take_last_outcome() -> Option<Outcome> {
/// 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,
+54 -18
View File
@@ -176,8 +176,10 @@ impl<T> Receiver<T> {
if g.senders == 0 {
return Err(RecvError);
}
let me = crate::actor::current_pid()
.expect("recv() called outside an actor");
let me = match crate::actor::current_pid() {
Some(me) => me,
None => panic!("smarm: recv() called outside an actor"),
};
debug_assert!(
g.parked_receiver.is_none_or(|(p, _)| p == me),
"channel has more than one receiver"
@@ -191,7 +193,10 @@ impl<T> Receiver<T> {
// Release the lock before parking — the unparker will need it.
crate::scheduler::park_current();
// Woken up — record it before looping to check the queue.
crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
Some(p) => p,
None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
}));
}
}
@@ -222,8 +227,10 @@ impl<T> Receiver<T> {
where
T: Send + 'static,
{
let me = crate::actor::current_pid()
.expect("recv_timeout() called outside an actor");
let me = match crate::actor::current_pid() {
Some(me) => me,
None => panic!("smarm: recv_timeout() called outside an actor"),
};
// Fast path + wait registration, one critical section.
let epoch;
@@ -254,7 +261,10 @@ impl<T> Receiver<T> {
crate::scheduler::insert_wait_timer(deadline, me, target, epoch);
crate::scheduler::park_current();
crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
Some(p) => p,
None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
}));
let mut g = self.inner.lock();
if let Some(v) = g.queue.pop_front() {
crate::preempt::note_message_received();
@@ -283,17 +293,23 @@ impl<T> Receiver<T> {
loop {
{
let mut g = self.inner.lock();
if let Some(i) = g.queue.iter().position(|v| pred(v)) {
if let Some(i) = g.queue.iter().position(&pred) {
// position() found it, so remove() returns Some.
crate::preempt::note_message_received();
return Ok(g.queue.remove(i).unwrap());
let v = match g.queue.remove(i) {
Some(v) => v,
None => panic!("smarm: channel queue.remove after position (logic bug)"),
};
return Ok(v);
}
if g.senders == 0 {
// Closed and nothing queued can ever match.
return Err(RecvError);
}
let me = crate::actor::current_pid()
.expect("recv_match() called outside an actor");
let me = match crate::actor::current_pid() {
Some(me) => me,
None => panic!("smarm: recv_match() called outside an actor"),
};
debug_assert!(
g.parked_receiver.is_none_or(|(p, _)| p == me),
"channel has more than one receiver"
@@ -303,7 +319,10 @@ impl<T> Receiver<T> {
}
// Release the lock before parking — the unparker will need it.
crate::scheduler::park_current();
crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
crate::te!(crate::trace::Event::RecvWake(match crate::actor::current_pid() {
Some(p) => p,
None => panic!("smarm: RecvWake outside an actor (core corrupt)"),
}));
}
}
@@ -316,9 +335,13 @@ impl<T> Receiver<T> {
F: Fn(&T) -> bool,
{
let mut g = self.inner.lock();
if let Some(i) = g.queue.iter().position(|v| pred(v)) {
if let Some(i) = g.queue.iter().position(&pred) {
crate::preempt::note_message_received();
return Ok(Some(g.queue.remove(i).unwrap()));
let v = match g.queue.remove(i) {
Some(v) => v,
None => panic!("smarm: channel queue.remove after position (logic bug)"),
};
return Ok(Some(v));
}
if g.senders == 0 {
return Err(RecvError);
@@ -466,7 +489,10 @@ impl<T> Selectable for Receiver<T> {
/// see [`try_select`] for the fallible form; channel-only selects cannot
/// fail).
pub fn select(arms: &[&dyn Selectable]) -> usize {
try_select(arms).expect("select(): fd arm failed to register (use try_select)")
match try_select(arms) {
Ok(i) => i,
Err(e) => panic!("smarm: select() fd arm failed to register (use try_select): {e}"),
}
}
/// [`select`], fallible: `Err` when an arm fails to register (only fd
@@ -476,7 +502,10 @@ pub fn select(arms: &[&dyn Selectable]) -> usize {
/// one has been unregistered.
pub fn try_select(arms: &[&dyn Selectable]) -> std::io::Result<usize> {
assert!(!arms.is_empty(), "select() on an empty arm list");
let me = crate::actor::current_pid().expect("select() called outside an actor");
let me = match crate::actor::current_pid() {
Some(me) => me,
None => panic!("smarm: select() called outside an actor"),
};
loop {
let epoch = crate::scheduler::begin_wait();
if let Some(i) = register_arms(me, epoch, arms)? {
@@ -626,8 +655,12 @@ pub fn select_timeout(
arms: &[&dyn Selectable],
timeout: std::time::Duration,
) -> Option<usize> {
try_select_timeout(arms, timeout)
.expect("select_timeout(): fd arm failed to register (use try_select_timeout)")
match try_select_timeout(arms, timeout) {
Ok(r) => r,
Err(e) => panic!(
"smarm: select_timeout() fd arm failed to register (use try_select_timeout): {e}"
),
}
}
/// [`select_timeout`], fallible: `Err` when an arm fails to register
@@ -638,7 +671,10 @@ pub fn try_select_timeout(
timeout: std::time::Duration,
) -> std::io::Result<Option<usize>> {
assert!(!arms.is_empty(), "select_timeout() on an empty arm list");
let me = crate::actor::current_pid().expect("select_timeout() called outside an actor");
let me = match crate::actor::current_pid() {
Some(me) => me,
None => panic!("smarm: select_timeout() called outside an actor"),
};
let epoch = crate::scheduler::begin_wait();
if let Some(i) = register_arms(me, epoch, arms)? {
return Ok(Some(i)); // ready now: the timer was never armed
+18
View File
@@ -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!(
+31 -12
View File
@@ -528,7 +528,10 @@ impl<G: GenServer> TimerHandle<G> {
/// [`TimerId`] you can pass to `cancel`. Timer fires arrive before info and
/// inbox messages.
pub fn arm_after(&self, after: Duration, msg: G::Timer) -> TimerId {
let mut reg = self.reg.lock().unwrap();
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"),
};
let local = reg.mint();
// The fire thunk carries the local id so the loop can retire the entry
// on dispatch; `msg` is moved in (no `Clone` needed for one-shots).
@@ -552,7 +555,10 @@ impl<G: GenServer> TimerHandle<G> {
where
G::Timer: Clone,
{
let mut reg = self.reg.lock().unwrap();
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"),
};
let local = reg.mint();
// A live periodic must keep the system arm open so the next tick can be
// re-armed; the first periodic installs the loop's re-arm sender.
@@ -577,7 +583,10 @@ impl<G: GenServer> TimerHandle<G> {
/// stops re-arming: the entry is removed so the loop will not schedule
/// another tick even if the pending one already escaped onto the channel.
pub fn cancel(&self, id: TimerId) -> bool {
let mut reg = self.reg.lock().unwrap();
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"),
};
if let Some(sub) = reg.oneshots.remove(&id) {
return cancel_timer(sub);
}
@@ -587,7 +596,7 @@ impl<G: GenServer> TimerHandle<G> {
reg.rearm_tx = None;
}
debug_assert!(
reg.rearm_tx.is_some() == !reg.periodics.is_empty(),
reg.rearm_tx.is_some() != reg.periodics.is_empty(),
"rearm_tx must be Some iff periodics is non-empty"
);
return beat;
@@ -846,7 +855,10 @@ fn server_loop<G: GenServer>(
impl<G: GenServer> Drop for Terminate<G> {
fn drop(&mut self) {
{
let mut reg = self.1.lock().unwrap();
let mut reg = match self.1.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"),
};
for (_, sub) in reg.oneshots.drain() {
cancel_timer(sub);
}
@@ -995,7 +1007,10 @@ fn server_loop<G: GenServer>(
// The one-shot fired: retire its registry entry so the
// live set tracks only still-pending timers, then
// dispatch.
reg.lock().unwrap().oneshots.remove(&id);
match reg.lock() {
Ok(mut g) => { g.oneshots.remove(&id); }
Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"),
}
guard.0.handle_timer(msg);
reset_idle(&mut idle_deadline);
}
@@ -1006,16 +1021,20 @@ fn server_loop<G: GenServer>(
// so the period is measured from fire-handling and a
// handler that cancels stops the just-armed instance.
let msg = {
let mut g = reg.lock().unwrap();
let mut g = match reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_server reg lock poisoned (core corrupt): {e}"),
};
let r = &mut *g;
if let Some(p) = r.periodics.get_mut(&id) {
let every = p.every;
let msg = (p.make)();
let tx = r
.rearm_tx
.as_ref()
.expect("a live periodic keeps rearm_tx Some")
.clone();
let tx = match r.rearm_tx.as_ref() {
Some(tx) => tx.clone(),
None => panic!(
"smarm: live periodic without rearm_tx (logic bug)"
),
};
p.live = send_after_to(every, tx, Sys::Tick(id));
Some(msg)
} else {
+607 -107
View File
@@ -14,13 +14,13 @@
//! [`on_start`](Machine::on_start), then one [`handle`](Machine::handle) per
//! inbox event — mirroring `gen_server`'s spawn/teardown idioms.
//!
//! The [`gen_statem!`](crate::gen_statem) macro is the authoring surface: it
//! The [`gen_statem!`](crate::gen_statem!) macro is the authoring surface: it
//! *generates* a `Machine` impl from per-state handler blocks. Edge-validity is
//! not a separate check — it falls out of the generated total `match (state,
//! event)` under denied lints (a forgotten or duplicated pair is a compile
//! error), so no proc-macro is needed. See `examples/gen_statem_macro.rs` for
//! the macro form and `examples/gen_statem_fused.rs` for the hand-written shape
//! it expands to.
//! the macro form and `examples/gen_statem_expanded.rs` for the hand-written
//! shape it expands to.
//!
//! ## The unified event
//!
@@ -38,19 +38,51 @@
//! `gen_server` call round-trip, but with the reply handle riding *inside* the
//! user's own event so a handler can answer it.
//!
//! [`Resolution::Postpone`] and the timeout arming on [`Cx`] are part of the
//! type surface but are not yet wired up.
//! ## Timeouts
//!
//! Two timeout flavours, both armed from a handler through [`Cx`] and both
//! surfacing back as ordinary **events** the machine matches in its `on State`
//! arms — unlike `gen_server`, which routes fires to a separate handler:
//!
//! - [`cx.state_timeout(d)`](Cx::state_timeout) fires a `state_timeout` event
//! after `d` *in the current state*, and is auto-reset on any state change.
//! Only one is ever pending; arming again replaces it.
//! - [`cx.timeout(name, d)`](Cx::timeout) fires a `timeout(name)` event after
//! `d`, **survives** state changes, and is keyed by `name` so several can be
//! in flight and each is independently [`cancel_timeout`](Cx::cancel_timeout)led.
//!
//! Both ride the same timer min-heap as `gen_server` (no separate timer
//! mechanism): a fire lands on the loop's own system channel — above the inbox,
//! so a timeout can't be starved by inbox traffic — and the loop turns it into
//! the corresponding internal event and runs it through the normal `handle`
//! dispatch.
//!
//! ## Postpone
//!
//! A `=> postpone` row defers the current event untouched, to be replayed after
//! the next **real transition**. The deferred event — a `cast`, `call`, or
//! `info` — moves whole onto a loop-side FIFO queue; a postponed `call` keeps
//! its [`Reply`] handle and is answered by whichever later state handles the
//! replay. On a transition the queue drains in order through the normal
//! [`handle`](Machine::handle) dispatch in the new state, ahead of any further
//! inbox or timer event; a replayed event may postpone again (it re-queues for
//! the next transition). See the macro docs for the row surface and [`Step`]
//! for how a postpone surfaces to the loop.
use crate::channel::{channel, Receiver, Sender};
use crate::channel::{channel, select, Receiver, Sender};
use crate::pid::Pid;
use crate::scheduler::spawn as spawn_actor;
use crate::scheduler::{cancel_timer, send_after_to, spawn as spawn_actor};
use crate::timer::TimerId;
use std::collections::{HashMap, VecDeque};
use std::marker::PhantomData;
use std::sync::{Arc, Mutex};
use std::time::Duration;
// ---------------------------------------------------------------------------
// Machine
// ---------------------------------------------------------------------------
/// A finite state machine driven by the [`statem`](crate::gen_statem) loop.
/// A finite state machine driven by the [`statem`](mod@crate::gen_statem) loop.
///
/// The implementor owns its current state tag and its persistent data. It is
/// the **sole writer** of the state cell (the loop never touches it): both
@@ -59,28 +91,45 @@ use std::marker::PhantomData;
/// [`handle`](Self::handle) per event.
pub trait Machine: Send + 'static {
/// The single payload this machine's inbox carries — the user's `cast` and
/// `call` enums folded together with the runtime's internal events. See the
/// module docs.
/// `call` enums folded together with the runtime's internal events
/// (`state_timeout`, `timeout(name)`, and an `Info` wrapper). See the module
/// docs.
type Ev: Send + 'static;
/// Wrap a fired state-timeout into this machine's event. The loop calls this
/// when the pending state-timeout fires, then runs the result through
/// [`handle`](Self::handle) like any other event. The macro generates it as
/// `Ev::StateTimeout`.
fn state_timeout_ev() -> Self::Ev;
/// Wrap a fired named timeout into this machine's event, carrying the name
/// it was armed under. The macro generates it as `Ev::Timeout(name)`.
fn timeout_ev(name: &'static str) -> Self::Ev;
/// Runs once inside the actor before the first event. The canonical body
/// runs the `enter` arm for the initial state.
fn on_start(&mut self, cx: &mut Cx<Self::Ev>);
/// React to one event. The body matches `(state, event)`, performs side
/// effects / replies, and ends each arm in a [`Resolution`] — typically a
/// state tag via `Tag.into()` (transition, or "stay" when it equals the
/// current tag). On a transition the body sets the state cell and runs the
/// new state's `enter`.
fn handle(&mut self, ev: Self::Ev, cx: &mut Cx<Self::Ev>);
/// React to one event, returning a [`Step`] the loop acts on. The body
/// first routes a `postpone` row (handing the event back untouched as
/// [`Step::Postponed`]); otherwise it matches `(state, event)`, performs
/// side effects / replies, and ends each arm in a [`Resolution`] — typically
/// a state tag via `Tag.into()` (transition, or "stay" when it equals the
/// current tag). On a transition the body sets the state cell, runs the new
/// state's `enter`, and returns [`Step::Transitioned`]; a stay or unmatched
/// event returns [`Step::Stayed`].
fn handle(&mut self, ev: Self::Ev, cx: &mut Cx<Self::Ev>) -> Step<Self::Ev>;
}
// ---------------------------------------------------------------------------
// Resolution
// ---------------------------------------------------------------------------
/// The outcome of handling one event, before the loop/handler acts on it:
/// dispatch reduces to `(state, event) -> Resolution<State>`.
/// The outcome of the **consuming** dispatch — the by-value `match (state,
/// event)` — before the apply-tail acts on it: `(state, event) ->
/// Resolution<State>`. Postpone is *not* here: a deferred event never reaches
/// this match (it is routed out first; see [`Step`] and the macro's two-phase
/// `handle`), so the only outcomes are a target state or "unhandled".
///
/// [`From<S>`](From) is why a bare state tag works as an arm tail and why
/// "stay" needs no keyword — `Tag.into()` is `To(Tag)`, and the handler treats
@@ -89,10 +138,6 @@ pub enum Resolution<S> {
/// End in state `s`. A **transition** when `s != current` (set the cell,
/// run `enter`); a **stay** when `s == current` (no `enter`).
To(S),
/// Defer the current event onto the postpone queue, to be replayed after
/// the next real transition. Produced by `cx.postpone()`; present
/// here for forward-compatibility but not yet generated.
Postpone,
/// No arm matched `(state, event)`: log-and-drop via
/// [`Cx::on_unhandled`], mirroring `gen_server`'s handling of unexpected
/// messages.
@@ -105,31 +150,187 @@ impl<S> From<S> for Resolution<S> {
}
}
/// What one [`handle`](Machine::handle) call did, as the loop needs to see it.
/// Unlike [`Resolution`] (internal to the consuming match), this is the
/// loop-visible result, because the loop must know two things the match alone
/// doesn't surface: whether an event was **deferred** (so the loop owns it for
/// the postpone queue) and whether a **real transition** happened (so the loop
/// replays that queue). The three cases are mutually exclusive.
pub enum Step<Ev> {
/// The event was deferred by a `postpone` row and handed back untouched.
/// The loop pushes it onto the postpone queue; no state change occurred.
Postponed(Ev),
/// A real transition happened (`enter` for the new state has already run).
/// The loop replays the postpone queue in the new state.
Transitioned,
/// Handled with no transition — a stay or an unmatched event. Nothing is
/// deferred and the postpone queue is left as-is.
Stayed,
}
// ---------------------------------------------------------------------------
// Cx — the per-handler context
// ---------------------------------------------------------------------------
// ---------------------------------------------------------------------------
// Sys — the loop's internal timer-fire channel
// ---------------------------------------------------------------------------
/// A timer fire landing on the loop's own system channel, selected above the
/// inbox so a timeout cannot be starved by inbox traffic. Each fire carries the
/// local id it was armed under so the loop can confirm it is still the live one
/// (a fire that a reset/cancel beat onto the channel is discarded).
enum Sys {
/// The state-timeout fired, carrying the local id it was armed under so the
/// loop can drop a stale fire (one a reset/cancel beat onto the channel).
StateTimeout(u64),
/// A named timeout fired, carrying its name and the local id it was armed
/// under.
Timeout(&'static str, u64),
}
// ---------------------------------------------------------------------------
// Timers — shared timer bookkeeping
// ---------------------------------------------------------------------------
/// Per-machine timer bookkeeping, shared between the loop and every [`Cx`]
/// borrow (a machine is single-threaded — handler calls and the loop never run
/// concurrently — so this `Mutex` is always uncontended; it is `Arc<Mutex>`
/// rather than `Rc<RefCell>` only because `Machine: Send` forces it).
///
/// Each arming mints a fresh **local id** carried in the fire payload and
/// recorded here alongside the substrate id used to cancel. On fire the loop
/// compares the payload's local id against the recorded one: a fire whose id no
/// longer matches — a reset/cancel armed a newer one or removed the entry after
/// this fire already escaped onto the channel — is stale and dropped. The local
/// id is minted up front (unlike the substrate id, which only exists once armed),
/// so it can ride the payload with no chicken-and-egg.
struct Timers {
/// Monotonic minter for local ids, kept distinct from substrate ids (the
/// latter are what we hand to [`cancel_timer`]).
next_local: u64,
/// The currently-armed state-timeout's `(local id, substrate id)`. Cancelled
/// and cleared on every real state change (auto-reset), replaced on re-arm.
state: Option<(u64, TimerId)>,
/// Live named timeouts: name → `(local id, substrate id)`. Survives state
/// changes; an entry lives until it fires or is cancelled.
named: HashMap<&'static str, (u64, TimerId)>,
}
impl Timers {
fn new() -> Self {
Timers { next_local: 0, state: None, named: HashMap::new() }
}
fn mint(&mut self) -> u64 {
let id = self.next_local;
self.next_local = self.next_local.wrapping_add(1);
id
}
}
// ---------------------------------------------------------------------------
// Cx — the per-handler context
// ---------------------------------------------------------------------------
/// The context handle injected into [`Machine::on_start`] and
/// [`Machine::handle`]. Non-state outcomes (postpone, timeout arming) live here
/// as method calls rather than keywords.
/// [`Machine::handle`]. Non-state outcomes timeout arming, and later
/// postpone — live here as method calls rather than keywords. It lives only on
/// the actor's own stack and is never sent.
///
/// For now it carries only the [`on_unhandled`](Self::on_unhandled) hook;
/// `cx.state_timeout(d)` / `cx.timeout(name, d)` and `cx.postpone()` will attach
/// here when implemented. It lives only on the actor's own
/// stack and is never sent.
/// Timeouts armed here land on the loop's system channel through `sys_tx` and
/// are tracked in the shared `reg` so reset/cancel can find the pending one.
pub struct Cx<Ev> {
sys_tx: Sender<Sys>,
reg: Arc<Mutex<Timers>>,
_ev: PhantomData<fn() -> Ev>,
}
impl<Ev> Cx<Ev> {
fn new() -> Self {
Cx { _ev: PhantomData }
fn new(sys_tx: Sender<Sys>, reg: Arc<Mutex<Timers>>) -> Self {
Cx { sys_tx, reg, _ev: PhantomData }
}
/// The default for an event no arm matched: **log-and-drop**. Overridable
/// hook wiring is a follow-on; for now an unmatched event is
/// silently dropped, as `gen_server` does with unexpected messages.
/// Arm the **state timeout**: fire a `state_timeout` event after `after` in
/// the current state. Auto-reset on any state change (the loop cancels and
/// clears it on every real transition), so it measures quiet time *within* a
/// state. Only one is ever pending — arming again cancels and replaces the
/// previous one.
pub fn state_timeout(&mut self, after: Duration) {
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
if let Some((_, old_sub)) = reg.state.take() {
cancel_timer(old_sub);
}
let local = reg.mint();
let sub = send_after_to(after, self.sys_tx.clone(), Sys::StateTimeout(local));
reg.state = Some((local, sub));
}
/// Arm a **named generic timeout**: fire a `timeout(name)` event after
/// `after`. Survives state changes; several may be in flight keyed by
/// `name`. Arming the same `name` again cancels and replaces the pending one
/// for that name.
pub fn timeout(&mut self, name: &'static str, after: Duration) {
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
if let Some((_, old_sub)) = reg.named.remove(name) {
cancel_timer(old_sub);
}
let local = reg.mint();
let sub = send_after_to(after, self.sys_tx.clone(), Sys::Timeout(name, local));
reg.named.insert(name, (local, sub));
}
/// Cancel the named timeout `name` if pending. Returns `true` if the cancel
/// beat the fire, `false` if it had already fired / was never armed.
pub fn cancel_timeout(&mut self, name: &'static str) -> bool {
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
match reg.named.remove(name) {
Some((_, sub)) => cancel_timer(sub),
None => false,
}
}
/// Cancel the pending state-timeout if any. Returns `true` if the cancel
/// beat the fire. Rarely needed by hand (the loop auto-resets on
/// transition); exposed for a handler that wants to disarm within a state.
pub fn cancel_state_timeout(&mut self) -> bool {
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
match reg.state.take() {
Some((_, sub)) => cancel_timer(sub),
None => false,
}
}
/// The default for an event no arm matched: **log-and-drop**, as
/// `gen_server` does with unexpected messages.
pub fn on_unhandled(&mut self) {}
/// Cancel and clear the pending state-timeout. Called by the macro on every
/// real transition (the state-timeout is auto-reset across state changes);
/// not part of the authoring surface. A handler re-arms in the new state's
/// `enter` if it wants one.
#[doc(hidden)]
pub fn __reset_state_timeout(&mut self) {
let mut reg = match self.reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
if let Some((_, sub)) = reg.state.take() {
cancel_timer(sub);
}
}
}
// ---------------------------------------------------------------------------
@@ -143,9 +344,8 @@ impl<Ev> Cx<Ev> {
/// machine dies, closing the channel).
///
/// Carrying the handle in the event — rather than the loop owning a reply slot
/// — is what lets a later chunk **postpone a call**: the whole event, reply
/// handle included, moves onto the postpone queue and is answered by a later
/// state.
/// — is what will let a **postponed call** work: the whole event, reply handle
/// included, moves onto the postpone queue and is answered by a later state.
pub struct Reply<T> {
tx: Sender<T>,
}
@@ -211,9 +411,7 @@ impl<M: Machine> GenStatemRef<M> {
/// stack unwinds, closing the channel and waking the caller).
///
/// `make` wraps the [`Reply`] all the way to `M::Ev` — typically
/// `|r| Ev::Call(MyCall::GetCount(r))`. (The deferred macro would generate a
/// thinner `call` that hides the `Ev::Call` wrap and takes the bare variant
/// constructor.)
/// `|r| Ev::Call(MyCall::GetCount(r))`.
pub fn call<T, F>(&self, make: F) -> Result<T, CallError>
where
T: Send + 'static,
@@ -241,18 +439,84 @@ pub fn spawn<M: Machine>(machine: M) -> GenStatemRef<M> {
GenStatemRef { tx, pid: handle.pid() }
}
/// The machine actor body: `on_start`, then one `handle` per inbox event until
/// the inbox closes (all refs dropped → graceful shutdown). Chunk 1 parks on
/// the inbox alone — no system arm, no timers — the analogue of `gen_server`'s
/// plain-inbox park.
/// The machine actor body: `on_start`, then one `handle` per event until the
/// inbox closes (all refs dropped → graceful shutdown).
///
/// Two intake sources are selected each iteration with the **timer arm above
/// the inbox**, so a timeout fire is never starved by inbox traffic: `sys_rx`
/// carries timer fires armed through `cx`, `rx` is the user inbox. A fire is
/// turned into the matching internal event (`state_timeout` / `timeout(name)`)
/// and run through the same `handle` dispatch as an inbox event — the
/// gen_statem model, where timeouts surface as ordinary events.
///
/// The loop owns the **postpone queue**: a `handle` that defers its event hands
/// it back ([`Step::Postponed`]) for the queue; a `handle` that transitions
/// ([`Step::Transitioned`]) triggers a [`replay`] of the queue in the new
/// state, ahead of the next intake.
fn statem_loop<M: Machine>(rx: Receiver<M::Ev>, mut machine: M) {
let mut cx = Cx::new();
let (sys_tx, sys_rx) = channel::<Sys>();
let reg = Arc::new(Mutex::new(Timers::new()));
// The loop owns `cx` (and through it a `sys_tx` clone) for its whole life,
// so the sys arm never closes from under us — no auto-close dance needed.
let mut cx = Cx::new(sys_tx, reg.clone());
// Events deferred by `postpone` rows, replayed FIFO on the next transition.
let mut postpone: VecDeque<M::Ev> = VecDeque::new();
machine.on_start(&mut cx);
loop {
match rx.recv() {
Ok(ev) => machine.handle(ev, &mut cx),
// All StatemRefs dropped → inbox closed → shutdown.
Err(_) => break,
// Timer arm first: a ready fire is taken in preference to the inbox.
let i = select(&[&sys_rx, &rx]);
if i == 0 {
match sys_rx.try_recv() {
Ok(Some(fire)) => {
// Confirm the fire is still the live one before dispatching:
// a reset/cancel may have replaced/removed it after it
// escaped onto the channel. A stale fire is dropped.
let ev = match fire {
Sys::StateTimeout(local) => {
let mut t = match reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
match t.state {
Some((live, _)) if live == local => {
t.state = None; // retire: it has now fired
Some(M::state_timeout_ev())
}
_ => None,
}
}
Sys::Timeout(name, local) => {
let mut t = match reg.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: gen_statem reg lock poisoned (core corrupt): {e}"),
};
match t.named.get(name) {
Some(&(live, _)) if live == local => {
t.named.remove(name); // retire on fire
Some(M::timeout_ev(name))
}
_ => None,
}
}
};
if let Some(ev) = ev {
dispatch(&mut machine, &mut cx, &mut postpone, ev);
}
}
// Single-receiver: nothing can drain the arm between select's
// ready and our try_recv.
Ok(None) => debug_assert!(false, "ready system arm was empty"),
// The loop holds a sys_tx for its whole life, so this is
// unreachable; fall through defensively.
Err(_) => {}
}
} else {
match rx.try_recv() {
Ok(Some(ev)) => dispatch(&mut machine, &mut cx, &mut postpone, ev),
Ok(None) => debug_assert!(false, "ready inbox was empty"),
// All GenStatemRefs dropped → inbox closed → shutdown.
Err(_) => break,
}
}
// Observation point so a machine fed a hot inbox stays preemptible and
// cancellable.
@@ -260,15 +524,60 @@ fn statem_loop<M: Machine>(rx: Receiver<M::Ev>, mut machine: M) {
}
}
/// Run one event through `handle` and act on its [`Step`]: stash a deferred
/// event on the postpone queue, or — on a real transition — [`replay`] the
/// queue in the new state. A stay/unmatched event needs nothing further.
fn dispatch<M: Machine>(
machine: &mut M,
cx: &mut Cx<M::Ev>,
postpone: &mut VecDeque<M::Ev>,
ev: M::Ev,
) {
match machine.handle(ev, cx) {
Step::Postponed(ev) => postpone.push_back(ev),
Step::Stayed => {}
Step::Transitioned => replay(machine, cx, postpone),
}
}
/// Replay deferred events after a real transition: each goes back through
/// `handle` in FIFO order, in the now-current state. An event that postpones
/// again re-queues (to wait for the *next* transition); one that transitions
/// re-arms the replay, so a later state can in turn drain what is still pending.
/// Subsequent events in a batch already see the post-transition state, since
/// `handle` reads the live state cell — the outer loop only re-runs to give
/// re-queued events another pass once a transition has occurred within a batch.
fn replay<M: Machine>(machine: &mut M, cx: &mut Cx<M::Ev>, postpone: &mut VecDeque<M::Ev>) {
loop {
if postpone.is_empty() {
return;
}
// Take the current backlog; anything deferred during this pass lands in
// the now-empty queue and is only retried if this pass also transitioned.
let batch = std::mem::take(postpone);
let mut transitioned = false;
for ev in batch {
match machine.handle(ev, cx) {
Step::Postponed(ev) => postpone.push_back(ev),
Step::Stayed => {}
Step::Transitioned => transitioned = true,
}
}
if !transitioned {
return;
}
}
}
// ---------------------------------------------------------------------------
// gen_statem! — the authoring macro (fused total-match variant)
// gen_statem! — the authoring macro (total-match dispatch)
// ---------------------------------------------------------------------------
/// Assemble a complete [`Machine`] from hand-written types, a glanceable
/// transition table, and free handler functions.
///
/// This is the **fused** authoring surface (see `examples/statem_fused.rs` for
/// the same machine written out by hand). You keep ownership of every type that
/// This is the authoring surface (see `examples/gen_statem_expanded.rs` for the
/// same machine written out by hand). You keep ownership of every type that
/// carries meaning — the state enum, the data struct, the `cast`/`call` enums,
/// and any per-row successor enums — and the macro emits only the mechanical
/// scaffolding: the unified event enum, the machine struct and its private
@@ -308,12 +617,12 @@ fn statem_loop<M: Machine>(rx: Receiver<M::Ev>, mut machine: M) {
/// machine lives in the same crate as this macro, and is **silently dropped
/// when you call `gen_statem!` from a downstream crate** — neither a crate
/// `#![deny]` nor `#![forbid]` overrides the suppression. This is the one
/// guarantee a `macro_rules!` cannot carry across the crate boundary; the
/// RFC's proc-macro could (it would stamp the arms with call-site spans).
/// guarantee a `macro_rules!` cannot carry across the crate boundary; a
/// proc-macro could (it would stamp the arms with call-site spans).
/// Guard against it in review, or with an in-crate test of the machine.
///
/// There is deliberately **no separate `transitions { … }` adjacency block**:
/// in this variant the `match` *is* the table and the successor enums *are* the
/// the `match` *is* the table and the successor enums *are* the
/// per-state target sets, so a second listing would be redundant and
/// un-cross-checkable without a proc-macro. The graph is read off the `on`
/// blocks and the successor enums at the top of the file.
@@ -323,14 +632,16 @@ fn statem_loop<M: Machine>(rx: Receiver<M::Ev>, mut machine: M) {
/// ```ignore
/// // ── your types (the macro never generates or inspects these) ───────────
/// #[derive(Clone, Copy, PartialEq, Eq, Debug)]
/// enum Switch { Off, On }
/// struct Counts { flips: u32, enters: u32 }
/// enum SwitchCast { Flip }
/// enum SwitchCall { GetCount(Reply<u32>) }
/// enum Door { Open, Closed, Locked }
/// struct Data { enters: u32 }
/// enum DoorCast { Push, Pull, Lock, Unlock(u32) }
/// enum DoorCall { GetState(Reply<Door>) }
///
/// gen_statem! {
/// machine: SwitchSm { state: Switch, data: Counts };
/// event: Ev { cast: SwitchCast, call: SwitchCall };
/// machine: DoorSm { state: Door, data: Data };
/// // The event clause folds three user enums together. `info` is for
/// // out-of-band messages; use `()` if you have none.
/// event: Ev { cast: DoorCast, call: DoorCall, info: () };
///
/// // Name the bindings your handler bodies use. A declarative macro can't
/// // hand you its own `self`/`cx` (hygiene), so you choose the identifiers
@@ -339,26 +650,51 @@ fn statem_loop<M: Machine>(rx: Receiver<M::Ev>, mut machine: M) {
/// context(data, prev, cx);
///
/// // `enter` runs on entry to a state; side effects only, returns ().
/// // Match the state tag (or `_`); the arm body is statements.
/// // Match the state tag (or `_`); the arm body is statements. Arm any
/// // state-timeout here (it is auto-reset on the way into a new state).
/// enter {
/// Door::Open => {
/// data.enters += 1;
/// cx.state_timeout(std::time::Duration::from_secs(30)); // auto-close
/// }
/// _ => data.enters += 1,
/// }
///
/// // The transition table. Group rows by current state with `on <pat>`.
/// // A row is: cast|call <event-pattern> [if <guard>] => <tail> ,
/// // A row is: <kind> <event-pattern> [if <guard>] => <tail> ,
/// // where <kind> is one of `cast`, `call`, `info`, `state_timeout`
/// // (no pattern — it is a unit event), or `timeout <name-pattern>`,
/// // and the tail is one of:
/// // * a state tag `Switch::On` (transition, or "stay"
/// // if it equals current)
/// // * a block ending in one `{ data.flips += 1; Switch::On }`
/// // * a successor-enum value `unlock(key)` (branching row)
/// // * the keyword `unhandled` (explicit refusal)
/// on Switch::Off => {
/// cast SwitchCast::Flip => { data.flips += 1; Switch::On },
/// call SwitchCall::GetCount(r) => { r.reply(data.flips); prev },
/// // * a state tag `Door::Closed` (transition, or "stay"
/// // if it equals current)
/// // * a block ending in one `{ data.enters += 1; Door::Closed }`
/// // * a successor-enum value `on_unlock(key)` (branching row)
/// // * the keyword `unhandled` (explicit refusal)
/// // * the keyword `postpone` (defer until next
/// // transition; cast/call/
/// // info only)
/// on Door::Open => {
/// cast DoorCast::Push => Door::Closed,
/// // An armed state-timeout surfaces as an ordinary event:
/// state_timeout => Door::Closed,
/// cast DoorCast::Pull | DoorCast::Lock | DoorCast::Unlock(_) => unhandled,
/// }
/// on Switch::On => {
/// cast SwitchCast::Flip => Switch::Off,
/// call SwitchCall::GetCount(r) => { r.reply(data.flips); prev },
/// on Door::Closed => {
/// cast DoorCast::Pull => Door::Open,
/// cast DoorCast::Lock => Door::Locked,
/// cast DoorCast::Push | DoorCast::Unlock(_) => unhandled,
/// // No state-timeout armed here, so refuse it.
/// state_timeout => unhandled,
/// }
/// on Door::Locked => {
/// cast DoorCast::Unlock(key) => on_unlock(key), // -> successor enum
/// cast DoorCast::Push | DoorCast::Pull | DoorCast::Lock => unhandled,
/// state_timeout => unhandled,
/// }
///
/// // state-independent queries (reply, then stay via `prev`)
/// on _ => {
/// call DoorCall::GetState(r) => { r.reply(prev); prev },
/// }
/// }
/// ```
@@ -368,30 +704,53 @@ fn statem_loop<M: Machine>(rx: Receiver<M::Ev>, mut machine: M) {
/// * **Every row ends in a comma** (block-bodied ones too) and every `on` block
/// is `on <state-pat> => { … }`. These two are macro-grammar requirements, not
/// style: a declarative matcher can't otherwise tell where a row's tail ends.
/// * **Qualify event patterns** (`SwitchCast::Flip`) and **state tags**
/// (`Switch::On`). A declarative macro can't prepend the enum name inside an
/// opaque pattern fragment, so the `cast`/`call` keyword only selects the
/// event wrapper — it does not qualify for you. The keyword also reads as a
/// sync/async marker at a glance.
/// * **Qualify event patterns** (`DoorCast::Push`) and **state tags**
/// (`Door::Closed`). A declarative macro can't prepend the enum name inside an
/// opaque pattern fragment, so the row keyword only selects the event wrapper
/// — it does not qualify for you. `cast`/`call` also read as a sync/async
/// marker at a glance.
/// * **Timeouts surface as events.** `state_timeout` (a unit event) and
/// `timeout <name>` (matching the `&'static str` a generic timeout was armed
/// under) are matched in `on` rows just like casts and calls — arm them via
/// `cx`, handle the fire here. Because a `timeout` name is an `&str`, a row
/// that matches specific names needs a `timeout _ => …` fallback to stay
/// exhaustive.
/// * **`info` defaults to a silent drop.** An out-of-band `info` you do not
/// match anywhere is dropped (the `gen_server` default), so a machine that
/// ignores info writes no `info` rows at all. State-timeouts and named
/// timeouts have **no** such default: a state that can see one must handle it
/// (or `unhandled` it) or the match is non-exhaustive.
/// * **Stay** = return the current tag. The `prev` you named in `context` is
/// bound to the pre-handler state for exactly this — handy in any-state
/// (`on _`) rows where there is no single literal tag to write.
/// * **`data` and `cx`** (the names you chose) are in scope in every body:
/// mutate `data`, reply through a `Reply` bound in the pattern, and (chunks
/// 23) arm timeouts or postpone via `cx`. You never write `self`.
/// mutate `data`, reply through a `Reply` bound in the pattern, and arm
/// timeouts via `cx`. You never write `self`.
/// * **Guards** are plain match guards. A guarded arm does not count toward
/// exhaustiveness, so pair it with an unguarded fallback or you'll (correctly)
/// trip `E0004`.
/// * **Branching rows** return a successor enum that `impl`s `From<_>` for the
/// state type; the macro supplies the outer `.into()`.
/// * **`postpone`** defers the current event untouched, to be replayed after the
/// next *real transition* (a stay does not trigger replay). It is available
/// for `cast`, `call`, and `info` rows — not the timeout events. The deferred
/// event keeps any `Reply` it carries, so a postponed `call` is answered by
/// whichever later state handles the replay. The body runs *no* code (the
/// event is untouched), so a `postpone` row is just `cast Foo(_) => postpone,`;
/// prefer a non-binding pattern, and if you guard it, the guard must not depend
/// on the event's payload (it is evaluated in a borrow pre-pass). On a
/// transition the queue drains FIFO, ahead of further inbox/timer events; a
/// replayed event may postpone again (it re-queues for the next transition).
///
/// # What it emits
///
/// `enum $Ev { Cast($Cast), Call($Call) }`, `struct $Sm { state, data }`,
/// `$Sm::start(init, data) -> GenStatemRef<$Sm>`, the `Machine` impl (`on_start`
/// running the initial `enter`; `handle` = the dispatch match + the
/// stay/transition/unhandled apply-tail, the cell's sole writer), and the
/// `enter` dispatch. Chunk 1: real time, no timers, no postpone.
/// The unified `enum $Ev` (the `Cast`/`Call`/`Info` wrappers plus the internal
/// `StateTimeout` / `Timeout` events), `struct $Sm { state, data }`,
/// `$Sm::start(init, data) -> GenStatemRef<$Sm>`, and the `Machine` impl:
/// `on_start` runs the initial `enter`; `handle` is the dispatch match plus the
/// stay/transition/unhandled apply-tail (the cell's sole writer, which also
/// auto-resets the state-timeout on every real transition); and the `enter`
/// dispatch.
///
/// # Limitation
///
@@ -403,16 +762,23 @@ macro_rules! gen_statem {
// ===== public entry =====================================================
(
machine: $sm:ident { state: $State:ty, data: $Data:ty } ;
event: $Ev:ident { cast: $Cast:ty, call: $Call:ty } ;
event: $Ev:ident { cast: $Cast:ty, call: $Call:ty, info: $Info:ty } ;
context ( $data:ident , $cur:ident , $cx:ident ) ;
enter { $( $est:pat => $ebody:expr ),+ $(,)? }
$( on $st:pat => { $($rows:tt)* } )+
) => {
/// Unified inbox payload: the user's `cast`/`call` enums folded together
/// (chunks 23 add the runtime's internal timeout variants here).
/// Unified inbox payload: the user's `cast`/`call`/`info` enums folded
/// together with the runtime's internal timeout events.
enum $Ev {
Cast($Cast),
Call($Call),
/// Out-of-band message, matched in `info` rows. An unmatched info is
/// silently dropped (the `gen_server` default).
Info($Info),
/// The state-timeout fired (matched in `state_timeout` rows).
StateTimeout,
/// A named timeout fired (matched in `timeout <pat>` rows).
Timeout(&'static str),
}
struct $sm {
@@ -438,6 +804,14 @@ macro_rules! gen_statem {
impl $crate::gen_statem::Machine for $sm {
type Ev = $Ev;
fn state_timeout_ev() -> $Ev {
$Ev::StateTimeout
}
fn timeout_ev(name: &'static str) -> $Ev {
$Ev::Timeout(name)
}
fn on_start(&mut self, $cx: &mut $crate::gen_statem::Cx<$Ev>) {
let s = self.state;
self.enter(s, $cx);
@@ -446,79 +820,205 @@ macro_rules! gen_statem {
#[allow(unused_variables)]
#[deny(unreachable_patterns)] // conflicting rows must fail even though
// this match is external-macro-expanded
fn handle(&mut self, ev: $Ev, $cx: &mut $crate::gen_statem::Cx<$Ev>) {
fn handle(&mut self, ev: $Ev, $cx: &mut $crate::gen_statem::Cx<$Ev>)
-> $crate::gen_statem::Step<$Ev>
{
// Caller-named bindings (shared call-site hygiene, so row bodies
// can see them): `$cur` = current state tag, `$data` = &mut Data.
let $cur = self.state;
let $data = &mut self.data;
// @arms emits a block: a postpone pre-pass that `return`s
// `Step::Postponed(ev)` for a deferred event (handing it back
// untouched), then the consuming `match (state, event)` whose
// value is this `Resolution`.
let next: $crate::gen_statem::Resolution<$State> =
$crate::gen_statem!(@arms ($Ev) ($cur, ev) [ ]
$crate::gen_statem!(@arms ($Ev) ($cur, ev) [ ] [ ]
$( on $st => { $($rows)* } )+);
match next {
$crate::gen_statem::Resolution::To(s) if s == $cur => {}
$crate::gen_statem::Resolution::To(s) if s == $cur => {
$crate::gen_statem::Step::Stayed
}
$crate::gen_statem::Resolution::To(s) => {
self.state = s; // <- sole writer of the state cell
// A real transition auto-resets the state-timeout: the
// new state re-arms in its `enter` if it wants one.
$cx.__reset_state_timeout();
self.enter(s, $cx);
$crate::gen_statem::Step::Transitioned
}
$crate::gen_statem::Resolution::Postpone => {
unreachable!("postpone is not generated yet")
$crate::gen_statem::Resolution::Unhandled => {
$cx.on_unhandled();
$crate::gen_statem::Step::Stayed
}
$crate::gen_statem::Resolution::Unhandled => $cx.on_unhandled(),
}
}
}
};
// ===== @arms: build the dispatch match ==================================
// No more on-blocks: emit the (total, catch-all-free) match.
(@arms ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ]) => {
match ($ss, $se) { $($arms)* }
// ===== @arms / @rows: build the two-phase dispatch ======================
// Two accumulators are threaded: `[ $($arms)* ]` is the phase-2 consuming
// match (by value); `[ $($post)* ]` is the phase-1 postpone router (by ref).
// A normal row feeds only phase-2; a `postpone` row feeds both — a `=> true`
// router and a `=> unreachable!` phase-2 filler that keeps the consuming
// match total.
//
// Terminal (no on-blocks left): emit the block the `handle` body assigns to
// `next`. Phase 1 routes a deferred event back out as `Step::Postponed`;
// phase 2 is the catch-all-free consuming match (the one macro-injected arm
// is the `Info` silent-drop — last and broadest, so per-state `info` rows
// stay reachable; cast/call/timeouts get no fallback, so a forgotten pair is
// still E0004).
(@arms ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ]) => {
{
// Phase 1 — postpone routing (borrow-only). A guard on a postpone
// row runs here, against by-ref bindings, so it must not depend on
// the event's payload.
let __defer = match ($ss, &$se) {
$($post)*
_ => false,
};
if __defer {
return $crate::gen_statem::Step::Postponed($se);
}
// Phase 2 — the consuming dispatch. Each postpone pair reappears here
// as `unreachable!` so the match stays total; phase 1 already
// returned for it.
match ($ss, $se) {
$($arms)*
(_, $Ev::Info(_)) => $crate::gen_statem::Resolution::Unhandled,
}
}
};
// Open an on-block: remember its state pat, drain its rows, then continue.
(@arms ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ]
(@arms ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ]
on $st:pat => { $($rows:tt)* } $($more:tt)*
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se) [ $($arms)* ] ($st)
$crate::gen_statem!(@rows ($Ev) ($ss, $se) [ $($arms)* ] [ $($post)* ] ($st)
{ $($rows)* } { $($more)* })
};
// ===== @rows: drain one on-block's rows, threading the global acc ========
// ===== @rows: drain one on-block's rows, threading both accs =============
// cast, explicit refusal
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ cast $ev:pat $(if $g:expr)? => unhandled , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Cast($ev)) $(if $g)? => $crate::gen_statem::Resolution::Unhandled, ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// cast, postpone (defer the event; the replay in a later state handles it)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ cast $ev:pat $(if $g:expr)? => postpone , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Cast($ev)) $(if $g)? => unreachable!("postponed event is replayed, not dispatched here"), ]
[ $($post)* ($st, $Ev::Cast($ev)) $(if $g)? => true, ]
($st) { $($rows)* } { $($more)* })
};
// cast, transition / stay / branch
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ cast $ev:pat $(if $g:expr)? => $tail:expr , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Cast($ev)) $(if $g)? => $crate::gen_statem::Resolution::To($tail.into()), ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// call, explicit refusal
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ call $ev:pat $(if $g:expr)? => unhandled , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Call($ev)) $(if $g)? => $crate::gen_statem::Resolution::Unhandled, ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// call, postpone (the Reply rides inside the event onto the queue)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ call $ev:pat $(if $g:expr)? => postpone , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Call($ev)) $(if $g)? => unreachable!("postponed event is replayed, not dispatched here"), ]
[ $($post)* ($st, $Ev::Call($ev)) $(if $g)? => true, ]
($st) { $($rows)* } { $($more)* })
};
// call, transition / stay / branch
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ call $ev:pat $(if $g:expr)? => $tail:expr , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Call($ev)) $(if $g)? => $crate::gen_statem::Resolution::To($tail.into()), ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// info, explicit refusal
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ info $ev:pat $(if $g:expr)? => unhandled , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Info($ev)) $(if $g)? => $crate::gen_statem::Resolution::Unhandled, ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// info, postpone
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ info $ev:pat $(if $g:expr)? => postpone , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Info($ev)) $(if $g)? => unreachable!("postponed event is replayed, not dispatched here"), ]
[ $($post)* ($st, $Ev::Info($ev)) $(if $g)? => true, ]
($st) { $($rows)* } { $($more)* })
};
// info, transition / stay / branch
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ info $ev:pat $(if $g:expr)? => $tail:expr , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Info($ev)) $(if $g)? => $crate::gen_statem::Resolution::To($tail.into()), ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// state_timeout, explicit refusal (unit event — no pattern; not postponable)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ state_timeout $(if $g:expr)? => unhandled , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::StateTimeout) $(if $g)? => $crate::gen_statem::Resolution::Unhandled, ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// state_timeout, transition / stay / branch
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ state_timeout $(if $g:expr)? => $tail:expr , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::StateTimeout) $(if $g)? => $crate::gen_statem::Resolution::To($tail.into()), ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// timeout, explicit refusal (pattern matches the name; not postponable)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ timeout $ev:pat $(if $g:expr)? => unhandled , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Timeout($ev)) $(if $g)? => $crate::gen_statem::Resolution::Unhandled, ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// timeout, transition / stay / branch
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ timeout $ev:pat $(if $g:expr)? => $tail:expr , $($rows:tt)* } { $($more:tt)* }
) => {
$crate::gen_statem!(@rows ($Ev) ($ss, $se)
[ $($arms)* ($st, $Ev::Timeout($ev)) $(if $g)? => $crate::gen_statem::Resolution::To($tail.into()), ]
[ $($post)* ]
($st) { $($rows)* } { $($more)* })
};
// this block is drained: hand the remaining on-blocks back to @arms
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat)
(@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] [ $($post:tt)* ] ($st:pat)
{ } { $($more:tt)* }
) => {
$crate::gen_statem!(@arms ($Ev) ($ss, $se) [ $($arms)* ] $($more)*)
$crate::gen_statem!(@arms ($Ev) ($ss, $se) [ $($arms)* ] [ $($post)* ] $($more)*)
};
}
+15 -9
View File
@@ -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<Completion> {
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;
+4 -1
View File
@@ -135,7 +135,10 @@ pub fn link<A>(target: Pid<A>) {
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);
+89 -26
View File
@@ -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<T> Mutex<T> {
}
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<MutexGuard<'_, T>, 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<T> Mutex<T> {
// 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<T> Mutex<T> {
// 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<T> Mutex<T> {
// 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<T> Mutex<T> {
pub fn try_lock(&self) -> Option<MutexGuard<'_, T>> {
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<T> Mutex<T> {
// 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<T> 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<T> 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<T: std::fmt::Debug> 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<T> 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);
+128 -90
View File
@@ -1,15 +1,17 @@
//! Process groups — a `name → multiset<Member>` map (RFC 012).
//! Process groups: one name, many actors.
//!
//! Sits parallel to [`registry`](crate::registry), not on top of it. The
//! registry is a *bimap*: at most one pid per name. A group is the opposite —
//! many pids per name, the same pid in many groups — so it cannot be a
//! generalization of the bimap; it is its own keyspace sharing only the
//! liveness signal source (monitors) and the lock *class*.
//! A process group is a named set of actors that you can look up, fan out to,
//! or pick a worker from. It is the natural home for a *worker pool* (several
//! interchangeable actors doing the same job), for *service discovery* (find
//! everyone currently offering some capability), and for *broadcast* (reach
//! every member of a group at once).
//!
//! ## Example
//! The set is *live*: members [`join`] it, and a member that dies is removed
//! automatically. You never deregister a dead actor — there is no bookkeeping
//! to get wrong, and [`members`] / [`pick`] never hand you an actor that has
//! already gone.
//!
//! A group is a live membership view: members join, and a member that dies is
//! evicted automatically — no deregistration call, no bookkeeping.
//! ## Joining and reading a group
//!
//! ```
//! use smarm::{channel, join, leave, members, pick, run, spawn};
@@ -25,8 +27,8 @@
//! join("pool", w2.pid());
//! assert_eq!(members("pool").len(), 2);
//!
//! // One worker dies. Nobody told the group — the death hook evicts it,
//! // so it is gone from `members` and never returned by `pick`.
//! // One worker dies. Nothing tells the group — smarm evicts it
//! // automatically, so it is gone from `members` and never picked.
//! tx1.send(()).unwrap();
//! w1.join().unwrap();
//! assert_eq!(members("pool"), vec![w2.pid()]);
@@ -41,60 +43,72 @@
//! });
//! ```
//!
//! ## Cleanup is eager and monitor-driven (unlike the registry)
//! [`members`] returns every live member in the order they joined; [`pick`]
//! returns one of them, or `None` when the group is empty. The same actor can
//! belong to any number of groups at once, and joining a group it is already in
//! is a harmless no-op.
//!
//! The registry prunes a stale binding lazily, on contact, because it only
//! ever resolves one binding at a time. A group is *iterated* — `members` fans
//! out to every member — so it must not carry dead members across a broadcast.
//! Each [`join`] installs a `monitor(pid)` and keeps the resulting [`Monitor`]
//! *alongside the group entry*; the actor's one-shot `Down` lands in that
//! monitor's channel when it dies. Every group operation [`reap`s][reap] the
//! group it touches first — draining each membership's monitor with a
//! non-blocking `try_recv` — and on the first sign of death sweeps that pid out
//! of *every* group via the predicate primitive. So the set is self-pruning on
//! contact rather than filtered on read; the read path additionally applies a
//! generation-checked liveness backstop so a member already dead but not yet
//! reaped is never *returned*, even though eviction remains the monitor's job.
//! ## Membership ends on its own
//!
//! [reap]: ProcessGroups::reap_group
//! You do not have to clean up after a member that dies. When an actor exits —
//! for any reason — it is removed from every group it had joined, before any
//! later read or send can observe it. [`leave`] is only for *voluntary*
//! departure, when a still-living actor wants out of a group.
//!
//! ## Eviction is one dumb primitive
//! This is the main difference from keeping your own `Vec<Pid>`: a plain list
//! goes stale the instant a member dies, and you would have to notice and prune
//! it yourself. A group prunes itself.
//!
//! [`ProcessGroups::remove_where`] removes every member matching a predicate
//! from every group. The primitive never knows *why* a member leaves — that is
//! the caller's concern. Its first caller is the monitor death hook (this RFC,
//! via `reap_group`); the later `evict_incarnation(node, inc)` sweep (RFC 010)
//! reuses the same predicate path, which is the whole reason to shape it as a
//! predicate.
//! ## Sending to a group
//!
//! ## Identity is cluster-shaped from the first commit
//! For a worker pool you usually want to hand a job to *one* available member.
//! [`dispatch`] picks a live member and sends it a message in a single step,
//! returning the member it reached:
//!
//! A [`Member`] is not a bare [`Pid`]: it is `(NodeId, Incarnation, Pid)`, a
//! field-for-field image of a modern BEAM pid (`NEW_PID_EXT`) so the eventual
//! ETF codec (RFC 011) is a near-identity mapping. In this RFC `NodeId` and
//! `Incarnation` are runtime-init constants — one node, one fixed incarnation
//! — threaded through storage and eviction anyway so the public surface never
//! has to change to acquire them once clustering (RFC 010) supplies real
//! values. No cluster types leak out of the public functions: callers pass and
//! receive [`Pid`]; the node/incarnation are filled from runtime identity.
//! ```ignore
//! use smarm::{dispatch, join, Addressable};
//!
//! ## Locking
//! struct Job(String);
//! struct Worker;
//! impl Addressable for Worker { type Msg = Job; }
//!
//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry`. The
//! group lock is never held with another Leaf (it never touches the registry
//! or a slot's cold lock). The two places that *do* need another lock are kept
//! off the group-lock path:
//! // Each worker has published a `Pid<Worker>` inbox and joined the pool.
//! join("workers", worker_a);
//! join("workers", worker_b);
//!
//! - `monitor()` / `demonitor()` acquire the target's cold lock (Leaf) and
//! so run *before* / *after* the group lock, never under it.
//! - draining a monitor with `try_recv` takes the channel's Channel-class
//! lock — permitted *under* a Leaf by the lock order (`raw_mutex.rs`), and
//! a channel critical section only does the lock-free unpark protocol, so
//! no Leaf is ever nested under it.
//! // Route one job to whichever live worker `pick` lands on.
//! match dispatch::<Worker>("workers", Job("resize image".into())) {
//! Ok(who) => println!("sent to {who:?}"),
//! Err(returned) => println!("no worker took it: {returned:?}"),
//! }
//! ```
//!
//! Evicted and rejected [`Monitor`]s are dropped only *after* the group lock is
//! released, so a receiver-drop never runs a wakeup under the lock (same
//! discipline as `demonitor`).
//! When you want the pids themselves rather than to send right away, [`pick_as`]
//! and [`members_as`] return typed [`Pid<A>`](Pid)s for a homogeneous group, so
//! the follow-up send stays compile-checked. The untyped [`pick`] and
//! [`members`] are for mixed groups, where all you can rely on is identity.
//!
//! ## Groups vs. the registry
//!
//! A group is the many-actors counterpart to the [`registry`](crate::registry).
//! The registry binds a name to *at most one* actor and re-resolves it on every
//! send — what you want for a single well-known service. A group binds a name to
//! *many* actors, and one actor may sit in many groups. Reach for the registry
//! when there is exactly one of something; reach for a group when there is a set.
//!
//! ## Identity and clustering
//!
//! A group member is described by a [`Member`] — a [`Pid`] plus a [`NodeId`] and
//! an [`Incarnation`]. Today everything is single-node, those two fields are
//! fixed defaults, and you only ever pass and receive a plain [`Pid`]: the extra
//! identity is carried so this API will not have to change when groups learn to
//! span a cluster.
//!
//! ## Running context
//!
//! Every function here addresses the current runtime, so each must be called
//! from inside [`run`](crate::run) (that is, on an actor thread). Calling one
//! from outside a running runtime panics.
use crate::monitor::{demonitor, monitor, Monitor};
use crate::pid::{assert_type, Addressable, Pid};
@@ -103,7 +117,7 @@ use crate::scheduler::with_runtime;
use std::collections::HashMap;
/// A cluster node handle. A `u32` integer handle, *not* an interned atom — the
/// single deliberate divergence from the BEAM wire shape (RFC 011 names it).
/// single deliberate divergence from the BEAM wire shape.
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
pub struct NodeId(u32);
@@ -149,9 +163,8 @@ impl From<u32> for Incarnation {
}
}
/// The fixed single-node identity used until clustering (RFC 010) supplies
/// real values. Carried like `wake_slot` so the public API never has to change
/// to acquire it.
/// The fixed single-node identity used until clustering supplies real values.
/// Carried like `wake_slot` so the public API never has to change to acquire it.
pub const DEFAULT_NODE_ID: NodeId = NodeId(0);
/// The fixed incarnation for the single-node default. Non-zero so it never
/// collides with a BEAM "any creation" wildcard at interop time.
@@ -161,7 +174,7 @@ pub const DEFAULT_INCARNATION: Incarnation = Incarnation(1);
///
/// Deliberately a field-for-field image of a modern BEAM pid (`NEW_PID_EXT`).
/// In memory it is a plain struct — no wire packing; the packed representation
/// belongs to the `RemoteRef` boundary (RFC 010/011), not here.
/// belongs to the remote-reference boundary, not here.
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
pub struct Member {
/// Which node the pid lives on. `DEFAULT_NODE_ID` while single-node.
@@ -174,7 +187,7 @@ pub struct Member {
}
/// One membership: a [`Member`] and the [`Monitor`] that watches its liveness.
/// The monitor lives *alongside* the group entry (RFC 012) so a group is
/// The monitor lives *alongside* the group entry so a group is
/// self-contained: draining the membership tells us whether the member is
/// still alive, and dropping the membership drops its monitor.
struct Membership {
@@ -185,7 +198,23 @@ struct Membership {
/// The store: `name → multiset<Member>`. Within a single group a `Member`
/// appears at most once (`join` is idempotent); the *multiset* framing is for
/// cluster-readiness — the same pid is freely a member of many groups, and the
/// width admits multiples in general. Held under one Leaf-class `RawMutex`.
/// width admits multiples in general.
///
/// Locking discipline. Held under one Leaf-class `RawMutex` on `RuntimeInner`,
/// mirroring the registry, and never held together with another Leaf lock (it
/// never touches the registry or a slot's cold lock). The two operations that
/// do need another lock are kept off the group-lock path:
///
/// - `monitor()` / `demonitor()` take the target's cold lock (also Leaf), so
/// they run *before* / *after* the group lock, never under it.
/// - draining a monitor with `try_recv` takes the channel's Channel-class
/// lock, which the lock order permits *under* a Leaf; a channel critical
/// section only does the lock-free unpark protocol, so no Leaf ever nests
/// under it.
///
/// Evicted and rejected [`Monitor`]s are therefore dropped only *after* the
/// group lock is released, so a receiver-drop never runs a wakeup under the
/// lock — the same discipline as `demonitor`.
pub(crate) struct ProcessGroups {
groups: HashMap<String, Vec<Membership>>,
}
@@ -223,9 +252,11 @@ impl ProcessGroups {
/// The one dumb eviction primitive: drop every member matching `pred` from
/// every group, pruning emptied groups, and return the evicted memberships'
/// monitors for the caller to drop outside the lock. The primitive does not
/// know *why* a member leaves. Callers: the death hook (`reap_group`) and,
/// later, `evict_incarnation` (RFC 010), over the same path. Insertion
/// order within a group is preserved (`members` / `pick` are order-stable).
/// know *why* a member leaves; that is the caller's concern. Its callers are
/// the death hook (`reap_group`) and, once clustering lands, an
/// incarnation-eviction sweep — both over this same predicate path, which is
/// the whole reason to shape eviction as a predicate. Insertion order within
/// a group is preserved (`members` / `pick` are order-stable).
fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) -> Vec<Monitor> {
let mut evicted = Vec::new();
self.groups.retain(|_, v| {
@@ -242,12 +273,18 @@ impl ProcessGroups {
evicted
}
/// Drain-on-contact death hook. Drains every membership monitor in `group`
/// with a non-blocking `try_recv`: a delivered `Down` (any reason) or a
/// closed channel means that member is dead. On the first death detected,
/// sweep *all* of the dead pids out of *every* group via [`remove_where`]
/// — a death is removed from each group it joined, not just the one being
/// touched. Returns the evicted monitors to drop outside the lock.
/// Drain-on-contact death hook. The registry can prune a stale binding
/// lazily, on contact, because it only ever resolves one binding at a time;
/// a group is *iterated* — `members` fans out to everyone — so it must not
/// carry a dead member across a broadcast. Every group operation reaps the
/// group it touches first.
///
/// Drains every membership monitor in `group` with a non-blocking
/// `try_recv`: a delivered `Down` (any reason) or a closed channel means
/// that member is dead. On the first death detected, sweep *all* of the
/// dead pids out of *every* group via [`remove_where`] — a death is removed
/// from each group it joined, not just the one being touched. Returns the
/// evicted monitors to drop outside the lock.
fn reap_group(&mut self, group: &str) -> Vec<Monitor> {
let dead: Vec<Pid> = {
let Some(v) = self.groups.get(group) else {
@@ -279,7 +316,7 @@ impl ProcessGroups {
}
/// Live members of `group`, in insertion order. The `is_live` oracle is the
/// read-path backstop (Phase 3): a member whose slot is already dead is
/// read-path backstop: a member whose slot is already dead is
/// dropped from the *result* even if its `Down` has not been drained yet.
/// Backstop only — the entry stays in storage; eviction is the monitor's
/// job (`reap_group`).
@@ -314,18 +351,19 @@ fn live(inner: &crate::runtime::RuntimeInner, pid: Pid) -> bool {
/// pid is a member at most once (idempotent). Returns `true` if this call newly
/// added the membership, `false` if it was already a member.
///
/// Installs a `monitor(pid)` whose one-shot `Down` drives eviction: the
/// registration races `finalize_actor` under the slot's cold lock exactly as
/// every other monitor does, so no death slips between the join and the
/// registration. A redundant (idempotent) join tears its extra monitor back
/// down.
/// Installs a monitor on `pid` so the actor's death evicts it from the group
/// automatically — you never have to remove a dead member yourself. A redundant
/// (idempotent) join tears its extra monitor back down.
///
/// Panics if called outside `Runtime::run()`.
pub fn join<A>(group: impl Into<String>, pid: Pid<A>) -> bool {
let group = group.into();
let pid = pid.erase();
// Install the monitor BEFORE taking the group lock: monitor() acquires the
// target's cold lock (Leaf), and two Leaf locks are never held at once.
// target's cold lock (Leaf), and two Leaf locks are never held at once. The
// registration races `finalize_actor` under that cold lock exactly as every
// other monitor does, so no death can slip between the join and the monitor
// being in place.
let mon = monitor(pid);
let (rejected, reaped) = with_runtime(|inner| {
@@ -372,12 +410,13 @@ pub fn leave<A>(group: &str, pid: Pid<A>) -> bool {
}
}
/// Fan-out read: every live member of `group`.
/// Every live member of `group`, in the order they joined. Returns an empty
/// vector if the group does not exist or has no live members.
///
/// The touched group is reaped first, so dead members are evicted before the
/// read. The generation-checked liveness read is a belt-and-braces backstop:
/// even in the finalize window where a member is already dead but its `Down`
/// has not yet landed, it is dropped from the result.
/// Dead members are never returned: the group is pruned of anything that has
/// died before the read, and as a backstop a member whose slot is already dead
/// is dropped from the result even in the brief window before its death has
/// been fully processed.
///
/// Panics if called outside `Runtime::run()`.
pub fn members(group: &str) -> Vec<Pid> {
@@ -391,11 +430,10 @@ pub fn members(group: &str) -> Vec<Pid> {
pids
}
/// Pool/discovery read: one live member of `group`, or `None` if empty.
/// Stateless first-live selection: the touched group is reaped first, and the
/// generation-checked liveness backstop skips any member already dead but not
/// yet reaped. Smarter routing is an explicitly later, clustered concern per
/// RFC 010.
/// One live member of `group`, or `None` if the group is empty (or every
/// member has died). Selection is a stateless first-live scan in join order,
/// with the same dead-member backstop as [`members`]; smarter, load-aware
/// routing is a later, clustered concern.
///
/// Panics if called outside `Runtime::run()`.
pub fn pick(group: &str) -> Option<Pid> {
@@ -409,11 +447,11 @@ pub fn pick(group: &str) -> Option<Pid> {
picked
}
/// Typed `pick`: one live member of `group` as a [`Pid<A>`] (RFC 014 §4.4).
/// Typed [`pick`]: one live member of `group` as a [`Pid<A>`](Pid).
/// For a homogeneous pool every member is an `A`, so the picked member comes
/// back typed and dispatch is an ordinary compile-checked [`send_to`] rather
/// than the [`send_dyn`](crate::send_dyn) escape hatch. Re-types via the
/// unchecked [`assert_type`] primitive — a wrong `A` degrades to
/// unchecked `assert_type` primitive — a wrong `A` degrades to
/// [`SendError::NoChannel`] on the next send, never a misdelivery.
///
/// Panics if called outside `Runtime::run()`.
+6 -5
View File
@@ -193,11 +193,12 @@ impl Mailbox {
/// legal under a Leaf (Leaf -> Channel).
fn clone_sender<M: Send + 'static>(&self) -> Option<Sender<M>> {
let ch = self.channels.get(&TypeId::of::<M>())?;
let tx = ch
.sender
.as_any()
.downcast_ref::<Sender<M>>()
.expect("channel keyed by TypeId but downcast to its own type failed — smarm bug");
let tx = match ch.sender.as_any().downcast_ref::<Sender<M>>() {
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::<M>(), "msg_type / TypeId disagree");
Some(tx.clone())
}
+27 -3
View File
@@ -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<Pid> {
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())
}
}
// ---------------------------------------------------------------------------
+89 -26
View File
@@ -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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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();
+129 -39
View File
@@ -39,7 +39,10 @@ pub(crate) fn with_runtime<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> 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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> R) -> R {
/// Same preemption gate as [`with_runtime`]; same reasons.
pub(crate) fn try_with_runtime<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> R) -> Option<R> {
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<A>(supervisor: Pid<A>, 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<A: crate::pid::Addressable>(
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<A: crate::pid::Addressable>(
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<M: Send + 'static>(
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<T: Send + 'static>(
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<dyn FnOnce() -> crate::io::IoResult + Send> = Box::new(move || {
let v: T = f();
Ok(Box::new(v) as Box<dyn std::any::Any + Send>)
@@ -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::<T>().expect("block_on_io: type mismatch"),
Ok(any) => match any.downcast::<T>() {
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);
+111 -35
View File
@@ -1,16 +1,114 @@
//! Supervision signals.
//! Supervision: keep a set of actors alive.
//!
//! Every actor has a supervisor, which is itself just an actor with a
//! `Receiver<Signal>`. When a child actor terminates, the scheduler sends
//! a `Signal` on the supervisor's channel. The supervisor decides what to
//! do — restart, escalate, ignore.
//! A *supervisor* is an actor whose only job is to start a fixed set of child
//! actors and react when one of them terminates — restarting it (and, depending
//! on the strategy, some of its siblings) according to a policy, or giving up
//! when failures arrive too fast. It is how you turn "an actor that might crash"
//! into "a service that stays up": a crash becomes a restart instead of a hole
//! in the process tree.
//!
//! For v0.1 there is no built-in restart-intensity cap. That's policy and
//! lives in user code; library is mechanism only.
//! The supervisor type is [`OneForOne`]. The name is historical — the restart
//! *strategy* is selectable via [`OneForOne::strategy`], and
//! [`Strategy::OneForOne`] is merely the default. You declare the children up
//! front as [`ChildSpec`]s, each carrying a [`Restart`] policy, then hand the
//! supervision loop an actor of its own with [`OneForOne::run`].
//!
//! ## A child that crashes and recovers
//!
//! ```
//! use smarm::{run, spawn, ChildSpec, OneForOne, Restart};
//! use std::sync::Arc;
//! use std::sync::atomic::{AtomicUsize, Ordering};
//! use std::time::Duration;
//!
//! run(|| {
//! // A flaky child: it panics on its first two starts, then settles.
//! let starts = Arc::new(AtomicUsize::new(0));
//! let s = starts.clone();
//! let child = move || {
//! let n = s.fetch_add(1, Ordering::SeqCst) + 1;
//! if n < 3 {
//! panic!("boom {n}");
//! }
//! // The third start returns normally.
//! };
//!
//! // The supervisor runs on its own actor. `Transient` restarts a child
//! // that panics but treats a clean return as "done", so once the child
//! // finally succeeds the supervisor has nothing left to do and `run()`
//! // returns. smarm catches the child's panic and turns it into a restart;
//! // it never reaches the process as a real crash.
//! let sup = spawn(move || {
//! OneForOne::new()
//! .intensity(5, Duration::from_secs(60))
//! .child(ChildSpec::new(Restart::Transient, child))
//! .run();
//! });
//! sup.join().unwrap();
//!
//! assert_eq!(starts.load(Ordering::SeqCst), 3); // one start, two restarts
//! });
//! ```
//!
//! ## Restart policies
//!
//! Each child carries a [`Restart`] policy that decides whether *that child*
//! comes back when it terminates:
//!
//! - [`Restart::Permanent`] restarts on any termination, normal or panic —
//! for a service that should never be down.
//! - [`Restart::Transient`] restarts only on an abnormal exit (a panic or a
//! cooperative stop); a clean return means "done" — for work that runs to
//! completion but should be retried if it crashes.
//! - [`Restart::Temporary`] never restarts; the death is simply noted.
//!
//! ## Strategies: which siblings get cycled
//!
//! When a restart is due, the [`Strategy`] decides which *other* children are
//! cycled along with the one that died. The triggering child's own policy still
//! decides whether anything restarts at all.
//!
//! - [`Strategy::OneForOne`] restarts only the child that died — the default.
//! - [`Strategy::OneForAll`] restarts every child: the survivors are stopped,
//! then the whole set is restarted.
//! - [`Strategy::RestForOne`] restarts the dead child and every child started
//! after it, leaving earlier children untouched.
//!
//! ## Stopping a sibling is cooperative
//!
//! Cycling a sibling means stopping it first, and a supervisor never tears a
//! running actor down from outside: smarm actors share a heap and rely on
//! Drop/RAII, so unwinding a peer's stack from elsewhere would be unsound.
//! Instead the supervisor *requests* the stop and the child unwinds at its next
//! observation point — a `check!()`, an allocation, or a blocking call. A child
//! wedged in a tight loop with no observation point cannot be stopped, for the
//! same reason it cannot be preempted.
//!
//! ## Giving up: the restart-intensity cap
//!
//! A child that crashes the instant it starts would otherwise restart forever.
//! [`OneForOne::intensity`] bounds that: at most `max` restarts within any
//! `period`-long sliding window. One terminating child counts as a single
//! restart event even when the strategy cycles several siblings. When the cap
//! trips, the supervisor stops restarting, cooperatively stops any survivors in
//! reverse start order, and `run()` returns.
//!
//! ## Running context
//!
//! [`OneForOne::run`] takes over the calling actor as the supervision loop, so
//! a supervisor gets an actor of its own — typically
//! `spawn(|| OneForOne::new()/* … */.run())`, all from inside
//! [`run`](crate::run). Each child is spawned beneath the supervisor's pid, so
//! every child termination funnels back to it as a [`Signal`].
use crate::pid::Pid;
use std::any::Any;
/// A child-termination notice delivered to its supervisor.
///
/// Every child a supervisor starts is spawned beneath the supervisor's pid, so
/// each child's termination funnels back to it as one of these. The variant
/// records *how* the child went — which is what its [`Restart`] policy keys off.
pub enum Signal {
/// The child exited normally.
Exit(Pid),
@@ -42,27 +140,6 @@ impl Signal {
}
}
// ---------------------------------------------------------------------------
// One-for-one supervisor
//
// A supervisor is itself an actor. It spawns each child under its own pid so
// that every child death funnels into one mailbox (the `supervisor_channel`),
// then loops: receive a Signal, decide per the child's `Restart` policy
// whether to restart, and enforce a restart-intensity cap so a child that
// crashes in a tight loop eventually gives up instead of spinning forever.
//
// What this does NOT do: *forcibly* terminate a running child. smarm actors
// share a heap and rely on Drop/RAII, so tearing down a peer's stack from
// outside is unsound. Stopping a sibling — whether for `one_for_all` /
// `rest_for_one`, for a propagated link death, or for the ordered shutdown
// below — is therefore *cooperative*: `request_stop` flags the child and it
// unwinds at its next observation point. A child with no observation points
// (a tight loop with no `check!()`, allocation, or blocking op) cannot be
// stopped, exactly as it cannot be preempted. When the intensity cap trips,
// this supervisor stops restarting and tears the remaining children down in
// reverse start order before returning.
// ---------------------------------------------------------------------------
use crate::channel::channel;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
@@ -115,11 +192,10 @@ pub enum Strategy {
/// A supervisor over a fixed set of children.
///
/// Despite the name (kept for backwards compatibility), the restart strategy
/// is selectable via [`OneForOne::strategy`]; the default is
/// [`Strategy::OneForOne`]. Build with `new()`, add children with `child()`,
/// tune the cap with `intensity()`, then drive it with `run()` from inside an
/// actor (typically `spawn(|| OneForOne::new()....run())`).
/// Build it with [`new`](Self::new), add children with [`child`](Self::child),
/// pick a [`strategy`](Self::strategy) and an [`intensity`](Self::intensity)
/// cap, then drive the loop with [`run`](Self::run) on an actor of its own. See
/// the [module docs](self) for the full picture.
pub struct OneForOne {
children: Vec<ChildSpec>,
strategy: Strategy,
@@ -253,7 +329,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 +375,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<Pid> = Vec::with_capacity(survivors.len());
for (pid, _) in &survivors {
crate::scheduler::request_stop(*pid);
+4 -1
View File
@@ -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;
+19 -6
View File
@@ -101,7 +101,7 @@ mod inner {
thread_local! {
static LOCAL_STATE: std::cell::RefCell<Option<LocalState>> =
std::cell::RefCell::new(None);
const { std::cell::RefCell::new(None) };
}
// -----------------------------------------------------------------------
@@ -115,14 +115,20 @@ mod inner {
let (tx, rx) = mpsc::channel::<Msg>();
let start = Instant::now();
*GLOBAL.lock().unwrap() = Some(Global { sender: tx, start });
match GLOBAL.lock() {
Ok(mut g) => *g = Some(Global { sender: tx, start }),
Err(e) => panic!("smarm: trace lock poisoned (core corrupt): {e}"),
}
// Drain thread: owns the Receiver, writes to disk.
let path_for_thread = path.clone();
std::thread::Builder::new()
match std::thread::Builder::new()
.name("smarm-trace-drain".into())
.spawn(move || drain_thread(rx, &path_for_thread))
.expect("failed to spawn trace drain thread");
{
Ok(_) => {}
Err(e) => panic!("smarm: failed to spawn trace drain thread: {e}"),
}
eprintln!("[smarm-trace] writing to {}", path);
}
@@ -133,7 +139,10 @@ mod inner {
// Drop the global sender so the drain thread's recv() returns Err
// after the Flush sentinel, signalling clean shutdown.
let sender = {
let mut g = GLOBAL.lock().unwrap();
let mut g = match GLOBAL.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: trace lock poisoned (core corrupt): {e}"),
};
g.take().map(|g| g.sender)
};
if let Some(tx) = sender {
@@ -162,7 +171,11 @@ mod inner {
let mut opt = cell.borrow_mut();
// Lazily initialise: one mutex hit per thread, ever.
if opt.is_none() {
if let Some(g) = GLOBAL.lock().unwrap().as_ref() {
let guard = match GLOBAL.lock() {
Ok(g) => g,
Err(e) => panic!("smarm: trace lock poisoned (core corrupt): {e}"),
};
if let Some(g) = guard.as_ref() {
let tx = g.sender.clone();
*opt = Some(LocalState { tx, start: g.start });
}
+317 -45
View File
@@ -1,76 +1,184 @@
//! gen_statem behaviour tests, driven through the `gen_statem!` macro: cast/call
//! round-trip, `enter` firing on start and on every real transition (but not on
//! a stay), and the machine-down path when a handler panics.
//! a stay), the machine-down path when a handler panics, and the timeout
//! surface (state-timeout firing + auto-reset across transitions; named
//! timeouts surviving transitions; cancellation).
use smarm::gen_statem;
use smarm::gen_statem::{CallError, Reply};
use smarm::run;
use std::sync::{Arc, Mutex};
use std::time::Duration;
// A two-state machine: Flip toggles, calls read counters, Boom panics.
// ===========================================================================
// Timeouts
// ===========================================================================
//
// A small timer machine. `Idle` is quiet; `Armed` arms a state-timeout on entry
// that — when it fires — bumps `st_fires` and returns to `Idle`. A named
// timeout ("ping") is armed independently, survives the Idle/Armed transition,
// and bumps `named_fires` when it fires. Counters are read back over calls.
//
// Durations are tiny and waits use `smarm::sleep` (parks the actor, leaves the
// timer wheel turning). These tests run against real time, so they are a touch
// slow; a controllable clock would tighten them.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum Switch {
Off,
On,
enum T {
Idle,
Armed,
}
struct Counts {
flips: u32,
enters: u32,
struct TData {
enters: u32, // total state entries (incl. initial)
st_fires: u32, // state-timeout fires
named_fires: u32, // named-timeout fires
st_window: u64, // ms for the Armed state-timeout (set per test intent)
}
enum Cast {
Flip,
enum TCast {
Arm, // Idle -> Armed
Disarm, // Armed -> Idle (a transition that should auto-reset the state-timeout)
Ping(u64), // arm a named "ping" timeout after the given ms
CancelPing, // cancel the named "ping"
}
enum Call {
GetFlips(Reply<u32>),
GetEnters(Reply<u32>),
Boom(Reply<u32>),
enum TCall {
Enters(Reply<u32>),
StFires(Reply<u32>),
NamedFires(Reply<u32>),
Boom(Reply<u32>), // panics, to exercise the call-to-dead-machine path
}
gen_statem! {
machine: Sm { state: Switch, data: Counts };
event: Ev { cast: Cast, call: Call };
machine: TimerSm { state: T, data: TData };
event: Ev2 { cast: TCast, call: TCall, info: () };
context(data, prev, cx);
enter {
_ => data.enters += 1,
// On entering Armed, arm the state-timeout. On Idle, nothing — and the
// loop has already auto-reset any pending state-timeout on the way in.
T::Armed => { data.enters += 1; cx.state_timeout(Duration::from_millis(data.st_window)); },
T::Idle => data.enters += 1,
}
on Switch::Off => {
cast Cast::Flip => { data.flips += 1; Switch::On },
}
on Switch::On => {
cast Cast::Flip => Switch::Off,
on T::Idle => {
cast TCast::Arm => T::Armed,
cast TCast::Disarm => unhandled,
state_timeout => unhandled,
}
// State-independent queries: reply, then stay via `prev`. Boom panics
// (`boom()` is typed as a state tag so the arm stays well-formed).
on T::Armed => {
// The state-timeout elapsed while still Armed: count it and go Idle.
state_timeout => { data.st_fires += 1; T::Idle },
cast TCast::Disarm => T::Idle,
cast TCast::Arm => unhandled,
}
// State-independent rows: the named-timeout (survives transitions), the
// arm/cancel casts, the counter reads, and the panicking call.
on _ => {
call Call::GetFlips(r) => { r.reply(data.flips); prev },
call Call::GetEnters(r) => { r.reply(data.enters); prev },
call Call::Boom(_r) => boom(),
cast TCast::Ping(ms) => { cx.timeout("ping", Duration::from_millis(ms)); prev },
cast TCast::CancelPing => { cx.cancel_timeout("ping"); prev },
timeout "ping" => { data.named_fires += 1; prev },
timeout _ => unhandled,
call TCall::Enters(r) => { r.reply(data.enters); prev },
call TCall::StFires(r) => { r.reply(data.st_fires); prev },
call TCall::NamedFires(r) => { r.reply(data.named_fires); prev },
call TCall::Boom(_r) => boom(),
}
}
fn boom() -> Switch {
fn boom() -> T {
panic!("boom")
}
// Casts are applied in order and a later call observes the accumulated data.
// A state-timeout armed on entry to Armed fires after its window, bumping the
// counter and returning the machine to Idle on its own.
#[test]
fn state_timeout_fires() {
let got = Arc::new(Mutex::new(0u32));
let got2 = got.clone();
run(move || {
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 5 });
m.send(Ev2::Cast(TCast::Arm)).unwrap(); // -> Armed, arms 5ms state-timeout
smarm::sleep(Duration::from_millis(40)); // let it fire
*got2.lock().unwrap() = m.call(|r| Ev2::Call(TCall::StFires(r))).unwrap();
});
assert_eq!(*got.lock().unwrap(), 1, "state-timeout fired exactly once");
}
// Leaving Armed before the window elapses auto-resets the state-timeout: it
// must not fire afterward, even though we wait well past its original window.
#[test]
fn state_timeout_auto_resets_on_transition() {
let got = Arc::new(Mutex::new(99u32));
let got2 = got.clone();
run(move || {
// Long window so the explicit Disarm beats it comfortably.
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 50 });
m.send(Ev2::Cast(TCast::Arm)).unwrap(); // -> Armed, arms 50ms state-timeout
m.send(Ev2::Cast(TCast::Disarm)).unwrap(); // -> Idle, auto-resets it
smarm::sleep(Duration::from_millis(80)); // past the original window
*got2.lock().unwrap() = m.call(|r| Ev2::Call(TCall::StFires(r))).unwrap();
});
assert_eq!(*got.lock().unwrap(), 0, "auto-reset cancelled the pending state-timeout");
}
// A named timeout survives a state change: armed in Idle, it still fires after
// the machine has moved to Armed and back.
#[test]
fn named_timeout_survives_transition() {
let got = Arc::new(Mutex::new(0u32));
let got2 = got.clone();
run(move || {
// Armed's own state-timeout is long so it doesn't interfere.
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 200 });
m.send(Ev2::Cast(TCast::Ping(20))).unwrap(); // arm "ping" for 20ms (in Idle)
m.send(Ev2::Cast(TCast::Arm)).unwrap(); // -> Armed (ping must survive this)
m.send(Ev2::Cast(TCast::Disarm)).unwrap(); // -> Idle (and this)
smarm::sleep(Duration::from_millis(60)); // let "ping" fire
*got2.lock().unwrap() = m.call(|r| Ev2::Call(TCall::NamedFires(r))).unwrap();
});
assert_eq!(*got.lock().unwrap(), 1, "named timeout fired across the transitions");
}
// Cancelling a named timeout before its window prevents the fire.
#[test]
fn named_timeout_cancel() {
let got = Arc::new(Mutex::new(99u32));
let got2 = got.clone();
run(move || {
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 200 });
m.send(Ev2::Cast(TCast::Ping(30))).unwrap(); // arm "ping" for 30ms
m.send(Ev2::Cast(TCast::CancelPing)).unwrap(); // cancel before it fires
smarm::sleep(Duration::from_millis(60)); // past the original window
*got2.lock().unwrap() = m.call(|r| Ev2::Call(TCall::NamedFires(r))).unwrap();
});
assert_eq!(*got.lock().unwrap(), 0, "cancel prevented the named-timeout fire");
}
// ===========================================================================
// Core behaviour (cast/call round-trip, enter semantics, panic -> Down)
// ===========================================================================
// Casts are applied in order and a later call observes the resulting state via
// its counters: two Arm/Disarm round-trips leave the machine back in Idle.
#[test]
fn cast_then_call_roundtrip() {
let got = Arc::new(Mutex::new(0u32));
let got2 = got.clone();
run(move || {
let sw = Sm::start(Switch::Off, Counts { flips: 0, enters: 0 });
sw.send(Ev::Cast(Cast::Flip)).unwrap(); // Off -> On (flips = 1)
sw.send(Ev::Cast(Cast::Flip)).unwrap(); // On -> Off (no flip count)
let flips = sw.call(|r| Ev::Call(Call::GetFlips(r))).unwrap();
*got2.lock().unwrap() = flips;
// Long state-timeout window so it never fires during the test.
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 10_000 });
m.send(Ev2::Cast(TCast::Arm)).unwrap(); // Idle -> Armed (enter)
m.send(Ev2::Cast(TCast::Disarm)).unwrap(); // Armed -> Idle (enter)
m.send(Ev2::Cast(TCast::Arm)).unwrap(); // Idle -> Armed (enter)
m.send(Ev2::Cast(TCast::Disarm)).unwrap(); // Armed -> Idle (enter)
// enters = 1 (start) + 4 transitions = 5.
*got2.lock().unwrap() = m.call(|r| Ev2::Call(TCall::Enters(r))).unwrap();
});
assert_eq!(*got.lock().unwrap(), 1, "turned On once across the two flips");
assert_eq!(*got.lock().unwrap(), 5, "one enter on start, one per real transition");
}
// `enter` fires once on start and once per *real* transition; a stay (a call
@@ -80,14 +188,14 @@ fn enter_on_start_and_each_transition_but_not_stay() {
let got = Arc::new(Mutex::new((0u32, 0u32, 0u32)));
let got2 = got.clone();
run(move || {
let sw = Sm::start(Switch::Off, Counts { flips: 0, enters: 0 }); // enter -> 1
let after_start = sw.call(|r| Ev::Call(Call::GetEnters(r))).unwrap();
// Two stays (the reads) must not bump enters.
let _ = sw.call(|r| Ev::Call(Call::GetFlips(r))).unwrap();
let still = sw.call(|r| Ev::Call(Call::GetEnters(r))).unwrap();
sw.send(Ev::Cast(Cast::Flip)).unwrap(); // Off -> On -> enter -> 2
let after_flip = sw.call(|r| Ev::Call(Call::GetEnters(r))).unwrap();
*got2.lock().unwrap() = (after_start, still, after_flip);
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 10_000 }); // enter -> 1
let after_start = m.call(|r| Ev2::Call(TCall::Enters(r))).unwrap();
// A stay (a counter read returns `prev`) must not bump enters.
let _ = m.call(|r| Ev2::Call(TCall::StFires(r))).unwrap();
let still = m.call(|r| Ev2::Call(TCall::Enters(r))).unwrap();
m.send(Ev2::Cast(TCast::Arm)).unwrap(); // Idle -> Armed -> enter -> 2
let after_arm = m.call(|r| Ev2::Call(TCall::Enters(r))).unwrap();
*got2.lock().unwrap() = (after_start, still, after_arm);
});
assert_eq!(*got.lock().unwrap(), (1, 1, 2));
}
@@ -100,9 +208,173 @@ fn call_to_panicking_handler_is_down() {
let got = Arc::new(Mutex::new(None::<Result<u32, CallError>>));
let got2 = got.clone();
run(move || {
let sw = Sm::start(Switch::Off, Counts { flips: 0, enters: 0 });
let r = sw.call(|rep| Ev::Call(Call::Boom(rep)));
let m = TimerSm::start(T::Idle, TData { enters: 0, st_fires: 0, named_fires: 0, st_window: 10_000 });
let r = m.call(|rep| Ev2::Call(TCall::Boom(rep)));
*got2.lock().unwrap() = Some(r);
});
assert_eq!(*got.lock().unwrap(), Some(Err(CallError::Down)));
}
// ===========================================================================
// Postpone
// ===========================================================================
//
// Two machines. `LatchSm` defers a `Take` *call* while `Empty` and answers it
// from `Filled` — the Reply rides inside the postponed event onto the queue and
// is honoured by whichever later state handles the replay. `RelaySm` defers a
// `Mark` cast through two states so a replayed event can postpone *again*,
// landing only once the handling state is reached.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum L {
Empty,
Filled,
}
struct LData {
value: u32, // the value a Fill stored, handed back by a Take
takes: u32, // completed takes
}
enum LCast {
Fill(u32), // Empty -> Filled, storing the value
}
enum LCall {
Take(Reply<u32>), // Filled: reply value & empty; Empty: postpone until filled
Takes(Reply<u32>), // completed-take count (stay)
Status(Reply<L>), // current state tag (stay)
}
gen_statem! {
machine: LatchSm { state: L, data: LData };
event: LEv { cast: LCast, call: LCall, info: () };
context(data, prev, cx);
enter { _ => {} }
on L::Empty => {
cast LCast::Fill(v) => { data.value = v; L::Filled },
// No value yet: defer the take (with its Reply) until a Fill arrives.
call LCall::Take(r) => postpone,
}
on L::Filled => {
cast LCast::Fill(_) => unhandled, // already full
call LCall::Take(r) => { r.reply(data.value); data.takes += 1; L::Empty },
}
on _ => {
call LCall::Takes(r) => { r.reply(data.takes); prev },
call LCall::Status(r) => { r.reply(prev); prev },
state_timeout => unhandled,
timeout _ => unhandled,
}
}
// A `Take` issued while the latch is Empty parks the caller and is postponed
// (Reply included). A later Fill transitions Empty -> Filled, whose replay
// answers the deferred call — the parked caller wakes with the filled value.
#[test]
fn postponed_call_answered_after_transition() {
let got = Arc::new(Mutex::new(None::<u32>));
let g2 = got.clone();
run(move || {
let m = LatchSm::start(L::Empty, LData { value: 0, takes: 0 });
// Child actor issues the Take while Empty; its call parks on the reply.
let m2 = m.clone();
let taken = Arc::new(Mutex::new(None::<u32>));
let t2 = taken.clone();
smarm::scheduler::spawn(move || {
let v = m2.call(|r| LEv::Call(LCall::Take(r))).unwrap();
*t2.lock().unwrap() = Some(v);
});
smarm::sleep(Duration::from_millis(20)); // let the Take land and be deferred
// While deferred, the latch is untouched: still Empty, no take completed.
// (These reads are stays — they don't disturb the postponed event.)
assert_eq!(m.call(|r| LEv::Call(LCall::Status(r))).unwrap(), L::Empty);
assert_eq!(m.call(|r| LEv::Call(LCall::Takes(r))).unwrap(), 0);
m.send(LEv::Cast(LCast::Fill(42))).unwrap(); // Empty -> Filled: replays the Take
smarm::sleep(Duration::from_millis(20)); // let the child wake with its reply
*g2.lock().unwrap() = *taken.lock().unwrap();
});
assert_eq!(*got.lock().unwrap(), Some(42), "postponed call answered by the Filled state");
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum R {
S0,
S1,
S2,
}
struct RData {
marks: u32, // Marks handled (only S2 handles one)
}
enum RCast {
Go, // S0 -> S1 -> S2 -> S0
Mark, // postponed in S0/S1, handled in S2
}
enum RCall {
Marks(Reply<u32>),
State(Reply<R>),
}
gen_statem! {
machine: RelaySm { state: R, data: RData };
event: REv { cast: RCast, call: RCall, info: () };
context(data, prev, cx);
enter { _ => {} }
on R::S0 => {
cast RCast::Go => R::S1,
cast RCast::Mark => postpone,
}
on R::S1 => {
cast RCast::Go => R::S2,
cast RCast::Mark => postpone, // a replay here defers again
}
on R::S2 => {
cast RCast::Go => R::S0,
cast RCast::Mark => { data.marks += 1; prev }, // finally handled
}
on _ => {
call RCall::Marks(r) => { r.reply(data.marks); prev },
call RCall::State(r) => { r.reply(prev); prev },
state_timeout => unhandled,
timeout _ => unhandled,
}
}
// A postponed event that is replayed into a state which *also* postpones it
// re-queues, and is handled only once a state that accepts it is reached. Each
// `Marks` call is a stay that acts as a sync barrier after the preceding casts.
#[test]
fn replayed_event_can_postpone_again() {
let got = Arc::new(Mutex::new((9u32, 9u32, 9u32, R::S0)));
let g2 = got.clone();
run(move || {
let m = RelaySm::start(R::S0, RData { marks: 0 });
m.send(REv::Cast(RCast::Mark)).unwrap(); // S0: postponed
let a = m.call(|r| REv::Call(RCall::Marks(r))).unwrap(); // deferred -> 0
m.send(REv::Cast(RCast::Go)).unwrap(); // S0 -> S1: replay Mark -> postponed again
let b = m.call(|r| REv::Call(RCall::Marks(r))).unwrap(); // re-deferred -> 0
m.send(REv::Cast(RCast::Go)).unwrap(); // S1 -> S2: replay Mark -> handled
let c = m.call(|r| REv::Call(RCall::Marks(r))).unwrap(); // -> 1
let s = m.call(|r| REv::Call(RCall::State(r))).unwrap(); // Mark was a stay in S2
*g2.lock().unwrap() = (a, b, c, s);
});
let (a, b, c, s) = *got.lock().unwrap();
assert_eq!(a, 0, "deferred while S0");
assert_eq!(b, 0, "re-deferred while S1");
assert_eq!(c, 1, "handled once S2 is reached");
assert_eq!(s, R::S2, "Mark handled as a stay in S2");
}