windows_threadpool_sys/timer.rs
1// Copyright (c) 2026 Mike Grier
2//! Thread-pool timers: `CreateThreadpoolTimer` / `SetThreadpoolTimer` /
3//! `WaitForThreadpoolTimerCallbacks` / `CloseThreadpoolTimer`.
4//!
5//! Two types share this machinery, and they differ in the one property that
6//! matters when writing the callback:
7//!
8//! - [`ThreadpoolTimer`] fires **exactly once per arming**, and re-arming from
9//! inside its own callback is applied only after that callback returns, so
10//! repetition driven that way never overlaps itself.
11//! - [`ThreadpoolPeriodicTimer`] repeats on a fixed period, and the pool may queue the
12//! next callback **while the previous one is still running**. Its callback
13//! must tolerate overlapping with itself.
14//!
15//! The platform models both with one object and a `period` argument. This crate
16//! separates them so that concurrency contract belongs to a type rather than to
17//! an argument that is easy to skim past.
18//!
19//! # Choosing between them
20//!
21//! Want a fixed cadence, and the callback is short or safe to overlap? Use
22//! [`ThreadpoolPeriodicTimer`]. Want the next delay measured from when the previous
23//! callback *finished*, with no overlap possible? Use a [`ThreadpoolTimer`] and re-arm it
24//! from inside its own callback with [`TimerFiring::rearm_after`].
25//!
26//! # Due times
27//!
28//! Relative due times ([`ThreadpoolTimer::set_after`]) count only time the system is
29//! awake, so sleep and hibernation do not consume the delay. Absolute due times
30//! ([`ThreadpoolTimer::set_at`]) name a wall-clock instant, which sleep and hibernation
31//! *do* pass through: a timer set for an instant that elapsed while the machine
32//! slept fires promptly on resume.
33
34mod periodic;
35
36pub use periodic::{PeriodicTick, ThreadpoolPeriodicTimer};
37
38use std::cell::Cell;
39use std::io;
40use std::ptr;
41use std::sync::atomic::{AtomicIsize, Ordering};
42use std::time::{Duration, SystemTime, UNIX_EPOCH};
43
44use crate::rearm::RearmSuppression;
45/// Only the test-only re-arm observer still needs this directly; the
46/// suppression count moved to [`RearmSuppression`].
47#[cfg(test)]
48use std::sync::Mutex;
49
50use windows_sys::Win32::Foundation::{FALSE, FILETIME, TRUE};
51use windows_sys::Win32::System::Threading::{
52 CloseThreadpoolTimer, CreateThreadpoolTimer, IsThreadpoolTimerSet, PTP_CALLBACK_INSTANCE,
53 PTP_TIMER, SetThreadpoolTimer, WaitForThreadpoolTimerCallbacks,
54};
55
56use crate::callback_env::CallbackEnviron;
57
58/// Conversion constants for Windows `FILETIME`, which counts 100-nanosecond
59/// intervals since 1601-01-01 UTC. Changing any value is a breaking change.
60mod filetime {
61 /// 100-nanosecond intervals per second.
62 pub const TICKS_PER_SECOND: u64 = 10_000_000;
63 /// Nanoseconds per 100-nanosecond interval.
64 pub const NANOS_PER_TICK: u32 = 100;
65 /// Seconds between the `FILETIME` epoch (1601-01-01) and the Unix epoch.
66 pub const SECONDS_1601_TO_1970: u64 = 11_644_473_600;
67}
68
69/// Split a 64-bit tick count into the `FILETIME` field pair.
70fn filetime_from_ticks(ticks: i64) -> FILETIME {
71 let bits = ticks as u64;
72 FILETIME {
73 dwLowDateTime: bits as u32,
74 dwHighDateTime: (bits >> 32) as u32,
75 }
76}
77
78/// Convert a delay into the negative tick count that means "relative" to
79/// `SetThreadpoolTimer`.
80///
81/// Saturates rather than overflowing: a delay beyond the representable range
82/// becomes the furthest representable relative time, far past any practical
83/// process lifetime.
84pub(crate) fn relative_filetime(delay: Duration) -> FILETIME {
85 let ticks = delay
86 .as_secs()
87 .saturating_mul(filetime::TICKS_PER_SECOND)
88 .saturating_add(u64::from(delay.subsec_nanos() / filetime::NANOS_PER_TICK));
89 let ticks = i64::try_from(ticks).unwrap_or(i64::MAX);
90 filetime_from_ticks(-ticks)
91}
92
93/// Convert a wall-clock instant into the positive tick count that means
94/// "absolute" to `SetThreadpoolTimer`.
95///
96/// Instants at or before the Unix epoch clamp to zero, which the pool treats as
97/// immediately due.
98pub(crate) fn absolute_filetime(when: SystemTime) -> FILETIME {
99 let since_unix = when.duration_since(UNIX_EPOCH).unwrap_or(Duration::ZERO);
100 let ticks = since_unix
101 .as_secs()
102 .saturating_add(filetime::SECONDS_1601_TO_1970)
103 .saturating_mul(filetime::TICKS_PER_SECOND)
104 .saturating_add(u64::from(
105 since_unix.subsec_nanos() / filetime::NANOS_PER_TICK,
106 ));
107 filetime_from_ticks(i64::try_from(ticks).unwrap_or(i64::MAX))
108}
109
110/// Saturate a coalescing window to the `u32` millisecond field.
111///
112/// This is the one length in either crate that saturates rather than being
113/// rejected, and deliberately so. A window is a permission -- "you may delay
114/// this firing by up to this much to batch it with others" -- and the pool is
115/// always free to fire earlier, so a saturated window asks for less coalescing
116/// rather than producing a wrong result. A truncated *buffer* silently loses
117/// data, which is why those are rejected instead.
118///
119/// Periods also pass through this, but cannot reach the saturation: they are
120/// validated against
121/// [`MAX_PERIOD`](crate::timer::ThreadpoolPeriodicTimer::MAX_PERIOD) at
122/// construction.
123pub(crate) fn millis_u32(duration: Duration) -> u32 {
124 u32::try_from(duration.as_millis()).unwrap_or(u32::MAX)
125}
126
127/// Arm a raw timer object.
128///
129/// The one record here covers both timer kinds, because both reach
130/// `SetThreadpoolTimer` through this function: `period_ms` is zero for a
131/// one-shot arming and non-zero for a periodic one, so the two are told apart
132/// by the record's own second slot rather than by separate targets.
133///
134/// SAFETY: `timer` must be a live `PTP_TIMER`.
135pub(crate) unsafe fn arm_raw(timer: PTP_TIMER, due: FILETIME, period_ms: u32, window_ms: u32) {
136 // SAFETY: forwarded from this function's contract; `due` is read only for
137 // the duration of the call.
138 crate::trace_call!("SetThreadpoolTimer", timer, period_ms, {
139 // SAFETY: forwarded; `due` is read only for the duration of the call.
140 unsafe { SetThreadpoolTimer(timer, &due, period_ms, window_ms) };
141 });
142 crate::trace_record!("timer", "armed", timer, period_ms);
143}
144
145/// Stop a raw timer object.
146///
147/// SAFETY: `timer` must be a live `PTP_TIMER`.
148pub(crate) unsafe fn disarm_raw(timer: PTP_TIMER) {
149 // SAFETY: forwarded; a null due time is the documented way to stop a timer.
150 crate::trace_call!("SetThreadpoolTimer(disarm)", timer, 0, {
151 // SAFETY: forwarded; a null due time stops the timer.
152 unsafe { SetThreadpoolTimer(timer, ptr::null(), 0, 0) };
153 });
154 crate::trace_record!("timer", "disarmed", timer);
155}
156
157/// Heap-allocated callback state kept alive for the lifetime of the timer.
158///
159/// `timer` is filled in after `CreateThreadpoolTimer` returns, because re-arming
160/// from inside a callback needs the object the callback belongs to.
161pub(crate) struct TimerContext {
162 pub(crate) timer: AtomicIsize,
163 /// Stops a callback re-arming this timer once a teardown has begun.
164 ///
165 /// Applying a deferred re-arm takes this lock and does nothing while the
166 /// count is non-zero. Deferring the re-arm to after the callback returns --
167 /// which is what makes the delay run from the end of the firing -- moves it
168 /// *past* any disarm performed from outside, so without this a drain could
169 /// complete with a due time installed. Its two users here are
170 /// [`ThreadpoolTimer::stop_and_drain`], which raises and lowers it, and
171 /// `Drop`, which raises it permanently -- the asymmetry
172 /// [`RearmSuppression`] is a count for. The lock is only ever held across
173 /// the native `SetThreadpoolTimer` call.
174 rearm: RearmSuppression,
175 /// Whether the timer has a due time that nothing has drained.
176 ///
177 /// A firing settles it because this type is one-shot: each arming produces
178 /// exactly one callback, so once that callback has entered, no further one
179 /// is coming. `IsThreadpoolTimerSet` is no help here -- it stays true after
180 /// a one-shot expires, so it cannot tell a timer that already fired from one
181 /// still waiting to. A callback that re-arms sets this again when the
182 /// deferred request is applied.
183 obligation: crate::obligation::CloseObligation,
184
185 /// This pool's entry in the self-heal registry.
186 ///
187 /// Held for its `Drop`, not read. The claim keeps the entry alive while
188 /// this object exists, which is what makes the entry's pre-created repair
189 /// work object available -- and bound to the pool, deferring its free --
190 /// if a wait on the same pool is later cancelled. Only a wait reads a
191 /// registration, because only a wait reaches the removal primitive that
192 /// owes a repair.
193 #[allow(dead_code)]
194 registration: crate::heal::Registration,
195 /// Records, for tests, whether each deferred re-arm was actually applied.
196 ///
197 /// The suppression this observes happens after the callback returns and
198 /// before the context is freed, so no user-reachable state can witness it;
199 /// this Arc is cloned by the test, which therefore outlives the context.
200 #[cfg(test)]
201 rearm_observer: Mutex<Option<std::sync::Arc<Mutex<Vec<bool>>>>>,
202 callback: Box<dyn Fn(&TimerFiring<'_>) + Send + Sync + 'static>,
203}
204impl TimerContext {
205 /// Lock the suppression count, recovering from a panicking holder.
206 fn suppression(&self) -> std::sync::MutexGuard<'_, u32> {
207 self.rearm.lock()
208 }
209
210 /// Start suppressing re-arming, and disarm under the same acquisition.
211 ///
212 /// Doing both under one lock is what makes the pair atomic against a
213 /// callback: a deferred re-arm either lands entirely before this, or is
214 /// suppressed by it. The lock is released before any drain.
215 fn suppress_and_disarm(&self) {
216 self.rearm.suppress_and(|suppressed| {
217 let timer = self.timer.load(Ordering::Acquire);
218 crate::trace_record!("timer", "suppress-and-disarm", timer, suppressed);
219 if timer == 0 {
220 return;
221 }
222 // SAFETY: `timer` is this object's live PTP_TIMER, published before
223 // any callback could run and valid until Drop closes it.
224 unsafe { disarm_raw(timer) };
225 });
226 }
227
228 /// Stop suppressing re-arming.
229 fn release_suppression(&self) {
230 self.rearm.release();
231 }
232}
233
234/// A re-arming a callback asked for, applied once the callback has returned.
235///
236/// Applying it immediately would start the delay from the moment of the call
237/// rather than from the end of the firing, which is both what the API documents
238/// and what keeps firings from overlapping: a callback that re-armed early and
239/// then ran longer than its delay could be entered again concurrently.
240#[derive(Clone, Copy)]
241enum PendingRearm {
242 After(Duration),
243 At(SystemTime),
244}
245
246/// One firing of a [`ThreadpoolTimer`], handed to its callback.
247///
248/// The timer is not armed while the callback runs, so re-arming from here is
249/// what produces repetition whose delay is measured from the *end* of this
250/// callback -- repetition that can never overlap itself.
251pub struct TimerFiring<'ctx> {
252 ctx: &'ctx TimerContext,
253 /// What the callback asked for, applied by the trampoline after it returns.
254 ///
255 /// A `Cell` rather than a lock because the firing is borrowed only by the
256 /// one callback invocation that owns it.
257 pending: Cell<Option<PendingRearm>>,
258}
259
260impl std::fmt::Debug for TimerFiring<'_> {
261 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
262 f.debug_struct("TimerFiring").finish_non_exhaustive()
263 }
264}
265
266impl TimerFiring<'_> {
267 /// Arm the timer again to fire once, `delay` from now.
268 ///
269 /// The arming is applied after this callback returns, so `delay` is measured
270 /// from the **end** of this firing regardless of where in the callback it is
271 /// requested. That is what keeps successive firings strictly sequential: a
272 /// callback that re-armed at its start and then ran longer than `delay`
273 /// would otherwise be entered again while still running.
274 ///
275 /// Calling this more than once in a firing keeps the last request.
276 pub fn rearm_after(&self, delay: Duration) {
277 self.pending.set(Some(PendingRearm::After(delay)));
278 // The request and its application are separated by the rest of the
279 // callback, so both ends are stamped: this says the callback asked, and
280 // `rearm-entered` later says the trampoline began acting on it.
281 crate::trace_record!(
282 "timer",
283 "rearm-requested",
284 self.ctx.timer.load(Ordering::Acquire),
285 delay.as_millis() as u64
286 );
287 }
288
289 /// Arm the timer again to fire once at the wall-clock instant `when`.
290 ///
291 /// Like [`TimerFiring::rearm_after`], this is applied after the callback
292 /// returns. An instant that has already passed by then fires immediately.
293 pub fn rearm_at(&self, when: SystemTime) {
294 self.pending.set(Some(PendingRearm::At(when)));
295 crate::trace_record!(
296 "timer",
297 "rearm-requested",
298 self.ctx.timer.load(Ordering::Acquire)
299 );
300 }
301
302 /// Apply whatever the callback asked for, once it has returned.
303 ///
304 /// Suppressed once teardown has begun, so a request made during the last
305 /// callback cannot re-arm the timer behind `Drop`'s disarm.
306 fn apply_pending(&self) {
307 let applied = self.apply_pending_reporting();
308 #[cfg(test)]
309 if let Some(applied) = applied {
310 let observer = self
311 .ctx
312 .rearm_observer
313 .lock()
314 .unwrap_or_else(|poison| poison.into_inner())
315 .clone();
316 if let Some(observer) = observer {
317 observer
318 .lock()
319 .unwrap_or_else(|poison| poison.into_inner())
320 .push(applied);
321 }
322 }
323 let _ = applied;
324 }
325
326 /// Apply the pending re-arm, reporting whether it was actually installed.
327 ///
328 /// `None` means the callback asked for nothing; `Some(false)` means it asked
329 /// but teardown suppressed the request.
330 fn apply_pending_reporting(&self) -> Option<bool> {
331 let pending = self.pending.get()?;
332 // Recorded before the lock, not after: the acquisition can block on a
333 // concurrent `suppress_and_disarm`, and a deferred re-arm parked on that
334 // mutex is one of the things a silent interval in the trace could be.
335 // This is the timer's counterpart to the wait's `rearm-entered`, and it
336 // sits here rather than in `rearm_after` because the request and its
337 // application are separated by the whole of the callback -- the request
338 // is stamped separately as `rearm-requested`.
339 let timer = self.ctx.timer.load(Ordering::Acquire);
340 crate::trace_record!("timer", "rearm-entered", timer);
341 // Taken before arming and held across it, so this either happens before
342 // a suppressing caller raises the count or is suppressed by it -- never
343 // in between.
344 let suppressed = self.ctx.suppression();
345 if *suppressed > 0 {
346 crate::trace_record!("timer", "rearm-suppressed", timer, *suppressed);
347 return Some(false);
348 }
349 debug_assert_ne!(timer, 0, "the timer must be published before callbacks");
350 let due = match pending {
351 PendingRearm::After(delay) => relative_filetime(delay),
352 PendingRearm::At(when) => absolute_filetime(when),
353 };
354 // Recorded before the arming is published, not after: by here the
355 // suppression check has already returned, so the arming is certain to
356 // happen, and the firing it produces can settle the obligation on
357 // another thread before this one gets any further. See
358 // `CloseObligation::record_live_before`.
359 self.ctx.obligation.record_live_before(|| {
360 // SAFETY: `timer` is this object's live PTP_TIMER, published before any
361 // callback could run.
362 unsafe { arm_raw(timer, due, 0, 0) };
363 });
364 drop(suppressed);
365 crate::trace_record!("timer", "rearm-left", timer);
366 Some(true)
367 }
368}
369
370/// Trampoline from the raw `PTP_TIMER_CALLBACK` ABI into the boxed closure.
371///
372/// SAFETY: `context` must point to a live [`TimerContext`] for the entire
373/// duration of every callback invocation, which [`ThreadpoolTimer`]'s `Drop` ordering
374/// guarantees.
375unsafe extern "system" fn timer_trampoline(
376 _instance: PTP_CALLBACK_INSTANCE,
377 context: *mut core::ffi::c_void,
378 _timer: PTP_TIMER,
379) {
380 // SAFETY: context is a valid *mut TimerContext for the full callback duration.
381 let ctx = unsafe { &*(context as *const TimerContext) };
382 crate::trace_record!("timer", "trampoline-entered", _timer);
383 // This firing consumed the arming: a one-shot produces exactly one callback
384 // per arming, so nothing is owed unless the callback re-arms below.
385 ctx.obligation.record_settled();
386 // Stamped before the callback, not after: a dispatch that is still running
387 // is evidence the pool is live, and a long callback must not look like
388 // silence to the self-heal.
389 let firing = TimerFiring {
390 ctx,
391 pending: Cell::new(None),
392 };
393 // Not contained: see the callback contract in the crate docs.
394 (ctx.callback)(&firing);
395 // Applied only now that the callback has returned, so a requested delay runs
396 // from the end of this firing and the next one cannot overlap it.
397 firing.apply_pending();
398 crate::trace_record!("timer", "trampoline-left", _timer);
399}
400
401/// An owned one-shot thread-pool timer.
402///
403/// Each arming produces exactly one firing. Arm it with
404/// [`ThreadpoolTimer::set_after`] or [`ThreadpoolTimer::set_at`], and stop it
405/// with [`ThreadpoolTimer::disarm`]. Arming again replaces the previous setting
406/// rather than adding to it.
407///
408/// For repetition, either re-arm from inside the callback with
409/// [`TimerFiring::rearm_after`] -- which keeps firings strictly sequential -- or
410/// use [`ThreadpoolPeriodicTimer`] when a fixed cadence matters more than
411/// avoiding overlap.
412///
413/// # When firings can overlap
414///
415/// Re-arming through [`TimerFiring`] never overlaps: the request is applied
416/// after the callback returns, so the next firing cannot begin until this one
417/// has finished. That is the intended way to repeat.
418///
419/// Arming from *outside* the callback is a different matter. Calling
420/// [`ThreadpoolTimer::set_after`] while a callback is running can queue the next
421/// firing before the current one returns, and the two then run concurrently on
422/// different pool threads. The callback is `Fn + Sync`, so this is permitted
423/// rather than unsound -- but it means a callback that assumes it is the only
424/// one running must not be driven that way. Re-arm from the callback, or use
425/// [`ThreadpoolTimer::disarm`] and [`ThreadpoolTimer::wait`] before re-arming
426/// externally.
427///
428/// [`Drop`] disarms before draining callbacks, so the captured closure stays
429/// valid for the full lifetime of every callback execution.
430///
431/// # Examples
432///
433/// Fire once:
434///
435/// ```
436/// use std::sync::mpsc;
437/// use std::time::Duration;
438/// use windows_threadpool_sys::timer::ThreadpoolTimer;
439///
440/// let (tx, rx) = mpsc::channel();
441/// let sender = std::sync::Mutex::new(tx);
442///
443/// let timer = ThreadpoolTimer::new(move |_firing| {
444/// let _ = sender.lock().expect("send").send(());
445/// }, None)?;
446///
447/// timer.set_after(Duration::from_millis(10));
448/// rx.recv().expect("the timer should fire");
449/// # Ok::<(), std::io::Error>(())
450/// ```
451///
452/// Repeat without ever overlapping, by re-arming from inside the callback. The
453/// gap is measured from the end of each firing, so a slow callback delays the
454/// next one instead of racing it:
455///
456/// ```
457/// use std::sync::Arc;
458/// use std::sync::atomic::{AtomicUsize, Ordering};
459/// use std::time::Duration;
460/// use windows_threadpool_sys::timer::ThreadpoolTimer;
461///
462/// let ticks = Arc::new(AtomicUsize::new(0));
463/// let counter = Arc::clone(&ticks);
464///
465/// let timer = ThreadpoolTimer::new(move |firing| {
466/// // Stop after three firings by simply not re-arming.
467/// if counter.fetch_add(1, Ordering::SeqCst) < 2 {
468/// firing.rearm_after(Duration::from_millis(1));
469/// }
470/// }, None)?;
471///
472/// timer.set_after(Duration::from_millis(1));
473/// while ticks.load(Ordering::SeqCst) < 3 {
474/// std::thread::yield_now();
475/// }
476/// timer.disarm();
477/// timer.wait();
478/// assert_eq!(ticks.load(Ordering::SeqCst), 3);
479/// # Ok::<(), std::io::Error>(())
480/// ```
481pub struct ThreadpoolTimer {
482 timer: PTP_TIMER,
483 // Kept alive as a raw pointer until Drop has disarmed and drained.
484 context: *mut TimerContext,
485}
486
487// SAFETY: PTP_TIMER is a cross-thread pool object, and the context holds a
488// callback that is Fn + Send + Sync; the pointer is only read until Drop frees
489// it after all callbacks have finished.
490unsafe impl Send for ThreadpoolTimer {}
491unsafe impl Sync for ThreadpoolTimer {}
492
493impl ThreadpoolTimer {
494 /// Create an idle timer that invokes `callback` each time it expires.
495 ///
496 /// Pass `Some(env)` to select a private pool or callback priority; `None`
497 /// uses the process-default pool with default priority.
498 ///
499 /// The callback runs on a shared, process-managed pool thread. It must
500 /// restore any thread state it changes and must not terminate its thread. It
501 /// must not panic: a panic unwinds to the `extern "system"` trampoline and
502 /// aborts the process.
503 ///
504 /// # Errors
505 ///
506 /// Returns the error from `CreateThreadpoolTimer`.
507 pub fn new<F>(callback: F, env: Option<&mut CallbackEnviron>) -> io::Result<Self>
508 where
509 F: Fn(&TimerFiring<'_>) + Send + Sync + 'static,
510 {
511 // Read before `env` is consumed below, and registered before the object
512 // exists: the entry must be able to repair this pool from the moment
513 // anything of ours can dispatch on it.
514 let registration = crate::heal::register(crate::heal::key_of(env.as_deref()));
515 let context = Box::into_raw(Box::new(TimerContext {
516 timer: AtomicIsize::new(0),
517 rearm: RearmSuppression::new(),
518 obligation: crate::obligation::CloseObligation::new(),
519 registration,
520 #[cfg(test)]
521 rearm_observer: Mutex::new(None),
522 callback: Box::new(callback),
523 }));
524 let env_ptr = env.map_or(ptr::null_mut(), |e| e.as_mut_ptr());
525
526 // SAFETY: context is a valid heap pointer that outlives every callback,
527 // and env_ptr is valid (or null) for the duration of this call.
528 let timer = crate::trace_call!("CreateThreadpoolTimer", 0, 0, {
529 // SAFETY: context is a valid heap pointer that outlives every callback,
530 // and env_ptr is valid (or null) for the duration of this call.
531 unsafe {
532 CreateThreadpoolTimer(Some(timer_trampoline), context.cast(), env_ptr.cast_const())
533 }
534 });
535
536 if timer == 0 {
537 let error = io::Error::last_os_error();
538 // SAFETY: the pool never saw context; reclaim it immediately.
539 unsafe { drop(Box::from_raw(context)) };
540 return Err(error);
541 }
542
543 // Publish the object before any callback can run. The timer is not armed
544 // yet, so no callback can observe the unpublished value.
545 // SAFETY: context is live and exclusively ours until the first arming.
546 unsafe { (*context).timer.store(timer, Ordering::Release) };
547 crate::trace_record!("timer", "created", timer);
548
549 Ok(Self { timer, context })
550 }
551
552 /// Fire once, `delay` from now.
553 ///
554 /// The delay counts only time the system is awake. A zero delay makes the
555 /// timer due immediately.
556 pub fn set_after(&self, delay: Duration) {
557 self.record_live_before(|| {
558 // SAFETY: timer is valid for the lifetime of self.
559 unsafe { arm_raw(self.timer, relative_filetime(delay), 0, 0) };
560 });
561 }
562
563 /// Fire once at the wall-clock instant `when`.
564 ///
565 /// Unlike a relative due time, an absolute one passes through sleep and
566 /// hibernation: if `when` elapses while the machine is asleep, the timer
567 /// fires promptly on resume. An instant already in the past fires
568 /// immediately.
569 pub fn set_at(&self, when: SystemTime) {
570 self.record_live_before(|| {
571 // SAFETY: timer is valid for the lifetime of self.
572 unsafe { arm_raw(self.timer, absolute_filetime(when), 0, 0) };
573 });
574 }
575
576 /// Fire once after `delay`, allowing the system a coalescing `window`.
577 ///
578 /// `window` is the tolerance the system may add to the due time so it can
579 /// group this timer with other expirations and wake the processor less
580 /// often. A larger window trades timing precision for power.
581 pub fn set_after_with_window(&self, delay: Duration, window: Duration) {
582 self.record_live_before(|| {
583 // SAFETY: timer is valid for the lifetime of self.
584 unsafe { arm_raw(self.timer, relative_filetime(delay), 0, millis_u32(window)) };
585 });
586 }
587
588 /// Note that an arming has made this timer live, so `Drop` owes a drain.
589 ///
590 /// One site for the three `set_*` methods, which differ only in how they
591 /// compute a due time.
592 fn record_live_before(&self, publish: impl FnOnce()) {
593 // SAFETY: the context outlives every callback and is freed only by Drop,
594 // which cannot run while this borrow of self is alive.
595 unsafe { &*self.context }
596 .obligation
597 .record_live_before(publish);
598 }
599
600 /// Stop the timer.
601 ///
602 /// New callbacks stop being queued, and a callback already queued is
603 /// discarded -- measured by
604 /// `disarming_cancels_a_queued_tick_which_a_waits_disarm_does_not`, which
605 /// holds the pool's only thread busy so a tick is provably queued, then
606 /// disarms. This is the asymmetry with a wait, whose disarm leaves a queued
607 /// callback to run. Disarming an idle timer is a no-op.
608 ///
609 /// A callback already *executing* is unaffected; use
610 /// [`wait`](Self::wait) or [`stop_and_drain`](Self::stop_and_drain) to
611 /// block until it finishes.
612 pub fn disarm(&self) {
613 // SAFETY: timer is valid for the lifetime of self.
614 unsafe { disarm_raw(self.timer) };
615 }
616
617 /// Record, into `observer`, whether each deferred re-arm is actually applied.
618 ///
619 /// The caller keeps its own clone, so the record survives this timer's
620 /// teardown -- which is the only moment a re-arm is suppressed.
621 #[cfg(test)]
622 pub(crate) fn observe_rearms(&self, observer: &std::sync::Arc<Mutex<Vec<bool>>>) {
623 // SAFETY: the context outlives self; Drop frees it after the drain.
624 let ctx = unsafe { &*self.context };
625 *ctx.rearm_observer
626 .lock()
627 .unwrap_or_else(|poison| poison.into_inner()) = Some(std::sync::Arc::clone(observer));
628 }
629 /// Whether the timer currently has a due time.
630 ///
631 /// This reports whether the timer has been armed and not since disarmed. It
632 /// is **not** a prediction that the timer will fire again: expiring does not
633 /// clear the due time, so a fired timer still reports `true`. Only
634 /// [`ThreadpoolTimer::disarm`] makes it `false`.
635 #[must_use]
636 pub fn is_set(&self) -> bool {
637 // SAFETY: timer is valid for the lifetime of self.
638 crate::trace_call!("IsThreadpoolTimerSet", self.timer, 0, {
639 // SAFETY: timer is valid for the lifetime of self.
640 unsafe { IsThreadpoolTimerSet(self.timer) != 0 }
641 })
642 }
643
644 /// Let every queued callback run, and block until none is executing.
645 ///
646 /// This does **not** leave a self-re-arming timer idle. A callback's
647 /// [`TimerFiring::rearm_after`] is applied after the callback returns, so a
648 /// firing that runs during this call installs a fresh due time and the timer
649 /// is armed again when it returns. Use
650 /// [`stop_and_drain`](Self::stop_and_drain) to reach quiescence.
651 pub fn wait(&self) {
652 // SAFETY: timer is valid for the lifetime of self.
653 crate::trace_call!("WaitForThreadpoolTimerCallbacks", self.timer, 0, {
654 // SAFETY: timer is valid for the lifetime of self.
655 unsafe { WaitForThreadpoolTimerCallbacks(self.timer, FALSE) };
656 });
657 }
658
659 /// Drop callbacks that have not started, then wait for any executing one.
660 ///
661 /// Like [`wait`](Self::wait), this does not by itself leave a self-re-arming
662 /// timer idle: it does not suppress the deferred re-arm of a callback that
663 /// is already running. Use [`stop_and_drain`](Self::stop_and_drain) when the
664 /// timer must actually be quiescent afterwards.
665 pub fn cancel_pending(&self) {
666 // SAFETY: timer is valid for the lifetime of self. A cancelled timer
667 // callback owns no storage, so dropping queued callbacks orphans nothing.
668 crate::trace_call!("WaitForThreadpoolTimerCallbacks(cancel)", self.timer, 1, {
669 // SAFETY: timer is valid for the lifetime of self.
670 unsafe { WaitForThreadpoolTimerCallbacks(self.timer, TRUE) };
671 });
672 }
673
674 /// Stop the timer and block until it is idle, leaving it reusable.
675 ///
676 /// This exists because neither [`disarm`](Self::disarm) nor
677 /// [`cancel_pending`](Self::cancel_pending) can stop a self-re-arming timer
678 /// on its own: a callback already running requests its re-arm through
679 /// [`TimerFiring::rearm_after`], and the trampoline applies it *after* the
680 /// callback returns -- which is after any disarm from outside. This
681 /// suppresses that deferred re-arm for the duration of the call, using the
682 /// same mechanism `Drop` uses, and lifts the suppression before returning so
683 /// the timer can be armed again afterwards.
684 ///
685 /// # What this guarantees
686 ///
687 /// On return, provided no other thread arms the timer during the call:
688 ///
689 /// - no callback is queued or executing, and
690 /// - the timer has no due time -- a re-arm requested by a callback that ran
691 /// during the call is discarded rather than deferred.
692 ///
693 /// # What it does not
694 ///
695 /// **A concurrent arm from another thread is not excluded.** `ThreadpoolTimer`
696 /// is `Sync` and [`set_after`](Self::set_after), [`set_at`](Self::set_at) and
697 /// [`set_after_with_window`](Self::set_after_with_window) all take `&self`,
698 /// so they do not pass through the suppression this uses. Nothing in this
699 /// crate orders such a call against this one.
700 ///
701 /// In practice the drain currently cancels a due time installed that way --
702 /// `WaitForThreadpoolTimerCallbacks` with cancellation clears one even when
703 /// no callback is queued, measurably so. That is not a documented contract
704 /// and is not relied upon here: if a caller needs the timer to be provably
705 /// idle, it must ensure nothing else arms it for the duration, by owning it
706 /// exclusively or serializing access to it.
707 ///
708 /// Calling this from inside the timer's own callback would deadlock, because
709 /// it waits for that callback to finish.
710 pub fn stop_and_drain(&self) {
711 // SAFETY: the context outlives every callback and is freed only by Drop,
712 // which cannot run while this borrow of self is alive.
713 unsafe { Self::stop_and_drain_parts(self.context.cast(), self.timer) };
714 }
715
716 /// [`stop_and_drain`](Self::stop_and_drain) against a detached context.
717 ///
718 /// One body rather than two, so this type and the cleanup-group member that
719 /// wraps the same object cannot drift apart: the suppression discipline here
720 /// is what makes the drain mean anything, and a second copy of it is a
721 /// second place for a later change to reach one and miss the other.
722 ///
723 /// # Safety
724 ///
725 /// `context` must come from [`into_parts`](Self::into_parts) on this type
726 /// and name a still-live object whose context the caller has not yet freed,
727 /// and `timer` must be that object.
728 pub(crate) unsafe fn stop_and_drain_parts(context: *mut core::ffi::c_void, timer: PTP_TIMER) {
729 // SAFETY: forwarded from this function's own contract.
730 let ctx = unsafe { &*context.cast::<TimerContext>() };
731 ctx.suppress_and_disarm();
732 // Drained with the lock released: a callback blocked on it would
733 // otherwise never finish, and this would never return.
734 crate::trace_call!("WaitForThreadpoolTimerCallbacks", timer, 0, {
735 // SAFETY: `timer` is live, and waiting without cancelling cannot
736 // orphan any storage.
737 unsafe { WaitForThreadpoolTimerCallbacks(timer, FALSE) };
738 });
739 // Settled after the drain: a callback running during it may have asked
740 // to re-arm, and the suppression discards that request, so the timer
741 // really is idle here. A later `set_*` makes it live again.
742 ctx.obligation.record_settled();
743 ctx.release_suppression();
744 }
745
746 /// Give up ownership, returning the raw object and its callback context.
747 ///
748 /// Used only by [`crate::cleanup_group::CleanupGroup`], which takes over
749 /// both: a group member is released by `CloseThreadpoolCleanupGroupMembers`
750 /// and must not close itself, so this suppresses this type's `Drop`.
751 pub(crate) fn into_parts(self) -> (PTP_TIMER, *mut core::ffi::c_void) {
752 let this = std::mem::ManuallyDrop::new(self);
753 (this.timer, this.context.cast())
754 }
755
756 /// Free a context returned by [`ThreadpoolTimer::into_parts`].
757 ///
758 /// # Safety
759 ///
760 /// `context` must come from `into_parts` on this type, its object must
761 /// already have been released, and it must be freed exactly once.
762 pub(crate) unsafe fn drop_context(context: *mut core::ffi::c_void) {
763 // SAFETY: forwarded from this function's own contract.
764 drop(unsafe { Box::from_raw(context.cast::<TimerContext>()) });
765 }
766
767 /// Whether `Drop` would report an undischarged drain obligation right now.
768 ///
769 /// Exists so the obligation's wiring can be asserted without depending on
770 /// the trace, whose filter is fixed before `main` and so cannot be narrowed
771 /// from inside a test.
772 #[cfg(test)]
773 pub(crate) fn obligation_owed(&self) -> bool {
774 // SAFETY: the context outlives every callback and is freed only by Drop.
775 unsafe { &*self.context }.obligation.is_owed()
776 }
777
778 /// Suppress this member's deferred re-arm and disarm it, before a
779 /// [`crate::cleanup_group::CleanupGroup`] bulk-releases its members.
780 ///
781 /// `CloseThreadpoolCleanupGroupMembers` waits for executing callbacks but
782 /// does not stop one from re-arming: a callback still running can request a
783 /// re-arm through [`TimerFiring::rearm_after`] that the trampoline applies
784 /// after it returns, which would re-arm an object the bulk release is
785 /// tearing down and then free its context under a freshly queued callback.
786 /// Raising the suppression before the release closes that door, exactly as
787 /// this type's own `Drop` does; the suppression is never lifted because the
788 /// member is being destroyed.
789 ///
790 /// # Safety
791 ///
792 /// `context` must come from [`into_parts`](Self::into_parts) on this type
793 /// and name a still-live object whose context the caller has not yet freed.
794 pub(crate) unsafe fn prepare_shutdown(context: *mut core::ffi::c_void) {
795 // SAFETY: forwarded; the context outlives the member until the group
796 // frees it, and `suppress_and_disarm` only touches this object.
797 let ctx = unsafe { &*context.cast::<TimerContext>() };
798 ctx.suppress_and_disarm();
799 }
800}
801
802impl Drop for ThreadpoolTimer {
803 fn drop(&mut self) {
804 // Close the door on re-arming before disarming, and do both under the
805 // same lock. Disarming alone is not enough: a callback still running can
806 // have a deferred re-arm that the trampoline applies after it returns,
807 // the drain below could then return with the timer armed, and the close
808 // and context free would race a freshly queued callback.
809 // SAFETY: the context outlives every callback; Drop frees it below,
810 // after the drain.
811 let ctx = unsafe { &*self.context };
812 // Raised and never released: unlike `stop_and_drain`, there is no
813 // afterwards for this object.
814 crate::trace_record!("timer", "drop-begin", self.timer);
815 // Read before the disarm and drain, and emitted before them: the record
816 // marks the start of the blocking interval it reports.
817 // Captured, not re-read later: a firing dispatched during the drain
818 // below settles the obligation, and the context is freed before the
819 // fail-fast, so this must be a value rather than a borrow.
820 let owed = ctx.obligation.is_owed();
821 if owed {
822 crate::trace_record!("timer", crate::obligation::DROP_OBLIGATION_OWED, self.timer);
823 }
824 ctx.suppress_and_disarm();
825 // The lock is released before draining: a callback blocked on it would
826 // otherwise never finish, and this wait would never return.
827 // As in `stop_and_drain`: the disarm above has already discarded any
828 // queued tick, so draining and cancelling are indistinguishable here.
829 self.wait();
830 crate::trace_record!("timer", "drop-drained", self.timer);
831
832 // SAFETY: no callback can be queued or executing, so the object can be
833 // closed and the context freed exactly once.
834 crate::trace_call!("CloseThreadpoolTimer", self.timer, 0, {
835 // SAFETY: no callback remains, so the object can be closed once.
836 unsafe { CloseThreadpoolTimer(self.timer) };
837 });
838 // SAFETY: nothing can reach the context again; free it exactly once.
839 unsafe { drop(Box::from_raw(self.context)) };
840 crate::trace_record!("timer", "drop-closed", self.timer);
841 // Last, after the drain, the close and the context free: a panic
842 // unwinds, so anything after it would be skipped.
843 crate::obligation::fail_fast_if_owed(owed, "ThreadpoolTimer", "stop_and_drain");
844 }
845}
846
847impl std::fmt::Debug for ThreadpoolTimer {
848 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
849 f.debug_struct("ThreadpoolTimer")
850 .field("is_set", &self.is_set())
851 .finish_non_exhaustive()
852 }
853}
854
855#[cfg(test)]
856mod tests;