diff --git a/src/scheduler.rs b/src/scheduler.rs index 9fddb08..215680b 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -432,6 +432,33 @@ pub fn send_after_named( with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, armed_by, fire)) } +/// Deliver `msg` onto a channel the caller owns after `after`, rather than to a +/// registry address. The sibling of [`send_after`] used by the gen_server timer +/// layer (RFC 015 §5): arming a server timer must land the fire on the loop's +/// own `Sys` channel — giving it the loop's arm position — not in the inbox. +/// +/// Same substrate as [`send_after`]: a `Reason::Send` entry, the same `armed` +/// set, the same [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. Only +/// the fire thunk differs — `tx.send(msg)` instead of a registry resolve — so a +/// send onto a channel whose receiver is gone is dropped, exactly as a failed +/// address resolve is (Erlang `send_after` semantics). `pid` is informational +/// (who armed it); delivery lives entirely in the thunk. +// Wired into `TimerHandle::arm_after` in the RFC 015 timer chunk; until then it +// is referenced only from tests, so the non-test build would flag it dead. +#[allow(dead_code)] +pub(crate) fn send_after_to( + after: std::time::Duration, + tx: Sender, + msg: T, +) -> crate::timer::TimerId { + let deadline = crate::timer::deadline_from_now(after); + let armed_by = current_pid().unwrap_or(Pid::new(0, 0)); + let fire = Box::new(move || { + let _ = tx.send(msg); + }); + with_runtime(|inner| inner.timers.lock().unwrap().insert_send(deadline, armed_by, fire)) +} + /// Cancel a timer armed by [`send_after`] / [`send_after_named`]. Returns /// `true` if it was still pending (delivery now prevented), `false` if it had /// already fired or been cancelled. @@ -701,3 +728,54 @@ pub fn register_supervisor_channel(pid: Pid, sender: Sender) { pub fn run(f: F) { crate::runtime::init(crate::runtime::Config::exact(1)).run(f); } + +#[cfg(test)] +mod send_after_to_tests { + use super::*; + use crate::channel::channel; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + use std::time::Duration; + + // A fired send_after_to lands its message on the caller's own channel. + #[test] + fn delivers_onto_the_channel() { + run(|| { + let (tx, rx) = channel::(); + send_after_to(Duration::from_millis(10), tx, 42); + // recv parks until the scheduler fires the timer thunk. + assert_eq!(rx.recv().unwrap(), 42); + }); + } + + // cancel_timer before the deadline prevents delivery and reports the race + // win; the channel then closes with no message once the lone sender (moved + // into the now-discarded thunk) is gone. + #[test] + fn cancel_prevents_delivery() { + run(|| { + let (tx, rx) = channel::(); + let id = send_after_to(Duration::from_millis(50), tx, 7); + assert!(cancel_timer(id), "cancel before fire should win the race"); + // No delivery: the discarded thunk drops the only sender, so recv + // sees a closed channel rather than the value. + assert!(rx.recv().is_err()); + }); + } + + // The thunk runs on the scheduler thread; a closed receiver makes the send + // a harmless no-op (Erlang send_after semantics) rather than a panic. + #[test] + fn send_to_closed_channel_is_harmless() { + let reached = Arc::new(AtomicBool::new(false)); + let r2 = reached.clone(); + run(move || { + let (tx, rx) = channel::(); + send_after_to(Duration::from_millis(10), tx, 1); + drop(rx); // receiver gone before the timer fires + crate::sleep(Duration::from_millis(30)); + r2.store(true, Ordering::SeqCst); + }); + assert!(reached.load(Ordering::SeqCst), "runtime survived the dead-channel fire"); + } +}