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.
This commit is contained in:
smarm-agent
@@ -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<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");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user