timer: send_after_to — channel-targeting send_after sibling (RFC 015 §5)
This commit is contained in:
@@ -432,6 +432,33 @@ pub fn send_after_named<M: Send + 'static>(
|
||||
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<T: Send + 'static>(
|
||||
after: std::time::Duration,
|
||||
tx: Sender<T>,
|
||||
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<Signal>) {
|
||||
pub fn run<F: FnOnce() + Send + 'static>(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::<u32>();
|
||||
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::<u32>();
|
||||
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::<u32>();
|
||||
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");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user