gen_server: TimerHandle arm_after/cancel + Sys::Timer dispatch (RFC 015 §4.3, §5)

src/gen_server.rs

@@ -53,8 +53,12 @@ use crate::channel::{channel, select, Receiver, Selectable, Sender};
 use crate::monitor::{demonitor, monitor, Down, Monitor};
 use crate::pid::Pid;
 use crate::registry::{register_with, resolve_named_sender, RegisterError};
-use crate::scheduler::{request_stop, spawn, spawn_under};
+use crate::scheduler::{cancel_timer, request_stop, send_after_to, spawn, spawn_under};
+use crate::timer::TimerId;
+use std::collections::HashMap;
 use std::marker::PhantomData;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
 
 /// Behaviour for a gen_server: a state value plus call/cast handlers.
 ///
@@ -261,28 +265,110 @@ enum Sys<G: GenServer> {
     /// A fired one-shot timer carrying its loop-local id (so the loop can retire
     /// it from the live set) and the server's payload, dispatched to
     /// [`GenServer::handle_timer`].
-    // Constructed by `TimerHandle::arm_after` in the next RFC 015 chunk; until
-    // then only the `Watch` variant is produced.
-    #[allow(dead_code)]
     Timer(crate::timer::TimerId, G::Timer),
 }
 
-/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently
-/// carries only the [`Watcher`]; opaque so fields can grow without breaking.
+/// The server loop's runtime hook, passed to [`GenServer::init`]. Hands out the
+/// loop's two clonable intake handles — the [`Watcher`] (monitors) and the
+/// [`TimerHandle`] (timers) — plus the one-shot idle-window setter. Opaque, so
+/// fields can grow without breaking.
 pub struct ServerCtx<G: GenServer> {
-    watcher: Watcher<G>,
+    sys_tx: Sender<Sys<G>>,
+    reg: Arc<Mutex<TimerReg>>,
 }
 
 impl<G: GenServer> ServerCtx<G> {
     /// A clonable handle to the loop's monitor intake. Store it in the state
     /// during `init` to watch monitors from later handlers.
     pub fn watcher(&self) -> Watcher<G> {
-        self.watcher.clone()
+        Watcher { tx: self.sys_tx.clone() }
     }
 
     /// Shorthand for `ctx.watcher().watch(m)` when watching during `init`.
     pub fn watch(&self, m: Monitor) {
-        self.watcher.watch(m)
+        self.watcher().watch(m)
+    }
+
+    /// A clonable handle to the loop's timer intake (the [`Watcher`] pattern,
+    /// for time). Store it on the state during `init` to
+    /// [`arm_after`](TimerHandle::arm_after) /
+    /// [`tick_every`](TimerHandle::tick_every) /
+    /// [`cancel`](TimerHandle::cancel) from any later handler.
+    pub fn timer(&self) -> TimerHandle<G> {
+        TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() }
+    }
+}
+
+/// Per-server timer bookkeeping, shared between the loop and every
+/// [`TimerHandle`] clone. A gen_server actor is single-threaded — handle calls
+/// (made from inside handlers) and the loop never run concurrently — so this
+/// `Mutex` is always uncontended; it is `Arc<Mutex>` rather than `Rc<RefCell>`
+/// 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 {
+    /// 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
+    /// to [`cancel_timer`], so the two id roles never mix.
+    next_local: u64,
+    /// Live one-shot timers: local id → current substrate id. Present from
+    /// [`arm_after`](TimerHandle::arm_after) until the loop retires it on fire
+    /// or [`cancel`](TimerHandle::cancel) removes it.
+    oneshots: HashMap<TimerId, crate::timer::TimerId>,
+}
+
+impl TimerReg {
+    /// Mint the next loop-local id.
+    fn mint(&mut self) -> TimerId {
+        let id = TimerId::from_raw(self.next_local);
+        self.next_local = self.next_local.wrapping_add(1);
+        id
+    }
+}
+
+/// Arms, cancels, and (chunk 4) periodically ticks server timers, the time-side
+/// twin of [`Watcher`] (RFC 015 §4.3). Handed out by [`ServerCtx::timer`] in
+/// `init` and stored on the state; later handlers arm timers without any change
+/// to their signatures. Clonable; the arm stays open while any clone (or an
+/// in-flight fire) lives.
+pub struct TimerHandle<G: GenServer> {
+    sys_tx: Sender<Sys<G>>,
+    reg: Arc<Mutex<TimerReg>>,
+}
+
+// Manual Clone for the same reason as `Watcher`: no `G: Clone` needed.
+impl<G: GenServer> Clone for TimerHandle<G> {
+    fn clone(&self) -> Self {
+        TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() }
+    }
+}
+
+impl<G: GenServer> TimerHandle<G> {
+    /// Arm a one-shot timer: deliver `msg` to [`GenServer::handle_timer`] after
+    /// `after`, unless [`cancel`](Self::cancel)led first. Returns a loop-local
+    /// [`TimerId`] for cancellation. A thin wrapper over the `send_after`
+    /// substrate that lands the fire on the loop's own system arm (above infos
+    /// and the inbox), not in the inbox.
+    pub fn arm_after(&self, after: Duration, msg: G::Timer) -> TimerId {
+        let mut reg = self.reg.lock().unwrap();
+        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).
+        let sub = send_after_to(after, self.sys_tx.clone(), Sys::Timer(local, msg));
+        reg.oneshots.insert(local, sub);
+        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.
+    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);
+        }
+        false
     }
 }
 
@@ -549,7 +635,11 @@ fn server_loop<G: GenServer>(
     // Watcher closes the arm, the first select observes the closure, and the
     // loop falls back to the plain-inbox park.
     let (sys_tx, sys_rx) = channel::<Sys<G>>();
-    guard.0.init(&ServerCtx { watcher: Watcher { tx: sys_tx } });
+    // Shared timer bookkeeping: the loop keeps one clone (to retire one-shots
+    // on fire, and — later — to drain on exit); every TimerHandle the state
+    // stores holds another.
+    let reg = Arc::new(Mutex::new(TimerReg::default()));
+    guard.0.init(&ServerCtx { sys_tx, reg: reg.clone() });
 
     let mut monitors: Vec<Monitor> = Vec::new();
     let mut sys_open = true;
@@ -594,7 +684,13 @@ fn server_loop<G: GenServer>(
             } else if i < nd + nw {
                 match sys_rx.try_recv() {
                     Ok(Some(Sys::Watch(m))) => monitors.push(m),
-                    Ok(Some(Sys::Timer(_id, msg))) => guard.0.handle_timer(msg),
+                    Ok(Some(Sys::Timer(id, msg))) => {
+                        // 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);
+                        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"),