diff --git a/examples/gen_statem_expanded.rs b/examples/gen_statem_expanded.rs index 9a8a7c6..b31aa7f 100644 --- a/examples/gen_statem_expanded.rs +++ b/examples/gen_statem_expanded.rs @@ -36,7 +36,7 @@ #![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,12 +58,14 @@ enum Cast { Pull, Lock, Unlock(u32), // carries a key + Knock, // counted when the door is reachable; postponed while Locked } enum Call { GetState(Reply), GetEnters(Reply), GetPushes(Reply), + GetKnocks(Reply), } enum Ev { @@ -120,7 +123,7 @@ impl DoorSm { fn start(init: Door) -> GenStatemRef { spawn(DoorSm { state: init, - data: Data { enters: 0, pushes: 0 }, + data: Data { enters: 0, pushes: 0, knocks: 0 }, }) } @@ -150,18 +153,32 @@ impl Machine for DoorSm { self.enter(cx); } - fn handle(&mut self, ev: Ev, cx: &mut Cx) { + fn handle(&mut self, ev: Ev, cx: &mut Cx) -> Step { 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 = 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 } @@ -169,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 } @@ -192,6 +217,10 @@ 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), @@ -209,14 +238,17 @@ impl Machine for DoorSm { // ---- 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(), } } } @@ -226,9 +258,10 @@ 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::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 @@ -241,11 +274,13 @@ fn main() { 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); // auto-closed by the state-timeout assert_eq!(enters, 5); // Closed(start) + Locked + Closed + Open + Closed assert_eq!(pushes, 0); // no push ever closed it this run + assert_eq!(knocks, 1); // the locked-door knock, replayed once unlocked println!("ok"); }); } diff --git a/examples/gen_statem_macro.rs b/examples/gen_statem_macro.rs index 0f920a9..f9f9e50 100644 --- a/examples/gen_statem_macro.rs +++ b/examples/gen_statem_macro.rs @@ -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), GetEnters(Reply), GetPushes(Reply), + GetKnocks(Reply), } const CODE: u32 = 1234; @@ -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,6 +125,7 @@ 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, @@ -126,23 +135,26 @@ gen_statem! { 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"); }); } diff --git a/src/gen_statem.rs b/src/gen_statem.rs index 5dce943..1fa44fe 100644 --- a/src/gen_statem.rs +++ b/src/gen_statem.rs @@ -57,13 +57,23 @@ //! the corresponding internal event and runs it through the normal `handle` //! dispatch. //! -//! [`Resolution::Postpone`] is part of the type surface but is not yet wired up. +//! ## 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, select, Receiver, Sender}; use crate::pid::Pid; use crate::scheduler::{cancel_timer, send_after_to, spawn as spawn_actor}; use crate::timer::TimerId; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::marker::PhantomData; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -100,20 +110,26 @@ pub trait Machine: Send + 'static { /// 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); + /// 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) -> Step; } // --------------------------------------------------------------------------- // Resolution // --------------------------------------------------------------------------- -/// The outcome of handling one event, before the loop/handler acts on it: -/// dispatch reduces to `(state, event) -> Resolution`. +/// The outcome of the **consuming** dispatch — the by-value `match (state, +/// event)` — before the apply-tail acts on it: `(state, event) -> +/// Resolution`. 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`](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 @@ -122,10 +138,6 @@ 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()`; 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. @@ -138,6 +150,24 @@ impl From for Resolution { } } +/// 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 { + /// 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 // --------------------------------------------------------------------------- @@ -403,12 +433,19 @@ pub fn spawn(machine: M) -> GenStatemRef { /// 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(rx: Receiver, mut machine: M) { let (sys_tx, sys_rx) = channel::(); 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 = VecDeque::new(); machine.on_start(&mut cx); loop { // Timer arm first: a ready fire is taken in preference to the inbox. @@ -442,7 +479,7 @@ fn statem_loop(rx: Receiver, mut machine: M) { } }; if let Some(ev) = ev { - machine.handle(ev, &mut cx); + dispatch(&mut machine, &mut cx, &mut postpone, ev); } } // Single-receiver: nothing can drain the arm between select's @@ -454,7 +491,7 @@ fn statem_loop(rx: Receiver, mut machine: M) { } } else { match rx.try_recv() { - Ok(Some(ev)) => machine.handle(ev, &mut cx), + 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, @@ -466,6 +503,51 @@ fn statem_loop(rx: Receiver, 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( + machine: &mut M, + cx: &mut Cx, + postpone: &mut VecDeque, + 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(machine: &mut M, cx: &mut Cx, postpone: &mut VecDeque) { + 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 (total-match dispatch) // --------------------------------------------------------------------------- @@ -567,6 +649,9 @@ fn statem_loop(rx: Receiver, mut machine: M) { /// // * 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: @@ -625,6 +710,16 @@ fn statem_loop(rx: Receiver, mut machine: M) { /// 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 /// @@ -704,138 +799,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 (catch-all-free for cast/call/timeouts) match. - // The one macro-injected arm is the `Info` silent-drop fallback (3c): info - // is out-of-band and defaults to a drop, matching `gen_server`. It is last - // and broadest, so per-state `info` rows above it stay reachable. Cast, - // call, and both timeout events get NO fallback — a forgotten pair is still - // a non-exhaustive-match error. - (@arms ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ]) => { - match ($ss, $se) { - $($arms)* - (_, $Ev::Info(_)) => $crate::gen_statem::Resolution::Unhandled, + // ===== @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)* ] ($st:pat) + (@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)* ] ($st:pat) + (@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) - (@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat) + // 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)* ] ($st:pat) + (@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) - (@rows ($Ev:ident) ($ss:expr, $se:expr) [ $($arms:tt)* ] ($st:pat) + // 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)* ] ($st:pat) + (@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)*) }; } diff --git a/tests/gen_statem.rs b/tests/gen_statem.rs index bb2b8a8..4f03fd8 100644 --- a/tests/gen_statem.rs +++ b/tests/gen_statem.rs @@ -214,3 +214,167 @@ fn call_to_panicking_handler_is_down() { }); 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), // Filled: reply value & empty; Empty: postpone until filled + Takes(Reply), // completed-take count (stay) + Status(Reply), // 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::)); + 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::)); + 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), + State(Reply), +} + +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"); +}