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::Mutex;
42use std::sync::atomic::{AtomicIsize, Ordering};
43use std::time::{Duration, SystemTime, UNIX_EPOCH};
44
45use windows_sys::Win32::Foundation::{FALSE, FILETIME, TRUE};
46use windows_sys::Win32::System::Threading::{
47 CloseThreadpoolTimer, CreateThreadpoolTimer, IsThreadpoolTimerSet, PTP_CALLBACK_INSTANCE,
48 PTP_TIMER, SetThreadpoolTimer, WaitForThreadpoolTimerCallbacks,
49};
50
51use crate::callback_env::CallbackEnviron;
52
53/// Conversion constants for Windows `FILETIME`, which counts 100-nanosecond
54/// intervals since 1601-01-01 UTC. Changing any value is a breaking change.
55mod filetime {
56 /// 100-nanosecond intervals per second.
57 pub const TICKS_PER_SECOND: u64 = 10_000_000;
58 /// Nanoseconds per 100-nanosecond interval.
59 pub const NANOS_PER_TICK: u32 = 100;
60 /// Seconds between the `FILETIME` epoch (1601-01-01) and the Unix epoch.
61 pub const SECONDS_1601_TO_1970: u64 = 11_644_473_600;
62}
63
64/// Split a 64-bit tick count into the `FILETIME` field pair.
65fn filetime_from_ticks(ticks: i64) -> FILETIME {
66 let bits = ticks as u64;
67 FILETIME {
68 dwLowDateTime: bits as u32,
69 dwHighDateTime: (bits >> 32) as u32,
70 }
71}
72
73/// Convert a delay into the negative tick count that means "relative" to
74/// `SetThreadpoolTimer`.
75///
76/// Saturates rather than overflowing: a delay beyond the representable range
77/// becomes the furthest representable relative time, far past any practical
78/// process lifetime.
79pub(crate) fn relative_filetime(delay: Duration) -> FILETIME {
80 let ticks = delay
81 .as_secs()
82 .saturating_mul(filetime::TICKS_PER_SECOND)
83 .saturating_add(u64::from(delay.subsec_nanos() / filetime::NANOS_PER_TICK));
84 let ticks = i64::try_from(ticks).unwrap_or(i64::MAX);
85 filetime_from_ticks(-ticks)
86}
87
88/// Convert a wall-clock instant into the positive tick count that means
89/// "absolute" to `SetThreadpoolTimer`.
90///
91/// Instants at or before the Unix epoch clamp to zero, which the pool treats as
92/// immediately due.
93pub(crate) fn absolute_filetime(when: SystemTime) -> FILETIME {
94 let since_unix = when.duration_since(UNIX_EPOCH).unwrap_or(Duration::ZERO);
95 let ticks = since_unix
96 .as_secs()
97 .saturating_add(filetime::SECONDS_1601_TO_1970)
98 .saturating_mul(filetime::TICKS_PER_SECOND)
99 .saturating_add(u64::from(
100 since_unix.subsec_nanos() / filetime::NANOS_PER_TICK,
101 ));
102 filetime_from_ticks(i64::try_from(ticks).unwrap_or(i64::MAX))
103}
104
105/// Saturate a coalescing window to the `u32` millisecond field.
106///
107/// This is the one length in either crate that saturates rather than being
108/// rejected, and deliberately so. A window is a permission -- "you may delay
109/// this firing by up to this much to batch it with others" -- and the pool is
110/// always free to fire earlier, so a saturated window asks for less coalescing
111/// rather than producing a wrong result. A truncated *buffer* silently loses
112/// data, which is why those are rejected instead.
113///
114/// Periods also pass through this, but cannot reach the saturation: they are
115/// validated against
116/// [`MAX_PERIOD`](crate::timer::ThreadpoolPeriodicTimer::MAX_PERIOD) at
117/// construction.
118pub(crate) fn millis_u32(duration: Duration) -> u32 {
119 u32::try_from(duration.as_millis()).unwrap_or(u32::MAX)
120}
121
122/// Arm a raw timer object.
123///
124/// SAFETY: `timer` must be a live `PTP_TIMER`.
125pub(crate) unsafe fn arm_raw(timer: PTP_TIMER, due: FILETIME, period_ms: u32, window_ms: u32) {
126 // SAFETY: forwarded from this function's contract; `due` is read only for
127 // the duration of the call.
128 unsafe { SetThreadpoolTimer(timer, &due, period_ms, window_ms) };
129}
130
131/// Stop a raw timer object.
132///
133/// SAFETY: `timer` must be a live `PTP_TIMER`.
134pub(crate) unsafe fn disarm_raw(timer: PTP_TIMER) {
135 // SAFETY: forwarded; a null due time is the documented way to stop a timer.
136 unsafe { SetThreadpoolTimer(timer, ptr::null(), 0, 0) };
137}
138
139/// Heap-allocated callback state kept alive for the lifetime of the timer.
140///
141/// `timer` is filled in after `CreateThreadpoolTimer` returns, because re-arming
142/// from inside a callback needs the object the callback belongs to.
143pub(crate) struct TimerContext {
144 pub(crate) timer: AtomicIsize,
145 /// How many callers are currently suppressing re-arming: zero means allowed.
146 ///
147 /// Applying a deferred re-arm takes this lock and does nothing while the
148 /// count is non-zero. Deferring the re-arm to after the callback returns --
149 /// which is what makes the delay run from the end of the firing -- moves it
150 /// *past* any disarm performed from outside, so without this a drain could
151 /// complete with a due time installed. For `Drop` that meant closing the
152 /// object and freeing its context with a fresh callback queued against it.
153 ///
154 /// A count rather than a flag because suppression has two users with
155 /// different lifetimes: [`ThreadpoolTimer::stop_and_drain`] raises it and
156 /// lowers it again, while `Drop` raises it permanently. With a flag, a
157 /// `stop_and_drain` finishing would clear a suppression that another
158 /// concurrent one still needed.
159 ///
160 /// The lock is only ever held across the native `SetThreadpoolTimer` call,
161 /// never across a callback drain, which would deadlock a callback that
162 /// happened to be blocked on it.
163 suppress_rearm: Mutex<u32>,
164 /// Records, for tests, whether each deferred re-arm was actually applied.
165 ///
166 /// The suppression this observes happens after the callback returns and
167 /// before the context is freed, so no user-reachable state can witness it;
168 /// this Arc is cloned by the test, which therefore outlives the context.
169 #[cfg(test)]
170 rearm_observer: Mutex<Option<std::sync::Arc<Mutex<Vec<bool>>>>>,
171 callback: Box<dyn Fn(&TimerFiring<'_>) + Send + Sync + 'static>,
172}
173impl TimerContext {
174 /// Lock the suppression count, recovering from a panicking holder.
175 fn suppression(&self) -> std::sync::MutexGuard<'_, u32> {
176 self.suppress_rearm
177 .lock()
178 .unwrap_or_else(|poison| poison.into_inner())
179 }
180
181 /// Start suppressing re-arming, and disarm under the same acquisition.
182 ///
183 /// Doing both under one lock is what makes the pair atomic against a
184 /// callback: a deferred re-arm either lands entirely before this, or is
185 /// suppressed by it. The lock is released before any drain.
186 fn suppress_and_disarm(&self) {
187 let mut suppressed = self.suppression();
188 *suppressed = suppressed.saturating_add(1);
189 let timer = self.timer.load(Ordering::Acquire);
190 if timer != 0 {
191 // SAFETY: `timer` is this object's live PTP_TIMER, published before
192 // any callback could run and valid until Drop closes it.
193 unsafe { disarm_raw(timer) };
194 }
195 }
196
197 /// Stop suppressing re-arming.
198 fn release_suppression(&self) {
199 let mut suppressed = self.suppression();
200 *suppressed = suppressed.saturating_sub(1);
201 }
202}
203
204/// A re-arming a callback asked for, applied once the callback has returned.
205///
206/// Applying it immediately would start the delay from the moment of the call
207/// rather than from the end of the firing, which is both what the API documents
208/// and what keeps firings from overlapping: a callback that re-armed early and
209/// then ran longer than its delay could be entered again concurrently.
210#[derive(Clone, Copy)]
211enum PendingRearm {
212 After(Duration),
213 At(SystemTime),
214}
215
216/// One firing of a [`ThreadpoolTimer`], handed to its callback.
217///
218/// The timer is not armed while the callback runs, so re-arming from here is
219/// what produces repetition whose delay is measured from the *end* of this
220/// callback -- repetition that can never overlap itself.
221pub struct TimerFiring<'ctx> {
222 ctx: &'ctx TimerContext,
223 /// What the callback asked for, applied by the trampoline after it returns.
224 ///
225 /// A `Cell` rather than a lock because the firing is borrowed only by the
226 /// one callback invocation that owns it.
227 pending: Cell<Option<PendingRearm>>,
228}
229
230impl std::fmt::Debug for TimerFiring<'_> {
231 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
232 f.debug_struct("TimerFiring").finish_non_exhaustive()
233 }
234}
235
236impl TimerFiring<'_> {
237 /// Arm the timer again to fire once, `delay` from now.
238 ///
239 /// The arming is applied after this callback returns, so `delay` is measured
240 /// from the **end** of this firing regardless of where in the callback it is
241 /// requested. That is what keeps successive firings strictly sequential: a
242 /// callback that re-armed at its start and then ran longer than `delay`
243 /// would otherwise be entered again while still running.
244 ///
245 /// Calling this more than once in a firing keeps the last request.
246 pub fn rearm_after(&self, delay: Duration) {
247 self.pending.set(Some(PendingRearm::After(delay)));
248 }
249
250 /// Arm the timer again to fire once at the wall-clock instant `when`.
251 ///
252 /// Like [`TimerFiring::rearm_after`], this is applied after the callback
253 /// returns. An instant that has already passed by then fires immediately.
254 pub fn rearm_at(&self, when: SystemTime) {
255 self.pending.set(Some(PendingRearm::At(when)));
256 }
257
258 /// Apply whatever the callback asked for, once it has returned.
259 ///
260 /// Suppressed once teardown has begun, so a request made during the last
261 /// callback cannot re-arm the timer behind `Drop`'s disarm.
262 fn apply_pending(&self) {
263 let applied = self.apply_pending_reporting();
264 #[cfg(test)]
265 if let Some(applied) = applied {
266 let observer = self
267 .ctx
268 .rearm_observer
269 .lock()
270 .unwrap_or_else(|poison| poison.into_inner())
271 .clone();
272 if let Some(observer) = observer {
273 observer
274 .lock()
275 .unwrap_or_else(|poison| poison.into_inner())
276 .push(applied);
277 }
278 }
279 let _ = applied;
280 }
281
282 /// Apply the pending re-arm, reporting whether it was actually installed.
283 ///
284 /// `None` means the callback asked for nothing; `Some(false)` means it asked
285 /// but teardown suppressed the request.
286 fn apply_pending_reporting(&self) -> Option<bool> {
287 let pending = self.pending.get()?;
288 // Taken before arming and held across it, so this either happens before
289 // a suppressing caller raises the count or is suppressed by it -- never
290 // in between.
291 let suppressed = self.ctx.suppression();
292 if *suppressed > 0 {
293 return Some(false);
294 }
295 let timer = self.ctx.timer.load(Ordering::Acquire);
296 debug_assert_ne!(timer, 0, "the timer must be published before callbacks");
297 let due = match pending {
298 PendingRearm::After(delay) => relative_filetime(delay),
299 PendingRearm::At(when) => absolute_filetime(when),
300 };
301 // SAFETY: `timer` is this object's live PTP_TIMER, published before any
302 // callback could run.
303 unsafe { arm_raw(timer, due, 0, 0) };
304 drop(suppressed);
305 Some(true)
306 }
307}
308
309/// Trampoline from the raw `PTP_TIMER_CALLBACK` ABI into the boxed closure.
310///
311/// SAFETY: `context` must point to a live [`TimerContext`] for the entire
312/// duration of every callback invocation, which [`ThreadpoolTimer`]'s `Drop` ordering
313/// guarantees.
314unsafe extern "system" fn timer_trampoline(
315 _instance: PTP_CALLBACK_INSTANCE,
316 context: *mut core::ffi::c_void,
317 _timer: PTP_TIMER,
318) {
319 // SAFETY: context is a valid *mut TimerContext for the full callback duration.
320 let ctx = unsafe { &*(context as *const TimerContext) };
321 let firing = TimerFiring {
322 ctx,
323 pending: Cell::new(None),
324 };
325 // Not contained: see the callback contract in the crate docs.
326 (ctx.callback)(&firing);
327 // Applied only now that the callback has returned, so a requested delay runs
328 // from the end of this firing and the next one cannot overlap it.
329 firing.apply_pending();
330}
331
332/// An owned one-shot thread-pool timer.
333///
334/// Each arming produces exactly one firing. Arm it with
335/// [`ThreadpoolTimer::set_after`] or [`ThreadpoolTimer::set_at`], and stop it
336/// with [`ThreadpoolTimer::disarm`]. Arming again replaces the previous setting
337/// rather than adding to it.
338///
339/// For repetition, either re-arm from inside the callback with
340/// [`TimerFiring::rearm_after`] -- which keeps firings strictly sequential -- or
341/// use [`ThreadpoolPeriodicTimer`] when a fixed cadence matters more than
342/// avoiding overlap.
343///
344/// # When firings can overlap
345///
346/// Re-arming through [`TimerFiring`] never overlaps: the request is applied
347/// after the callback returns, so the next firing cannot begin until this one
348/// has finished. That is the intended way to repeat.
349///
350/// Arming from *outside* the callback is a different matter. Calling
351/// [`ThreadpoolTimer::set_after`] while a callback is running can queue the next
352/// firing before the current one returns, and the two then run concurrently on
353/// different pool threads. The callback is `Fn + Sync`, so this is permitted
354/// rather than unsound -- but it means a callback that assumes it is the only
355/// one running must not be driven that way. Re-arm from the callback, or use
356/// [`ThreadpoolTimer::disarm`] and [`ThreadpoolTimer::wait`] before re-arming
357/// externally.
358///
359/// [`Drop`] disarms before draining callbacks, so the captured closure stays
360/// valid for the full lifetime of every callback execution.
361///
362/// # Examples
363///
364/// Fire once:
365///
366/// ```
367/// use std::sync::mpsc;
368/// use std::time::Duration;
369/// use windows_threadpool_sys::timer::ThreadpoolTimer;
370///
371/// let (tx, rx) = mpsc::channel();
372/// let sender = std::sync::Mutex::new(tx);
373///
374/// let timer = ThreadpoolTimer::new(move |_firing| {
375/// let _ = sender.lock().expect("send").send(());
376/// }, None)?;
377///
378/// timer.set_after(Duration::from_millis(10));
379/// rx.recv().expect("the timer should fire");
380/// # Ok::<(), std::io::Error>(())
381/// ```
382///
383/// Repeat without ever overlapping, by re-arming from inside the callback. The
384/// gap is measured from the end of each firing, so a slow callback delays the
385/// next one instead of racing it:
386///
387/// ```
388/// use std::sync::Arc;
389/// use std::sync::atomic::{AtomicUsize, Ordering};
390/// use std::time::Duration;
391/// use windows_threadpool_sys::timer::ThreadpoolTimer;
392///
393/// let ticks = Arc::new(AtomicUsize::new(0));
394/// let counter = Arc::clone(&ticks);
395///
396/// let timer = ThreadpoolTimer::new(move |firing| {
397/// // Stop after three firings by simply not re-arming.
398/// if counter.fetch_add(1, Ordering::SeqCst) < 2 {
399/// firing.rearm_after(Duration::from_millis(1));
400/// }
401/// }, None)?;
402///
403/// timer.set_after(Duration::from_millis(1));
404/// while ticks.load(Ordering::SeqCst) < 3 {
405/// std::thread::yield_now();
406/// }
407/// timer.disarm();
408/// timer.wait();
409/// assert_eq!(ticks.load(Ordering::SeqCst), 3);
410/// # Ok::<(), std::io::Error>(())
411/// ```
412pub struct ThreadpoolTimer {
413 timer: PTP_TIMER,
414 // Kept alive as a raw pointer until Drop has disarmed and drained.
415 context: *mut TimerContext,
416}
417
418// SAFETY: PTP_TIMER is a cross-thread pool object, and the context holds a
419// callback that is Fn + Send + Sync; the pointer is only read until Drop frees
420// it after all callbacks have finished.
421unsafe impl Send for ThreadpoolTimer {}
422unsafe impl Sync for ThreadpoolTimer {}
423
424impl ThreadpoolTimer {
425 /// Create an idle timer that invokes `callback` each time it expires.
426 ///
427 /// Pass `Some(env)` to select a private pool or callback priority; `None`
428 /// uses the process-default pool with default priority.
429 ///
430 /// The callback runs on a shared, process-managed pool thread. It must
431 /// restore any thread state it changes and must not terminate its thread. It
432 /// must not panic: a panic unwinds to the `extern "system"` trampoline and
433 /// aborts the process.
434 ///
435 /// # Errors
436 ///
437 /// Returns the error from `CreateThreadpoolTimer`.
438 pub fn new<F>(callback: F, env: Option<&mut CallbackEnviron>) -> io::Result<Self>
439 where
440 F: Fn(&TimerFiring<'_>) + Send + Sync + 'static,
441 {
442 let context = Box::into_raw(Box::new(TimerContext {
443 timer: AtomicIsize::new(0),
444 suppress_rearm: Mutex::new(0),
445 #[cfg(test)]
446 rearm_observer: Mutex::new(None),
447 callback: Box::new(callback),
448 }));
449 let env_ptr = env.map_or(ptr::null_mut(), |e| e.as_mut_ptr());
450
451 // SAFETY: context is a valid heap pointer that outlives every callback,
452 // and env_ptr is valid (or null) for the duration of this call.
453 let timer = unsafe {
454 CreateThreadpoolTimer(Some(timer_trampoline), context.cast(), env_ptr.cast_const())
455 };
456
457 if timer == 0 {
458 let error = io::Error::last_os_error();
459 // SAFETY: the pool never saw context; reclaim it immediately.
460 unsafe { drop(Box::from_raw(context)) };
461 return Err(error);
462 }
463
464 // Publish the object before any callback can run. The timer is not armed
465 // yet, so no callback can observe the unpublished value.
466 // SAFETY: context is live and exclusively ours until the first arming.
467 unsafe { (*context).timer.store(timer, Ordering::Release) };
468
469 Ok(Self { timer, context })
470 }
471
472 /// Fire once, `delay` from now.
473 ///
474 /// The delay counts only time the system is awake. A zero delay makes the
475 /// timer due immediately.
476 pub fn set_after(&self, delay: Duration) {
477 // SAFETY: timer is valid for the lifetime of self.
478 unsafe { arm_raw(self.timer, relative_filetime(delay), 0, 0) };
479 }
480
481 /// Fire once at the wall-clock instant `when`.
482 ///
483 /// Unlike a relative due time, an absolute one passes through sleep and
484 /// hibernation: if `when` elapses while the machine is asleep, the timer
485 /// fires promptly on resume. An instant already in the past fires
486 /// immediately.
487 pub fn set_at(&self, when: SystemTime) {
488 // SAFETY: timer is valid for the lifetime of self.
489 unsafe { arm_raw(self.timer, absolute_filetime(when), 0, 0) };
490 }
491
492 /// Fire once after `delay`, allowing the system a coalescing `window`.
493 ///
494 /// `window` is the tolerance the system may add to the due time so it can
495 /// group this timer with other expirations and wake the processor less
496 /// often. A larger window trades timing precision for power.
497 pub fn set_after_with_window(&self, delay: Duration, window: Duration) {
498 // SAFETY: timer is valid for the lifetime of self.
499 unsafe { arm_raw(self.timer, relative_filetime(delay), 0, millis_u32(window)) };
500 }
501
502 /// Stop the timer.
503 ///
504 /// New callbacks stop being queued, but a callback already queued still
505 /// runs; use [`ThreadpoolTimer::cancel_pending`] to drop those as well. Disarming an
506 /// idle timer is a no-op.
507 pub fn disarm(&self) {
508 // SAFETY: timer is valid for the lifetime of self.
509 unsafe { disarm_raw(self.timer) };
510 }
511
512 /// Record, into `observer`, whether each deferred re-arm is actually applied.
513 ///
514 /// The caller keeps its own clone, so the record survives this timer's
515 /// teardown -- which is the only moment a re-arm is suppressed.
516 #[cfg(test)]
517 pub(crate) fn observe_rearms(&self, observer: &std::sync::Arc<Mutex<Vec<bool>>>) {
518 // SAFETY: the context outlives self; Drop frees it after the drain.
519 let ctx = unsafe { &*self.context };
520 *ctx.rearm_observer
521 .lock()
522 .unwrap_or_else(|poison| poison.into_inner()) = Some(std::sync::Arc::clone(observer));
523 }
524 /// Whether the timer currently has a due time.
525 ///
526 /// This reports whether the timer has been armed and not since disarmed. It
527 /// is **not** a prediction that the timer will fire again: expiring does not
528 /// clear the due time, so a fired timer still reports `true`. Only
529 /// [`ThreadpoolTimer::disarm`] makes it `false`.
530 #[must_use]
531 pub fn is_set(&self) -> bool {
532 // SAFETY: timer is valid for the lifetime of self.
533 unsafe { IsThreadpoolTimerSet(self.timer) != 0 }
534 }
535
536 /// Let every queued callback run, and block until none is executing.
537 ///
538 /// This does **not** leave a self-re-arming timer idle. A callback's
539 /// [`TimerFiring::rearm_after`] is applied after the callback returns, so a
540 /// firing that runs during this call installs a fresh due time and the timer
541 /// is armed again when it returns. Use
542 /// [`stop_and_drain`](Self::stop_and_drain) to reach quiescence.
543 pub fn wait(&self) {
544 // SAFETY: timer is valid for the lifetime of self.
545 unsafe { WaitForThreadpoolTimerCallbacks(self.timer, FALSE) };
546 }
547
548 /// Drop callbacks that have not started, then wait for any executing one.
549 ///
550 /// Like [`wait`](Self::wait), this does not by itself leave a self-re-arming
551 /// timer idle: it does not suppress the deferred re-arm of a callback that
552 /// is already running. Use [`stop_and_drain`](Self::stop_and_drain) when the
553 /// timer must actually be quiescent afterwards.
554 pub fn cancel_pending(&self) {
555 // SAFETY: timer is valid for the lifetime of self. A cancelled timer
556 // callback owns no storage, so dropping queued callbacks orphans nothing.
557 unsafe { WaitForThreadpoolTimerCallbacks(self.timer, TRUE) };
558 }
559
560 /// Stop the timer and block until it is idle, leaving it reusable.
561 ///
562 /// This exists because neither [`disarm`](Self::disarm) nor
563 /// [`cancel_pending`](Self::cancel_pending) can stop a self-re-arming timer
564 /// on its own: a callback already running requests its re-arm through
565 /// [`TimerFiring::rearm_after`], and the trampoline applies it *after* the
566 /// callback returns -- which is after any disarm from outside. This
567 /// suppresses that deferred re-arm for the duration of the call, using the
568 /// same mechanism `Drop` uses, and lifts the suppression before returning so
569 /// the timer can be armed again afterwards.
570 ///
571 /// # What this guarantees
572 ///
573 /// On return, provided no other thread arms the timer during the call:
574 ///
575 /// - no callback is queued or executing, and
576 /// - the timer has no due time -- a re-arm requested by a callback that ran
577 /// during the call is discarded rather than deferred.
578 ///
579 /// # What it does not
580 ///
581 /// **A concurrent arm from another thread is not excluded.** `ThreadpoolTimer`
582 /// is `Sync` and [`set_after`](Self::set_after), [`set_at`](Self::set_at) and
583 /// [`set_after_with_window`](Self::set_after_with_window) all take `&self`,
584 /// so they do not pass through the suppression this uses. Nothing in this
585 /// crate orders such a call against this one.
586 ///
587 /// In practice the drain currently cancels a due time installed that way --
588 /// `WaitForThreadpoolTimerCallbacks` with cancellation clears one even when
589 /// no callback is queued, measurably so. That is not a documented contract
590 /// and is not relied upon here: if a caller needs the timer to be provably
591 /// idle, it must ensure nothing else arms it for the duration, by owning it
592 /// exclusively or serializing access to it.
593 ///
594 /// Calling this from inside the timer's own callback would deadlock, because
595 /// it waits for that callback to finish.
596 pub fn stop_and_drain(&self) {
597 // SAFETY: the context outlives every callback and is freed only by Drop,
598 // which cannot run while this borrow of self is alive.
599 let ctx = unsafe { &*self.context };
600 ctx.suppress_and_disarm();
601 // Drained with the lock released: a callback blocked on it would
602 // otherwise never finish, and this would never return.
603 self.cancel_pending();
604 ctx.release_suppression();
605 }
606
607 /// Give up ownership, returning the raw object and its callback context.
608 ///
609 /// Used only by [`crate::cleanup_group::CleanupGroup`], which takes over
610 /// both: a group member is released by `CloseThreadpoolCleanupGroupMembers`
611 /// and must not close itself, so this suppresses this type's `Drop`.
612 pub(crate) fn into_parts(self) -> (PTP_TIMER, *mut core::ffi::c_void) {
613 let this = std::mem::ManuallyDrop::new(self);
614 (this.timer, this.context.cast())
615 }
616
617 /// Free a context returned by [`ThreadpoolTimer::into_parts`].
618 ///
619 /// # Safety
620 ///
621 /// `context` must come from `into_parts` on this type, its object must
622 /// already have been released, and it must be freed exactly once.
623 pub(crate) unsafe fn drop_context(context: *mut core::ffi::c_void) {
624 // SAFETY: forwarded from this function's own contract.
625 drop(unsafe { Box::from_raw(context.cast::<TimerContext>()) });
626 }
627
628 /// Suppress this member's deferred re-arm and disarm it, before a
629 /// [`crate::cleanup_group::CleanupGroup`] bulk-releases its members.
630 ///
631 /// `CloseThreadpoolCleanupGroupMembers` waits for executing callbacks but
632 /// does not stop one from re-arming: a callback still running can request a
633 /// re-arm through [`TimerFiring::rearm_after`] that the trampoline applies
634 /// after it returns, which would re-arm an object the bulk release is
635 /// tearing down and then free its context under a freshly queued callback.
636 /// Raising the suppression before the release closes that door, exactly as
637 /// this type's own `Drop` does; the suppression is never lifted because the
638 /// member is being destroyed.
639 ///
640 /// # Safety
641 ///
642 /// `context` must come from [`into_parts`](Self::into_parts) on this type
643 /// and name a still-live object whose context the caller has not yet freed.
644 pub(crate) unsafe fn prepare_shutdown(context: *mut core::ffi::c_void) {
645 // SAFETY: forwarded; the context outlives the member until the group
646 // frees it, and `suppress_and_disarm` only touches this object.
647 let ctx = unsafe { &*context.cast::<TimerContext>() };
648 ctx.suppress_and_disarm();
649 }
650}
651
652impl Drop for ThreadpoolTimer {
653 fn drop(&mut self) {
654 // Close the door on re-arming before disarming, and do both under the
655 // same lock. Disarming alone is not enough: a callback still running can
656 // have a deferred re-arm that the trampoline applies after it returns,
657 // the drain below could then return with the timer armed, and the close
658 // and context free would race a freshly queued callback.
659 // SAFETY: the context outlives every callback; Drop frees it below,
660 // after the drain.
661 let ctx = unsafe { &*self.context };
662 // Raised and never released: unlike `stop_and_drain`, there is no
663 // afterwards for this object.
664 ctx.suppress_and_disarm();
665 // The lock is released before draining: a callback blocked on it would
666 // otherwise never finish, and this wait would never return.
667 self.cancel_pending();
668
669 // SAFETY: no callback can be queued or executing, so the object can be
670 // closed and the context freed exactly once.
671 unsafe {
672 CloseThreadpoolTimer(self.timer);
673 drop(Box::from_raw(self.context));
674 }
675 }
676}
677
678impl std::fmt::Debug for ThreadpoolTimer {
679 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
680 f.debug_struct("ThreadpoolTimer")
681 .field("is_set", &self.is_set())
682 .finish_non_exhaustive()
683 }
684}
685
686#[cfg(test)]
687mod tests;