From c0cfa01f37505e69ffed0ef0aa0cda271be7091a Mon Sep 17 00:00:00 2001 From: smarm-agent Date: Thu, 18 Jun 2026 19:13:52 +0000 Subject: [PATCH] =?UTF-8?q?gen=5Fserver:=20tick=5Fevery=20periodic=20sugar?= =?UTF-8?q?=20=E2=80=94=20loop-driven=20re-arm=20via=20factory=20+=20Sys::?= =?UTF-8?q?Tick=20(RFC=20015=20=C2=A74.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/gen_server.rs | 125 +++++++++++++++++++++++++++++++++++++++++--- tests/gen_server.rs | 45 ++++++++++++++++ 2 files changed, 162 insertions(+), 8 deletions(-) diff --git a/src/gen_server.rs b/src/gen_server.rs index b18bac3..f7485ad 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -266,6 +266,10 @@ enum Sys { /// it from the live set) and the server's payload, dispatched to /// [`GenServer::handle_timer`]. Timer(crate::timer::TimerId, G::Timer), + /// A fired periodic tick carrying its stable local id. The loop resolves the + /// payload from the registry's factory, dispatches it to + /// [`GenServer::handle_timer`], and re-arms the next tick (RFC 015 §4.3). + Tick(crate::timer::TimerId), } /// The server loop's runtime hook, passed to [`GenServer::init`]. Hands out the @@ -274,7 +278,7 @@ enum Sys { /// fields can grow without breaking. pub struct ServerCtx { sys_tx: Sender>, - reg: Arc>, + reg: Arc>>, } impl ServerCtx { @@ -305,8 +309,7 @@ impl ServerCtx { /// `Mutex` is always uncontended; it is `Arc` rather than `Rc` /// only because the handle is stored on `self` and `GenServer: Send` forces the /// handle (hence its shared state) to be `Send`. -#[derive(Default)] -struct TimerReg { +struct TimerReg { /// Monotonic minter for loop-local [`TimerId`]s — the ids handed to users, /// kept distinct from the substrate `seq` (which changes on every periodic /// re-arm). A local id is resolved only through this registry, never passed @@ -316,9 +319,47 @@ struct TimerReg { /// [`arm_after`](TimerHandle::arm_after) until the loop retires it on fire /// or [`cancel`](TimerHandle::cancel) removes it. oneshots: HashMap, + /// Live periodic timers: stable local id → its bookkeeping. The stable id + /// is minted once by [`tick_every`](TimerHandle::tick_every) and survives + /// every re-arm; the entry lives until [`cancel`](TimerHandle::cancel) + /// removes it (steady re-arm never self-removes). + periodics: HashMap>, + /// A `Sys` sender held *only while ≥1 periodic is live*, used by the loop to + /// re-arm the next tick. `Some` exactly when `periodics` is non-empty: this + /// keeps the system arm open for a server whose only system use is a + /// periodic, while leaving a non-timer server's arm to auto-close (the + /// `unused_ctx` behaviour) since this stays `None` for it. + rearm_tx: Option>>, } -impl TimerReg { +/// One periodic timer's loop-side bookkeeping. +struct Periodic { + /// Re-arm interval; each fire schedules the next at `now + every` (§4.3). + every: Duration, + /// The currently-armed substrate timer for this periodic (changes on every + /// re-arm); cancelled by [`cancel`](TimerHandle::cancel). + live: crate::timer::TimerId, + /// Produces a fresh payload for each tick. Built in `tick_every` as + /// `move || msg.clone()` — so the `Clone` bound lives there, on the one + /// method that needs it, and never leaks onto `type Timer` or the loop + /// (which is monomorphised per server and only *calls* this box). + make: Box G::Timer + Send>, +} + +// Manual Default: deriving would demand `G: Default`, but a fresh registry is +// independent of the server type. +impl Default for TimerReg { + fn default() -> Self { + TimerReg { + next_local: 0, + oneshots: HashMap::new(), + periodics: HashMap::new(), + rearm_tx: None, + } + } +} + +impl TimerReg { /// Mint the next loop-local id. fn mint(&mut self) -> TimerId { let id = TimerId::from_raw(self.next_local); @@ -334,7 +375,7 @@ impl TimerReg { /// in-flight fire) lives. pub struct TimerHandle { sys_tx: Sender>, - reg: Arc>, + reg: Arc>>, } // Manual Clone for the same reason as `Watcher`: no `G: Clone` needed. @@ -360,14 +401,54 @@ impl TimerHandle { local } - /// Cancel an armed timer. Returns the substrate's race signal — `true` if - /// the cancel beat the fire (delivery is now prevented), `false` if the - /// timer had already fired or been cancelled / is unknown. + /// Arm a periodic timer: deliver a fresh `msg` to + /// [`GenServer::handle_timer`] every `every`, until + /// [`cancel`](Self::cancel)led. Loop-managed sugar over the one-shot + /// substrate (RFC 015 §4.3): the substrate stays one-shot; the loop re-arms + /// at `now + every` after each fire and exposes one stable [`TimerId`], so a + /// single `cancel` stops the re-arm *and* the pending instance. + /// + /// Requires `Self::Timer: Clone` — a periodic re-delivers the same logical + /// message each tick, so the loop needs a fresh copy per period. The bound + /// sits on this method alone; one-shot [`arm_after`](Self::arm_after) and + /// the `type Timer` declaration stay unconstrained. + pub fn tick_every(&self, every: Duration, msg: G::Timer) -> TimerId + where + G::Timer: Clone, + { + let mut reg = self.reg.lock().unwrap(); + 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. + if reg.periodics.is_empty() { + reg.rearm_tx = Some(self.sys_tx.clone()); + } + let make: Box G::Timer + Send> = Box::new(move || msg.clone()); + // First instance fires after `every`; the payload is produced loop-side + // from `make` on fire, so the tick carries only the stable id. + let sub = send_after_to(every, self.sys_tx.clone(), Sys::Tick(local)); + reg.periodics.insert(local, Periodic { every, live: sub, make }); + local + } + + /// Cancel an armed timer (one-shot or periodic). Returns the substrate's + /// race signal — `true` if the cancel beat the (pending) fire, `false` if + /// it had already fired / been cancelled / is unknown. For a periodic this + /// also stops the re-arm: the entry is dropped, so the loop will not + /// schedule another tick even if the pending one already escaped onto the + /// channel (that in-flight tick is discarded on dispatch). pub fn cancel(&self, id: TimerId) -> bool { let mut reg = self.reg.lock().unwrap(); if let Some(sub) = reg.oneshots.remove(&id) { return cancel_timer(sub); } + if let Some(p) = reg.periodics.remove(&id) { + let beat = cancel_timer(p.live); + if reg.periodics.is_empty() { + reg.rearm_tx = None; + } + return beat; + } false } } @@ -691,6 +772,34 @@ fn server_loop( reg.lock().unwrap().oneshots.remove(&id); guard.0.handle_timer(msg); } + Ok(Some(Sys::Tick(id))) => { + // A periodic fired. Under the lock: if it is still live + // (cancel didn't beat us here), produce its payload and + // re-arm the next tick at now + every *before* dispatch, + // 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 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(); + p.live = send_after_to(every, tx, Sys::Tick(id)); + Some(msg) + } else { + // Cancelled between fire and dispatch: discard. + None + } + }; + if let Some(msg) = msg { + guard.0.handle_timer(msg); + } + } // 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"), diff --git a/tests/gen_server.rs b/tests/gen_server.rs index 164eb74..61c467c 100644 --- a/tests/gen_server.rs +++ b/tests/gen_server.rs @@ -453,6 +453,7 @@ use smarm::TimerId; enum TkCast { Arm(Duration), + Tick(Duration), CancelLast, } @@ -482,6 +483,7 @@ impl GenServer for Timed { let t = self.timer.as_ref().expect("init ran first"); match c { TkCast::Arm(d) => self.last = Some(t.arm_after(d, 7)), + TkCast::Tick(d) => self.last = Some(t.tick_every(d, 9)), TkCast::CancelLast => { let id = self.last.take().expect("a timer was armed"); *self.cancel_won.lock().unwrap() = Some(t.cancel(id)); @@ -535,3 +537,46 @@ fn cancel_before_fire_suppresses_it() { assert_eq!(*cancel_won.lock().unwrap(), Some(true), "cancel beat the fire"); assert!(fired.lock().unwrap().is_empty()); } + +// tick_every re-arms: a periodic fires repeatedly off one arm, each tick +// carrying a fresh payload. (Timer fires outrank the inbox by arm position, so +// a periodic cannot be starved by userspace traffic — the same property the +// info_outranks_inbox test pins for infos.) +#[test] +fn tick_every_rearms_repeatedly() { + let fired = Arc::new(Mutex::new(Vec::new())); + let f2 = fired.clone(); + run(move || { + let cw = Arc::new(Mutex::new(None)); + let server = start(timed(f2, cw)); + server.cast(TkCast::Tick(Duration::from_millis(20))).unwrap(); + let _ = server.call(()).unwrap(); // sync: periodic armed + smarm::sleep(Duration::from_millis(130)); // ~6 periods + let count = server.call(()).unwrap(); + assert!(count >= 3, "periodic should have re-armed several times, got {count}"); + }); + // Every tick delivered the same payload. + assert!(fired.lock().unwrap().iter().all(|&v| v == 9)); +} + +// Cancelling a periodic stops the re-arm: no further ticks land after cancel. +#[test] +fn cancel_stops_a_periodic() { + let fired = Arc::new(Mutex::new(Vec::new())); + let cancel_won = Arc::new(Mutex::new(None)); + let f2 = fired.clone(); + let c2 = cancel_won.clone(); + run(move || { + let server = start(timed(f2, c2)); + server.cast(TkCast::Tick(Duration::from_millis(20))).unwrap(); + let _ = server.call(()).unwrap(); + smarm::sleep(Duration::from_millis(70)); // a few ticks + server.cast(TkCast::CancelLast).unwrap(); + let after_cancel = server.call(()).unwrap(); // sync: cancel handled + smarm::sleep(Duration::from_millis(80)); // would be several more ticks + let later = server.call(()).unwrap(); + assert_eq!(later, after_cancel, "no ticks may land after cancel"); + }); + // The periodic had fired at least once before being cancelled. + assert!(!fired.lock().unwrap().is_empty()); +}