gen_server: drop-guard drains live timers + debug_assert no leak (RFC 015 §4.7)
This commit is contained in:
+33
-13
@@ -708,10 +708,30 @@ fn server_loop<G: GenServer>(
|
|||||||
mut infos: Vec<Receiver<G::Info>>,
|
mut infos: Vec<Receiver<G::Info>>,
|
||||||
) {
|
) {
|
||||||
// terminate() must run on every exit path (clean close, panic, stop), so it
|
// terminate() must run on every exit path (clean close, panic, stop), so it
|
||||||
// lives in this guard's Drop rather than after the loop.
|
// lives in this guard's Drop rather than after the loop. The guard also owns
|
||||||
struct Terminate<G: GenServer>(G);
|
// a registry handle so the *same* drop drains every live timer (RFC 015
|
||||||
|
// §4.7): once the loop is gone no periodic can re-arm anyway, but cancelling
|
||||||
|
// here keeps no armed substrate entry lingering past the server, and the
|
||||||
|
// assert pins the invariant.
|
||||||
|
struct Terminate<G: GenServer>(G, Arc<Mutex<TimerReg<G>>>);
|
||||||
impl<G: GenServer> Drop for Terminate<G> {
|
impl<G: GenServer> Drop for Terminate<G> {
|
||||||
fn drop(&mut self) {
|
fn drop(&mut self) {
|
||||||
|
{
|
||||||
|
let mut reg = self.1.lock().unwrap();
|
||||||
|
for (_, sub) in reg.oneshots.drain() {
|
||||||
|
cancel_timer(sub);
|
||||||
|
}
|
||||||
|
for (_, p) in reg.periodics.drain() {
|
||||||
|
cancel_timer(p.live);
|
||||||
|
}
|
||||||
|
reg.rearm_tx = None;
|
||||||
|
// A fired-but-gone send already no-ops on the closed Sys channel;
|
||||||
|
// this guards the armed-but-unfired re-arming case specifically.
|
||||||
|
debug_assert!(
|
||||||
|
reg.oneshots.is_empty() && reg.periodics.is_empty(),
|
||||||
|
"an armed timer survived gen_server loop exit"
|
||||||
|
);
|
||||||
|
}
|
||||||
self.0.terminate();
|
self.0.terminate();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -728,18 +748,18 @@ fn server_loop<G: GenServer>(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let mut guard = Terminate(state);
|
// Shared timer bookkeeping: the loop keeps one clone (to retire one-shots on
|
||||||
|
// fire, re-arm periodics, and read the idle window); the guard holds another
|
||||||
// The system intake arm: Watchers feed Monitors and (next chunk) armed
|
// to drain on exit; every TimerHandle the state stores holds more.
|
||||||
// timers feed fires to the loop through it. The ctx (and with it the loop's
|
|
||||||
// own sender) drops right after init — a state that didn't clone the
|
|
||||||
// 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>>();
|
|
||||||
// Shared timer bookkeeping: the loop keeps one clone (to retire one-shots
|
|
||||||
// on fire, and to drain on exit); every TimerHandle the state stores holds
|
|
||||||
// another.
|
|
||||||
let reg = Arc::new(Mutex::new(TimerReg::default()));
|
let reg = Arc::new(Mutex::new(TimerReg::default()));
|
||||||
|
let mut guard = Terminate(state, reg.clone());
|
||||||
|
|
||||||
|
// The system intake arm: Watchers feed Monitors and armed timers feed fires
|
||||||
|
// to the loop through it. The ctx (and with it the loop's own sender) drops
|
||||||
|
// right after init — a state that cloned no Watcher/TimerHandle 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>>();
|
||||||
// Bind the ctx so the idle window set during init can be read back, then
|
// Bind the ctx so the idle window set during init can be read back, then
|
||||||
// drop it — that drops the loop's own Sys sender, so a state that cloned no
|
// drop it — that drops the loop's own Sys sender, so a state that cloned no
|
||||||
// Watcher/TimerHandle lets the arm auto-close (the unused-ctx behaviour).
|
// Watcher/TimerHandle lets the arm auto-close (the unused-ctx behaviour).
|
||||||
|
|||||||
@@ -646,3 +646,27 @@ fn traffic_resets_the_idle_window() {
|
|||||||
assert_eq!(*before_quiet.lock().unwrap(), 0, "steady traffic must suppress idle");
|
assert_eq!(*before_quiet.lock().unwrap(), 0, "steady traffic must suppress idle");
|
||||||
assert!(*idles.lock().unwrap() >= 1, "idle fires once the inbox falls quiet");
|
assert!(*idles.lock().unwrap() >= 1, "idle fires once the inbox falls quiet");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RFC 015 §4.7 — no armed timer survives loop exit. A server with a live
|
||||||
|
// periodic is dropped; the loop's drop guard drains and cancels it (a surviving
|
||||||
|
// re-arming timer would trip the in-Drop debug_assert), runs terminate, and no
|
||||||
|
// further tick lands after exit.
|
||||||
|
#[test]
|
||||||
|
fn no_timer_survives_exit() {
|
||||||
|
let fired = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
let f_server = fired.clone();
|
||||||
|
let f_read = fired.clone();
|
||||||
|
run(move || {
|
||||||
|
let server = start(timed(f_server, Arc::new(Mutex::new(None))));
|
||||||
|
server.cast(TkCast::Tick(Duration::from_millis(15))).unwrap();
|
||||||
|
let _ = server.call(()).unwrap(); // sync: periodic armed
|
||||||
|
smarm::sleep(Duration::from_millis(45)); // a couple of ticks
|
||||||
|
let mon = smarm::monitor(server.pid());
|
||||||
|
drop(server); // inbox closes → loop exits → guard drains timers
|
||||||
|
// Clean Down ⇒ the loop returned without the no-leak assert aborting.
|
||||||
|
assert!(mon.rx.recv().is_ok());
|
||||||
|
let at_exit = f_read.lock().unwrap().len();
|
||||||
|
smarm::sleep(Duration::from_millis(90)); // would be several more ticks
|
||||||
|
assert_eq!(f_read.lock().unwrap().len(), at_exit, "no tick may fire after exit");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user