diff --git a/examples/statem_switch.rs b/examples/statem_switch.rs new file mode 100644 index 0000000..e034d4b --- /dev/null +++ b/examples/statem_switch.rs @@ -0,0 +1,152 @@ +//! A hand-written `gen_statem`, written directly against the chunk-1 primitives +//! in `smarm::statem` — no macro. This is the RFC 017 `Switch` example, and its +//! purpose is to be the **evaluation artifact**: it shows exactly the shape the +//! deferred `statem!` macro would have to generate, so we can judge what the +//! macro actually buys before committing to building one. +//! +//! Run with: `cargo run --example statem_switch` +//! +//! Sections are tagged // USER (you'd hand-write this with or without a macro) +//! and // MACRO (the boilerplate a `statem!` would emit for you). + +use smarm::run; +use smarm::statem::{self, Cx, Machine, Reply, Resolution, StatemRef}; + +// ---- USER: the four hand-written types ------------------------------------ +// In the macro world these are unchanged — the macro never generates them. + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +enum Switch { + Off, + On, +} + +struct Counts { + flips: u32, + enters: u32, +} + +enum SwitchCast { + Flip, +} + +enum SwitchCall { + GetCount(Reply), + GetEnters(Reply), +} + +// ---- MACRO: the unified event --------------------------------------------- +// The user's two message enums folded together. Chunk 1 has no internal +// variants yet; chunks 2–3 add `StateTimeout` / `Timeout(name)` here. + +enum Ev { + Cast(SwitchCast), + Call(SwitchCall), +} + +// ---- MACRO: the machine struct + state cell ------------------------------- +// `state` is the private cell; the `handle` body below is its sole writer. + +struct SwitchSm { + state: Switch, + data: Counts, +} + +impl SwitchSm { + // MACRO: `Switch::start` in the RFC; spawns the actor, hands back a ref. + fn start(init: Switch, data: Counts) -> StatemRef { + statem::spawn(SwitchSm { state: init, data }) + } + + // MACRO: the `enter` dispatch, assembled from the per-state `enter` arms. + // USER wrote the arm bodies (`data.enters += 1`); the macro wrote the match + // and the `()` return. `enter` runs side effects only — it cannot transition. + fn enter(&mut self, s: Switch, _cx: &mut Cx) { + match s { + Switch::Off => self.data.enters += 1, + Switch::On => self.data.enters += 1, + } + } +} + +impl Machine for SwitchSm { + type Ev = Ev; + + // MACRO: run the initial state's enter. + fn on_start(&mut self, cx: &mut Cx) { + let s = self.state; + self.enter(s, cx); + } + + fn handle(&mut self, ev: Ev, cx: &mut Cx) { + let prev = self.state; + + // MACRO: the `(state, event)` dispatch table. USER wrote each arm tail + // (the body + the ending state tag); the macro qualified bare tags + // (`On` -> `Switch::On`), wrapped them in `.into()`, assembled the match, + // and — in the real macro — would have edge-checked each tail tag + // against the `transitions { Off => On, On => Off }` table at expand + // time. Here that check is on the honour system. + let next: Resolution = match (self.state, ev) { + (Switch::Off, Ev::Cast(SwitchCast::Flip)) => { + self.data.flips += 1; + Switch::On.into() + } + (Switch::On, Ev::Cast(SwitchCast::Flip)) => Switch::Off.into(), + (Switch::Off, Ev::Call(SwitchCall::GetCount(r))) => { + r.reply(self.data.flips); + Switch::Off.into() // stay = return the current tag + } + (Switch::On, Ev::Call(SwitchCall::GetCount(r))) => { + r.reply(self.data.flips); + Switch::On.into() + } + (Switch::Off, Ev::Call(SwitchCall::GetEnters(r))) => { + r.reply(self.data.enters); + Switch::Off.into() + } + (Switch::On, Ev::Call(SwitchCall::GetEnters(r))) => { + r.reply(self.data.enters); + Switch::On.into() + } + // The macro would emit `_ => Resolution::Unhandled` here for a + // machine whose table is non-exhaustive; this one covers every + // (state, event) pair, so an added arm would be unreachable. + }; + + // MACRO: the resolution dispatch — identical in every generated machine. + match next { + Resolution::To(s) if s == prev => {} // stay: no enter, no reset + Resolution::To(s) => { + self.state = s; // <- sole writer of the state cell + // chunk 2: self.timers.clear_state_timeout(); + self.enter(s, cx); + // chunk 3: cx.replay(&mut self.postponed); + } + Resolution::Postpone => {} // chunk 3 + Resolution::Unhandled => cx.on_unhandled(), + } + } +} + +fn main() { + run(|| { + // USER: the client side. With a macro these would be `sw.cast(..)` / + // `sw.call(SwitchCall::GetCount)`; without it, the `Ev::Cast` / `Ev::Call` + // wrap is explicit — note how mechanical it is. + let sw = SwitchSm::start(Switch::Off, Counts { flips: 0, enters: 0 }); + + sw.send(Ev::Cast(SwitchCast::Flip)).unwrap(); // Off -> On + sw.send(Ev::Cast(SwitchCast::Flip)).unwrap(); // On -> Off + + let flips = sw.call(|r| Ev::Call(SwitchCall::GetCount(r))).unwrap(); + let enters = sw.call(|r| Ev::Call(SwitchCall::GetEnters(r))).unwrap(); + + // RFC's stated end state after two flips: Counts { flips: 1, enters: 3 } + // (initial Off entry + On entry + Off entry). + println!("after two flips: flips={flips}, enters={enters}"); + assert_eq!(flips, 1, "turned On once across the two flips"); + assert_eq!(enters, 3, "initial Off + On + Off entries"); + println!("ok"); + }); +} diff --git a/src/lib.rs b/src/lib.rs index 0552e09..cc2e1fb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,6 +27,7 @@ pub mod registry; pub mod pg; pub mod link; pub mod gen_server; +pub mod statem; pub mod introspect; #[cfg(feature = "observer")] pub mod observer; @@ -57,6 +58,10 @@ pub use gen_server::{ call, cast, shutdown, whereis_server, CallError, CallTimeoutError, CastError, GenServer, NamedServerBuilder, ServerBuilder, ServerCtx, ServerName, ServerRef, TimerHandle, Watcher, }; +pub use statem::{ + CallError as StatemCallError, Cx, Machine, Reply, Resolution, SendError as StatemSendError, + StatemRef, +}; pub use introspect::{ actor_info, snapshot, tree, tree_from, ActorInfo, ActorState, RuntimeSnapshot, RuntimeTree, TreeNode, SNAPSHOT_FORMAT_VERSION, diff --git a/src/statem.rs b/src/statem.rs new file mode 100644 index 0000000..81e58a7 --- /dev/null +++ b/src/statem.rs @@ -0,0 +1,266 @@ +//! gen_statem — generic finite state machine behaviour (RFC 017). +//! +//! The sibling of [`gen_server`](crate::gen_server): where a `gen_server` +//! carries one undifferentiated blob of state and a single `handle` that +//! re-derives "what mode am I in" on every message, a `gen_statem` makes the +//! state an explicit **tag**, routes event handling by it, and (in later +//! chunks) adds the machinery state machines need — state-entry callbacks, a +//! timeout taxonomy, and event postponement. +//! +//! ## What this layer is +//! +//! This module is the **runtime support** a state machine runs on, *not* the +//! authoring surface. A machine is any type implementing [`Machine`]: it owns +//! its state tag and data, and its [`handle`](Machine::handle) reduces a +//! `(state, event)` pair to a [`Resolution`]. The loop here drives it — spawn, +//! [`on_start`](Machine::on_start), then one [`handle`](Machine::handle) per +//! inbox event — mirroring `gen_server`'s spawn/teardown idioms. +//! +//! The RFC's `statem!` macro (deferred) would *generate* a `Machine` impl from +//! a declarative `transitions { … }` graph plus per-state handler blocks, and +//! add the expansion-time edge-lint. Until then a machine is hand-written +//! against these primitives; see `examples/statem_switch.rs` for the shape the +//! macro would target. +//! +//! ## The unified event +//! +//! A machine's [`Ev`](Machine::Ev) is the single payload its inbox carries. +//! By convention (and in the macro's desugaring) it folds the user's `cast` +//! and `call` enums together with the runtime's own internal events: +//! +//! ```ignore +//! enum Ev { Cast(MyCast), Call(MyCall) /* later: StateTimeout, Timeout(name) */ } +//! ``` +//! +//! [`StatemRef::send`] pushes any event (a cast is just a `send`); +//! [`StatemRef::call`] builds a one-shot [`Reply`] channel, hands it to a +//! `call` variant, and parks until the machine answers — exactly the +//! `gen_server` call round-trip, but with the reply handle riding *inside* the +//! user's own event so a handler can answer (or, later, postpone) it. +//! +//! ## Chunk status (RFC 017 §Sequencing) +//! +//! This is **chunk 1**: macro-free dispatch against real time — spawn, +//! `on_start`, stay/transition, and the `enter` callback (a method on the +//! machine, run on entry to a state). [`Resolution::Postpone`] and the timeout +//! arming on [`Cx`] are part of the type surface but are not yet acted on; they +//! land in chunks 2–3. + +use crate::channel::{channel, Receiver, Sender}; +use crate::pid::Pid; +use crate::scheduler::spawn as spawn_actor; +use std::marker::PhantomData; + +// --------------------------------------------------------------------------- +// Machine +// --------------------------------------------------------------------------- + +/// A finite state machine driven by the [`statem`](crate::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 +/// transition legality and, later, observer reads have a single source of +/// truth. The loop only calls [`on_start`](Self::on_start) once and +/// [`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. + type Ev: Send + 'static; + + /// 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); + + /// 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); +} + +// --------------------------------------------------------------------------- +// Resolution +// --------------------------------------------------------------------------- + +/// The outcome of handling one event, before the loop/handler acts on it: +/// dispatch reduces to `(state, event) -> Resolution` (RFC §Semantics). +/// +/// [`From`](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 +/// `To(s)` with `s == current` as stay. +pub enum Resolution { + /// 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()` (chunk 3); 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. + Unhandled, +} + +impl From for Resolution { + fn from(s: S) -> Self { + Resolution::To(s) + } +} + +// --------------------------------------------------------------------------- +// 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 (RFC §"Non-state outcomes live on cx"). +/// +/// In chunk 1 it carries only the [`on_unhandled`](Self::on_unhandled) hook; +/// `cx.state_timeout(d)` / `cx.timeout(name, d)` (chunk 2) and `cx.postpone()` +/// (chunk 3) attach here as those chunks land. It lives only on the actor's own +/// stack and is never sent. +pub struct Cx { + _ev: PhantomData Ev>, +} + +impl Cx { + fn new() -> Self { + Cx { _ev: PhantomData } + } + + /// The default for an event no arm matched: **log-and-drop**. Overridable + /// hook wiring is a follow-on (RFC Open Q5); for now an unmatched event is + /// silently dropped, as `gen_server` does with unexpected messages. + pub fn on_unhandled(&mut self) {} +} + +// --------------------------------------------------------------------------- +// Reply — the move-only reply handle a `call` variant carries +// --------------------------------------------------------------------------- + +/// The reply side of a synchronous `call`, carried *inside* the machine's own +/// `call` event variant (`GetCount(Reply)`). Move-only: answering consumes +/// it, so a handler replies at most once. Built by [`StatemRef::call`]; the +/// caller parks on the matching receiver until `reply` is invoked (or the +/// 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 (RFC §Postpone). +pub struct Reply { + tx: Sender, +} + +impl Reply { + /// Answer the call. A dropped/abandoned caller (e.g. one that timed out) + /// makes the send fail harmlessly — the machine's reply is simply + /// discarded, exactly as `gen_server`'s reply send behaves. + pub fn reply(self, value: T) { + let _ = self.tx.send(value); + } +} + +// --------------------------------------------------------------------------- +// Client handle +// --------------------------------------------------------------------------- + +/// Returned by [`StatemRef::call`] when the machine is unreachable. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum CallError { + /// The machine was already gone, or died before replying. + Down, +} + +/// Returned by [`StatemRef::send`] when the machine's inbox is closed. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum SendError { + /// The machine is gone (its inbox is closed). + Down, +} + +/// A clonable handle to a running machine. Cloning yields another sender to the +/// same inbox; the machine lives until the last `StatemRef` is dropped, at which +/// point its inbox closes and the loop exits. +pub struct StatemRef { + tx: Sender, + pid: Pid, +} + +impl Clone for StatemRef { + fn clone(&self) -> Self { + StatemRef { tx: self.tx.clone(), pid: self.pid } + } +} + +impl StatemRef { + /// The machine actor's pid — usable with `monitor`, `request_stop`, `link`. + pub fn pid(&self) -> Pid { + self.pid + } + + /// Push one event into the inbox and return immediately (fire-and-forget). + /// A cast is just a `send` of the cast-tagged event. [`SendError::Down`] if + /// the inbox is already closed. + pub fn send(&self, ev: M::Ev) -> Result<(), SendError> { + self.tx.send(ev).map_err(|_| SendError::Down) + } + + /// Synchronous request-reply. Builds a one-shot [`Reply`] channel, hands it + /// to `make` to construct the call-tagged event, sends it, and parks until + /// the machine replies — or returns [`CallError::Down`] if the machine is or + /// becomes unreachable first (the reply sender is dropped as the machine's + /// 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.) + pub fn call(&self, make: F) -> Result + where + T: Send + 'static, + F: FnOnce(Reply) -> M::Ev, + { + let (tx, rx) = channel::(); + let ev = make(Reply { tx }); + self.send(ev).map_err(|_| CallError::Down)?; + rx.recv().map_err(|_| CallError::Down) + } +} + +// --------------------------------------------------------------------------- +// Spawn + loop +// --------------------------------------------------------------------------- + +/// Spawn `machine` as an actor and hand back its [`StatemRef`]. Shape mirrors +/// `gen_server::start`: make the inbox, spawn the loop, return the ref; the +/// backing join handle is dropped (lifetime is governed by refs, not joining). +/// +/// Panics if called outside `Runtime::run()`. +pub fn spawn(machine: M) -> StatemRef { + let (tx, rx) = channel::(); + let handle = spawn_actor(move || statem_loop(rx, machine)); + StatemRef { 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. +fn statem_loop(rx: Receiver, mut machine: M) { + let mut cx = Cx::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, + } + // Observation point so a machine fed a hot inbox stays preemptible and + // cancellable. + crate::check!(); + } +} diff --git a/tests/statem.rs b/tests/statem.rs new file mode 100644 index 0000000..ef986ce --- /dev/null +++ b/tests/statem.rs @@ -0,0 +1,142 @@ +//! gen_statem (RFC 017) chunk-1 tests: call/cast round-trip against a +//! hand-written machine, `enter` firing on start and on every real transition +//! (but not on a stay), and the machine-down path when a handler panics. +//! +//! These drive the `smarm::statem` primitives directly, the same way a +//! `statem!`-generated machine eventually will. + +use smarm::run; +use smarm::statem::{self, CallError, Cx, Machine, Reply, Resolution, StatemRef}; +use std::sync::{Arc, Mutex}; + +// --------------------------------------------------------------------------- +// A hand-written two-state machine: Flip toggles, calls read, Boom panics. +// --------------------------------------------------------------------------- + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +enum Switch { + Off, + On, +} + +struct Counts { + flips: u32, + enters: u32, +} + +enum Cast { + Flip, +} + +enum Call { + GetFlips(Reply), + GetEnters(Reply), + Boom(Reply), +} + +enum Ev { + Cast(Cast), + Call(Call), +} + +struct Sm { + state: Switch, + data: Counts, +} + +impl Sm { + fn start(init: Switch) -> StatemRef { + statem::spawn(Sm { state: init, data: Counts { flips: 0, enters: 0 } }) + } + + fn enter(&mut self, _s: Switch, _cx: &mut Cx) { + self.data.enters += 1; + } +} + +impl Machine for Sm { + type Ev = Ev; + + fn on_start(&mut self, cx: &mut Cx) { + let s = self.state; + self.enter(s, cx); + } + + fn handle(&mut self, ev: Ev, cx: &mut Cx) { + let prev = self.state; + let next: Resolution = match (self.state, ev) { + (Switch::Off, Ev::Cast(Cast::Flip)) => { + self.data.flips += 1; + Switch::On.into() + } + (Switch::On, Ev::Cast(Cast::Flip)) => Switch::Off.into(), + (s, Ev::Call(Call::GetFlips(r))) => { + r.reply(self.data.flips); + s.into() // stay + } + (s, Ev::Call(Call::GetEnters(r))) => { + r.reply(self.data.enters); + s.into() // stay + } + (_, Ev::Call(Call::Boom(_r))) => panic!("boom"), + }; + match next { + Resolution::To(s) if s == prev => {} + Resolution::To(s) => { + self.state = s; + self.enter(s, cx); + } + Resolution::Postpone => {} + Resolution::Unhandled => cx.on_unhandled(), + } + } +} + +// Casts are applied in order and a later call observes the accumulated data. +#[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); + sw.send(Ev::Cast(Cast::Flip)).unwrap(); // Off -> On + sw.send(Ev::Cast(Cast::Flip)).unwrap(); // On -> Off + let flips = sw.call(|r| Ev::Call(Call::GetFlips(r))).unwrap(); + *got2.lock().unwrap() = flips; + }); + assert_eq!(*got.lock().unwrap(), 1, "turned On once across the two flips"); +} + +// `enter` fires once on start and once per *real* transition; a stay (a call +// that returns the current tag) does not re-enter. +#[test] +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); // on_start enter -> enters = 1 + let after_start = sw.call(|r| Ev::Call(Call::GetEnters(r))).unwrap(); + // Two stays (the reads above + below) 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 -> enters = 2 + let after_flip = sw.call(|r| Ev::Call(Call::GetEnters(r))).unwrap(); + *got2.lock().unwrap() = (after_start, still, after_flip); + }); + assert_eq!(*got.lock().unwrap(), (1, 1, 2)); +} + +// A handler that panics tears the loop down; the in-flight call's reply channel +// closes as the stack unwinds, so the parked caller wakes with Down (mirrors +// gen_server's panicking-handler path). +#[test] +fn call_to_panicking_handler_is_down() { + let got = Arc::new(Mutex::new(None::>)); + let got2 = got.clone(); + run(move || { + let sw = Sm::start(Switch::Off); + let r = sw.call(|rep| Ev::Call(Call::Boom(rep))); + *got2.lock().unwrap() = Some(r); + }); + assert_eq!(*got.lock().unwrap(), Some(Err(CallError::Down))); +}