ntex_util/time/callbacks.rs
1use std::num::NonZeroUsize;
2use std::{cell::Cell, cmp, fmt, rc::Rc, rc::Weak};
3
4use crate::HashMap;
5use crate::time::Millis;
6
7use super::wheel::{TIMER, TimerEntry};
8
9/// A reusable, local one-shot callback timer.
10///
11/// Each construction or reset arms one invocation of the retained callback.
12/// The callback may capture `Rc` state; this handle is neither `Send` nor `Sync`
13/// and cannot be cloned. Keep the handle alive until the callback should run.
14/// Dropping it cancels pending delivery, including an expired callback that has
15/// not started, and releases its captures. An active invocation retains its
16/// captures until it finishes.
17///
18/// Expiration and cancellation retain the callback for reuse. The callback
19/// implements `Fn`, so use interior mutability for state that changes between
20/// invocations.
21///
22/// Callbacks run outside wheel access and may create, reset, cancel or drop
23/// timers, including themselves. They must be short, synchronous and
24/// non-blocking. Callbacks due on the same tick have no guaranteed order.
25///
26/// Callback panics propagate through the runtime. They may stop the shared
27/// driver and disarm other timers; remaining callbacks need not be delivered.
28/// Runtime shutdown suppresses pending delivery. A surviving handle retains
29/// its callback and can be explicitly reset in a new runtime on the same thread.
30/// Starting a new runtime does not automatically rearm the handle.
31///
32/// # Examples
33///
34/// Retain local state across invocations, cancel an arm, then reuse the timer:
35///
36/// ```
37/// use std::{cell::Cell, rc::Rc};
38/// use ntex::time::{CallbackTimer, Millis, sleep};
39///
40/// #[ntex::main]
41/// async fn main() {
42/// let calls = Rc::new(Cell::new(0));
43/// let count = calls.clone();
44/// let timer = CallbackTimer::new(Millis::ZERO, move || {
45/// count.set(count.get() + 1);
46/// });
47/// assert_eq!(calls.get(), 0); // Zero duration never invokes inline.
48/// sleep(Millis(50)).await;
49/// assert!(timer.is_elapsed());
50/// assert_eq!(calls.get(), 1);
51///
52/// timer.reset(Millis(1_000));
53/// timer.cancel();
54/// assert!(timer.is_elapsed());
55///
56/// timer.reset(Millis::ZERO);
57/// sleep(Millis(50)).await;
58/// assert_eq!(calls.get(), 2);
59/// }
60/// ```
61pub struct CallbackTimer {
62 pub(super) hnd: NonZeroUsize,
63 pub(super) state: Rc<CallbackState>,
64}
65
66impl CallbackTimer {
67 /// Arms a callback after `duration`, using the wheel's rounding and limits.
68 ///
69 /// The base granularity is approximately 16 ms; longer delays use coarser
70 /// rounding, and delays beyond the wheel's capacity (about 24 days) are
71 /// clamped. Callback delivery may be delayed further while the runtime is busy.
72 ///
73 /// A zero duration waits for a subsequent timer tick, as with [`crate::time::Sleep`];
74 /// the callback is never invoked inline.
75 pub fn new(duration: Millis, callback: impl Fn() + 'static) -> Self {
76 let state = Rc::new(CallbackState {
77 callback: Box::new(callback),
78 arm: Cell::new(0),
79 pending: Cell::new(true),
80 });
81 TIMER.with(|timer| {
82 let (idx, expiry) = timer.calc_bucket(u64::from(cmp::max(duration.0, 1)));
83 let no = timer.with_wheel(|wheel| {
84 let no = wheel.timers.insert(TimerEntry {
85 bucket: None,
86 bucket_entry: 0,
87 callback: true,
88 task: None,
89 });
90 wheel
91 .callbacks
92 .get_or_insert_with(|| Box::new(Callbacks::default()))
93 .entries
94 .insert(
95 no,
96 CallbackTicket {
97 state: Rc::downgrade(&state),
98 arm: 0,
99 generation: timer.generation.get(),
100 },
101 );
102 wheel.link(no, idx);
103 no
104 });
105 let handle = Self {
106 hnd: NonZeroUsize::new(no).unwrap(),
107 state,
108 };
109 timer.update_next_expiry(expiry);
110 handle
111 })
112 }
113
114 /// Arms another invocation, invalidating any previous pending delivery.
115 ///
116 /// This also suppresses an expired invocation that has been queued but has
117 /// not started. It can be called before or after expiration or cancellation.
118 ///
119 /// A zero duration waits for a subsequent tick. Resetting from within the
120 /// callback arms a new invocation, which survives completion of the old one.
121 pub fn reset(&self, duration: Millis) {
122 TIMER.with(|timer| {
123 let (idx, expiry) = timer.calc_bucket(u64::from(cmp::max(duration.0, 1)));
124 timer.with_wheel(|wheel| {
125 let ticket = wheel
126 .callbacks
127 .as_mut()
128 .unwrap()
129 .entries
130 .get_mut(&self.hnd.get())
131 .unwrap();
132 ticket.arm = ticket.arm.wrapping_add(1);
133 ticket.generation = timer.generation.get();
134 self.state.arm.set(ticket.arm);
135 self.state.pending.set(true);
136 wheel.relink(self.hnd.get(), idx);
137 });
138 timer.update_next_expiry(expiry);
139 });
140 }
141
142 /// Disarms the timer and suppresses pending delivery, retaining the callback.
143 ///
144 /// This also suppresses an expired invocation that has not started. Calling
145 /// this method on an already disarmed timer is harmless.
146 ///
147 /// An invocation that has already started cannot be interrupted.
148 pub fn cancel(&self) {
149 self.state.pending.set(false);
150 TIMER.with(|timer| timer.with_wheel(|wheel| wheel.unlink(self.hnd.get())));
151 }
152
153 /// Returns `true` when disarmed, including after cancellation or shutdown.
154 ///
155 /// Expiration marks the timer elapsed before callback delivery. This query
156 /// does not indicate whether the callback has run.
157 pub fn is_elapsed(&self) -> bool {
158 TIMER.with(|timer| timer.with_wheel(|wheel| wheel.timers[self.hnd.get()].bucket.is_none()))
159 }
160}
161
162impl fmt::Debug for CallbackTimer {
163 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
164 f.debug_struct("CallbackTimer")
165 .field("handle", &self.hnd)
166 .finish_non_exhaustive()
167 }
168}
169
170impl Drop for CallbackTimer {
171 fn drop(&mut self) {
172 self.state.pending.set(false);
173 let _ = TIMER.try_with(|timer| {
174 timer.with_wheel(|wheel| {
175 wheel.unlink(self.hnd.get());
176 wheel.timers.remove(self.hnd.get());
177 wheel
178 .callbacks
179 .as_mut()
180 .unwrap()
181 .entries
182 .remove(&self.hnd.get());
183 });
184 });
185 }
186}
187
188pub(super) struct CallbackState {
189 pub(super) callback: Box<dyn Fn()>,
190 pub(super) arm: Cell<u64>,
191 pub(super) pending: Cell<bool>,
192}
193
194#[derive(Clone)]
195pub(super) struct CallbackTicket {
196 // Weak identity prevents slab reuse from redirecting a delivery and lets
197 // handle drop release captures without waiting for the queued ticket.
198 pub(super) state: Weak<CallbackState>,
199 pub(super) arm: u64,
200 pub(super) generation: u64,
201}
202
203#[derive(Default)]
204pub(super) struct Callbacks {
205 pub(super) entries: HashMap<usize, CallbackTicket>,
206 pub(super) ready: Vec<CallbackTicket>,
207}