Skip to main content

windows_threadpool_sys/
wait.rs

1// Copyright (c) 2026 Mike Grier
2//! Thread-pool waits: `CreateThreadpoolWait` / `SetThreadpoolWait` /
3//! `WaitForThreadpoolWaitCallbacks` / `CloseThreadpoolWait`.
4//!
5//! A wait object watches one waitable handle and queues its callback when the
6//! handle is signalled or the wait times out. Two SDK contracts shape this API:
7//!
8//! - **The handle must stay valid while a wait is pending.** [`ThreadpoolWait`]
9//!   therefore holds its handle rather than borrowing one -- owning it, or
10//!   sharing it as a `SharedHandle` -- so it cannot be closed underneath a
11//!   pending wait. Use [`ThreadpoolWait::handle`] to inspect it; to signal an
12//!   event the wait holds, build the wait from a `win-sync-sys` [`Event`] and
13//!   keep a clone of it.
14//! - **A wait fires at most once per arming.** The SDK requires the wait to be
15//!   rearmed explicitly for each activation, so the callback receives a
16//!   [`WaitActivation`] carrying [`WaitActivation::rearm`]. A callback that does
17//!   not rearm simply stops watching.
18//!
19//! Mutex handles are not supported by the thread pool and must not be passed to
20//! [`ThreadpoolWait::new`].
21
22use std::io;
23use std::os::windows::io::{AsHandle, AsRawHandle, BorrowedHandle, OwnedHandle};
24use std::ptr;
25use std::sync::atomic::{AtomicIsize, Ordering};
26use std::time::Duration;
27
28use crate::rearm::RearmSuppression;
29
30use win_shared_os_owned_handle::SharedHandle;
31use win_sync_sys::{Event, ResetMode};
32use windows_sys::Win32::Foundation::{FALSE, FILETIME, HANDLE, TRUE, WAIT_TIMEOUT};
33use windows_sys::Win32::System::Threading::{
34    CloseThreadpoolWait, CreateThreadpoolWait, PTP_CALLBACK_INSTANCE, PTP_WAIT, SetThreadpoolWait,
35    WaitForThreadpoolWaitCallbacks,
36};
37use windows_sys::core::BOOL;
38
39use crate::callback_env::CallbackEnviron;
40
41/// Wait results the pool reports. Changing either value is a breaking change.
42mod wait_status {
43    /// `WAIT_OBJECT_0`: the handle was signalled.
44    pub const SIGNALLED: u32 = 0;
45}
46
47/// 100-nanosecond intervals per second, for building a relative `FILETIME`.
48/// Changing this value is a breaking change.
49const FILETIME_TICKS_PER_SECOND: u64 = 10_000_000;
50/// Nanoseconds per 100-nanosecond interval. Changing it is a breaking change.
51const FILETIME_NANOS_PER_TICK: u32 = 100;
52
53/// Build the negative tick count that means "relative timeout".
54fn relative_filetime(timeout: Duration) -> FILETIME {
55    let ticks = timeout
56        .as_secs()
57        .saturating_mul(FILETIME_TICKS_PER_SECOND)
58        .saturating_add(u64::from(timeout.subsec_nanos() / FILETIME_NANOS_PER_TICK));
59    let ticks = i64::try_from(ticks).unwrap_or(i64::MAX);
60    let bits = (-ticks) as u64;
61    FILETIME {
62        dwLowDateTime: bits as u32,
63        dwHighDateTime: (bits >> 32) as u32,
64    }
65}
66
67/// A Win32 routine that closes a wait target.
68///
69/// This is the shape Win32 close routines already have, so one can be passed
70/// directly with no shim: `CloseHandle` and `FindCloseChangeNotification` both
71/// match it. The return value is ignored -- there is nothing a destructor could
72/// usefully do with a close failure.
73pub type WaitCloseFn = unsafe extern "system" fn(HANDLE) -> BOOL;
74
75/// Owns a handle that is closed with a routine other than `CloseHandle`.
76///
77/// The `Drop` lives here rather than on [`WaitTarget`] so that the enum itself
78/// has no destructor and can be taken apart by an ordinary `match`.
79pub(crate) struct CustomClose {
80    raw: HANDLE,
81    close: WaitCloseFn,
82}
83
84impl Drop for CustomClose {
85    fn drop(&mut self) {
86        // Bracketed like a Win32 call because that is what it usually is: the
87        // routine is caller-supplied and the case this exists for,
88        // `FindCloseChangeNotification`, is a kernel close. It runs while the
89        // owning wait is being torn down, so a slow one shows up as a stalled
90        // teardown with no other explanation.
91        crate::trace_call!("WaitCloseFn", self.raw as usize, self.close as usize, {
92            // SAFETY: the handle was vouched for by `assume_waitable_with` and is
93            // still open -- every owner drains the wait before dropping this, so the
94            // pool is no longer watching it. This runs exactly once, because `Drop`
95            // does.
96            unsafe { (self.close)(self.raw) };
97        });
98    }
99}
100
101/// Owns a wait target and closes it with the routine that target requires.
102///
103/// Most handles are closed with `CloseHandle`, which is what a std
104/// [`OwnedHandle`] does, so that stays the default. Some are not: a
105/// `FindFirstChangeNotification` handle must be closed with
106/// `FindCloseChangeNotification`, and closing it the usual way is wrong. The
107/// second variant carries the caller's routine so those targets can be owned by
108/// the pool on the same terms as any other.
109pub(crate) enum WaitTarget {
110    /// The default: closed by [`OwnedHandle`]'s own drop, with `CloseHandle`.
111    Owned(OwnedHandle),
112    /// One holder of a shared handle, closed with `CloseHandle` when the last
113    /// holder drops. The wait's clone is one of them, so the handle stays open
114    /// while the wait holds it.
115    Shared(SharedHandle),
116    /// Closed with the caller-supplied routine.
117    Custom(CustomClose),
118}
119
120// SAFETY: a handle is an opaque OS-owned value, not a pointer into this
121// process, and both variants only ever read it or hand it to a thread-safe Win32
122// call. This restores what the plain `OwnedHandle` field had automatically.
123unsafe impl Send for WaitTarget {}
124unsafe impl Sync for WaitTarget {}
125
126impl WaitTarget {
127    /// The raw handle, for arming the wait and for the callback context.
128    pub(crate) fn raw(&self) -> HANDLE {
129        match self {
130            WaitTarget::Owned(handle) => handle.as_raw_handle(),
131            WaitTarget::Shared(handle) => handle.as_raw_handle(),
132            WaitTarget::Custom(custom) => custom.raw,
133        }
134    }
135
136    /// Borrow the handle, for signalling or inspecting it.
137    pub(crate) fn borrow(&self) -> BorrowedHandle<'_> {
138        match self {
139            WaitTarget::Owned(handle) => handle.as_handle(),
140            WaitTarget::Shared(handle) => handle.as_handle(),
141            // SAFETY: the handle stays open for as long as `self` owns it, and
142            // the returned borrow cannot outlive that.
143            WaitTarget::Custom(custom) => unsafe { BorrowedHandle::borrow_raw(custom.raw) },
144        }
145    }
146}
147
148impl std::fmt::Debug for WaitTarget {
149    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
150        // The close routine is a code pointer with no useful rendering.
151        f.debug_struct("WaitTarget")
152            .field("raw", &self.raw())
153            .field(
154                "close",
155                &match self {
156                    WaitTarget::Owned(_) => "CloseHandle",
157                    WaitTarget::Shared(_) => "CloseHandle, by the last holder",
158                    WaitTarget::Custom(_) => "custom",
159                },
160            )
161            .finish()
162    }
163}
164
165/// A handle the thread pool is able to wait on.
166///
167/// The pool does not support every waitable object: a mutex handle in
168/// particular produces undefined behaviour rather than an error. Requiring this
169/// type instead of a bare [`OwnedHandle`] moves that precondition from prose
170/// into the type system, so a safe caller cannot reach the undefined case.
171///
172/// Construct one safely from a `win-sync-sys` [`Event`] (`From<Event>`) or with
173/// [`WaitableHandle::event`], or vouch for a handle
174/// obtained elsewhere with the narrow [`WaitableHandle::assume_waitable`] seam --
175/// [`WaitableHandle::assume_waitable_shared`] when the handle is shared with
176/// other holders, or [`WaitableHandle::assume_waitable_with`] when the handle
177/// needs a close routine other than `CloseHandle`.
178/// This mirrors `UnassociatedEndpoint` in `windows-overlapped-io-sys`, which
179/// pairs a safe `open` with an `assume_overlapped` escape hatch for the same
180/// reason.
181#[derive(Debug)]
182pub struct WaitableHandle {
183    target: WaitTarget,
184}
185
186impl WaitableHandle {
187    /// Create an event and wrap it as a waitable handle.
188    ///
189    /// An event is always a supported wait target, so this needs no `unsafe`.
190    /// A `manual_reset` event stays signalled until it is reset; an auto-reset
191    /// event returns to unsignalled as soon as one wait is satisfied, which
192    /// makes it the usual choice for handing off work one activation at a time.
193    ///
194    /// The event is a `win-sync-sys` [`Event`], so this crate creates none of
195    /// its own. To be able to set the event once the wait holds it, create the
196    /// [`Event`] directly, keep a clone, and convert the other with
197    /// `From<Event>`: the two share one handle. A wait made by this function
198    /// keeps the only [`Event`], so [`into_handle`](Self::into_handle) returns
199    /// its handle.
200    ///
201    /// # Errors
202    ///
203    /// Returns the error from [`Event::new`].
204    pub fn event(manual_reset: bool, initially_signalled: bool) -> io::Result<Self> {
205        let reset = if manual_reset {
206            ResetMode::Manual
207        } else {
208            ResetMode::Auto
209        };
210        Ok(Event::new(reset, initially_signalled)?.into())
211    }
212
213    /// Wrap a handle whose wait support the caller vouches for.
214    ///
215    /// This is the extensibility seam for wait targets this crate cannot create
216    /// itself -- semaphores, waitable timers, processes, threads, console input,
217    /// change notifications, and so on.
218    ///
219    /// # Safety
220    ///
221    /// The caller guarantees that:
222    ///
223    /// - the handle is a waitable object the thread pool supports, and in
224    ///   particular is **not a mutex**, which the SDK does not support and which
225    ///   yields undefined behaviour rather than an error; and
226    /// - ownership transfers exclusively into the returned value, so nothing
227    ///   else closes the handle while a wait on it is pending.
228    #[must_use]
229    pub unsafe fn assume_waitable(handle: OwnedHandle) -> Self {
230        Self {
231            target: WaitTarget::Owned(handle),
232        }
233    }
234
235    /// Wrap a handle that must be closed with a routine other than
236    /// `CloseHandle`.
237    ///
238    /// Some waitable objects have their own destructor: a
239    /// `FindFirstChangeNotification` handle is closed with
240    /// `FindCloseChangeNotification`, and handing it to
241    /// [`assume_waitable`](Self::assume_waitable) --
242    /// which takes a std [`OwnedHandle`] and therefore closes it with
243    /// `CloseHandle` -- would be wrong. `close` is invoked exactly once, and
244    /// only after the wait has been drained, whether the object is torn down by
245    /// [`ThreadpoolWait`]'s own drop or by a
246    /// [`CleanupGroup`](crate::cleanup_group::CleanupGroup) release.
247    ///
248    /// `close` has the shape Win32 close routines already have, so it can be
249    /// passed directly with no wrapper.
250    ///
251    /// # Safety
252    ///
253    /// The caller guarantees that:
254    ///
255    /// - the handle is a waitable object the thread pool supports, and in
256    ///   particular is **not a mutex**, which the SDK does not support and which
257    ///   yields undefined behaviour rather than an error;
258    /// - ownership transfers exclusively into the returned value, so nothing
259    ///   else closes the handle while a wait on it is pending; and
260    /// - `close` is the correct destructor for this handle and is safe to call
261    ///   once on it after the pool has stopped watching it.
262    #[must_use]
263    pub unsafe fn assume_waitable_with(handle: HANDLE, close: WaitCloseFn) -> Self {
264        Self {
265            target: WaitTarget::Custom(CustomClose { raw: handle, close }),
266        }
267    }
268
269    /// Wrap a shared handle whose wait support the caller vouches for.
270    ///
271    /// The wait holds `handle`, one clone of a [`SharedHandle`], and the
272    /// caller may keep others: to signal the object, or to hand it to another
273    /// waiter. The handle closes only when the last holder drops, so the clone
274    /// the wait holds keeps it open for as long as the wait does, which is the
275    /// guarantee [`assume_waitable`](Self::assume_waitable) gets from exclusive
276    /// ownership.
277    ///
278    /// # Safety
279    ///
280    /// The caller guarantees that the handle is a waitable object the thread
281    /// pool supports, and in particular is **not a mutex**, which the SDK does
282    /// not support and which yields undefined behaviour rather than an error.
283    #[must_use]
284    pub unsafe fn assume_waitable_shared(handle: SharedHandle) -> Self {
285        Self {
286            target: WaitTarget::Shared(handle),
287        }
288    }
289
290    /// Borrow the underlying handle, for signalling or inspecting it.
291    #[must_use]
292    pub fn handle(&self) -> BorrowedHandle<'_> {
293        self.target.borrow()
294    }
295
296    /// Consume the wrapper and recover the owned handle.
297    ///
298    /// # Errors
299    ///
300    /// Returns the wrapper back unchanged when it carries a custom close
301    /// routine (see [`assume_waitable_with`](Self::assume_waitable_with)): an
302    /// [`OwnedHandle`] closes what it holds with `CloseHandle`, which is
303    /// precisely the wrong destructor for such a target, so there is no correct
304    /// value to hand back. The handle is neither closed nor leaked -- ownership
305    /// simply stays where it was.
306    ///
307    /// A shared target (see
308    /// [`assume_waitable_shared`](Self::assume_waitable_shared)) gives its
309    /// handle back only as its last holder, as
310    /// [`SharedHandle::try_into_owned`] does; while another holder remains, the
311    /// wrapper is returned unchanged, still holding its clone.
312    pub fn into_handle(self) -> Result<OwnedHandle, Self> {
313        match self.target {
314            WaitTarget::Owned(handle) => Ok(handle),
315            WaitTarget::Shared(handle) => handle.try_into_owned().map_err(|handle| Self {
316                target: WaitTarget::Shared(handle),
317            }),
318            target @ WaitTarget::Custom(_) => Err(Self { target }),
319        }
320    }
321
322    /// Consume the wrapper and recover the owner, whichever kind it is.
323    pub(crate) fn into_target(self) -> WaitTarget {
324        self.target
325    }
326}
327
328/// An event is always a target the pool supports, so this needs no `unsafe`.
329///
330/// An [`Event`] is guaranteed to be an event rather than a mutex -- adopting a
331/// handle as one is `unsafe` in `win-sync-sys` for exactly this reason -- and
332/// to carry the access a wait needs.
333///
334/// The target is shared: the wait holds the event's handle with every clone of
335/// the [`Event`] the caller keeps, and the handle closes when the last of them
336/// drops.
337///
338/// A wait built this way is a shared target, so
339/// [`WaitableHandle::into_handle`] gives the handle back only when no clone of
340/// the [`Event`] remains, and returns the wrapper unchanged while one does. A
341/// caller that keeps a clone to signal the event therefore does not get the
342/// handle back from `into_handle`; it reaches the event through its clone.
343impl From<Event> for WaitableHandle {
344    fn from(event: Event) -> Self {
345        Self {
346            target: WaitTarget::Shared(event.into()),
347        }
348    }
349}
350
351/// Why a wait callback ran.
352#[derive(Debug, Clone, Copy, PartialEq, Eq)]
353pub enum WaitResult {
354    /// The watched handle became signalled.
355    Signalled,
356    /// The timeout given when the wait was armed elapsed first.
357    TimedOut,
358    /// The pool reported a result this crate does not model. The raw value is
359    /// preserved so a caller can inspect it rather than having it discarded.
360    Other(u32),
361}
362
363impl WaitResult {
364    fn from_raw(value: u32) -> Self {
365        match value {
366            wait_status::SIGNALLED => Self::Signalled,
367            WAIT_TIMEOUT => Self::TimedOut,
368            other => Self::Other(other),
369        }
370    }
371}
372
373/// Heap-allocated callback state kept alive for the lifetime of the wait object.
374///
375/// `wait` is filled in after `CreateThreadpoolWait` returns, because rearming
376/// from inside a callback needs the object the callback belongs to.
377struct WaitContext {
378    wait: AtomicIsize,
379    handle: HANDLE,
380    /// Stops a callback re-arming this wait once a teardown has begun.
381    ///
382    /// Arming takes this lock and does nothing while the count is non-zero, so
383    /// a callback that re-arms cannot start watching again after a disarm from
384    /// outside. Its two users here are [`ThreadpoolWait::stop_and_drain`],
385    /// which raises and lowers it, and `Drop`, which raises it permanently --
386    /// the asymmetry [`RearmSuppression`] is a count for. The lock is only ever
387    /// held across the native `SetThreadpoolWait` call.
388    rearm: RearmSuppression,
389    /// Whether the object is armed with nothing having drained it.
390    ///
391    /// A dispatch settles it because `SetThreadpoolWait` arms for exactly one
392    /// activation: once the callback has entered, the object is no longer
393    /// watching, so there is nothing for `Drop` to wait on and nothing to
394    /// report. A callback that re-arms sets it again, which is why this is
395    /// cleared at trampoline entry rather than on the way out.
396    obligation: crate::obligation::CloseObligation,
397    /// This pool's entry in the self-heal registry.
398    ///
399    /// **The one registration that is read rather than only held.** A wait is
400    /// the only object that reaches the removal primitive a cancellation owes a
401    /// repair for, so `try_cancel_pending`, `recover_repair` and `owe_repair`
402    /// all go through this field; the other four context types hold theirs
403    /// purely for its `Drop`.
404    ///
405    /// On the context rather than on [`ThreadpoolWait`] so it survives
406    /// [`into_parts`](ThreadpoolWait::into_parts) into a cleanup-group member,
407    /// which is how the group reaches it after the object itself is gone.
408    registration: crate::heal::Registration,
409    callback: Box<dyn Fn(&WaitActivation<'_>) + Send + Sync + 'static>,
410}
411
412impl WaitContext {
413    /// Lock the suppression count, recovering from a panicking holder.
414    fn suppression(&self) -> std::sync::MutexGuard<'_, u32> {
415        self.rearm.lock()
416    }
417
418    /// Start suppressing re-arming, and disarm under the same acquisition.
419    ///
420    /// Doing both under one lock is what makes the pair atomic against a
421    /// callback: a re-arm either lands entirely before this, or is suppressed by
422    /// it. The lock is released before any drain.
423    fn suppress_and_disarm(&self) {
424        self.rearm.suppress_and(|suppressed| {
425            let wait = self.wait.load(Ordering::Acquire);
426            crate::trace_record!("wait", "suppress-and-disarm", wait, suppressed);
427            if wait == 0 {
428                return;
429            }
430            // SAFETY: `wait` is this object's live PTP_WAIT, published before any
431            // callback could run and valid until Drop closes it.
432            unsafe { disarm_raw(wait) };
433        });
434    }
435
436    /// Stop suppressing re-arming.
437    fn release_suppression(&self) {
438        self.rearm.release();
439    }
440}
441
442// SAFETY: `handle` is a raw handle owned by the ThreadpoolWait that outlives
443// this context; it is only passed back to SetThreadpoolWait, never closed here.
444unsafe impl Send for WaitContext {}
445unsafe impl Sync for WaitContext {}
446
447/// One activation of a [`ThreadpoolWait`], handed to its callback.
448///
449/// The wait is not armed when the callback runs. Call [`WaitActivation::rearm`]
450/// to watch the handle again; doing nothing leaves the wait idle.
451pub struct WaitActivation<'ctx> {
452    result: WaitResult,
453    ctx: &'ctx WaitContext,
454}
455
456impl std::fmt::Debug for WaitActivation<'_> {
457    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
458        // The context holds the boxed callback and the raw wait object, neither
459        // of which is meaningful to a reader; the result is the whole story.
460        f.debug_struct("WaitActivation")
461            .field("result", &self.result)
462            .finish_non_exhaustive()
463    }
464}
465
466impl WaitActivation<'_> {
467    /// Why this callback ran.
468    #[must_use]
469    pub fn result(&self) -> WaitResult {
470        self.result
471    }
472
473    /// Whether the watched handle was signalled.
474    #[must_use]
475    pub fn is_signalled(&self) -> bool {
476        self.result == WaitResult::Signalled
477    }
478
479    /// Borrow the handle this activation was for.
480    ///
481    /// The wait owns the handle and outlives every callback, so it is open for
482    /// the duration of this borrow. This is what makes the documented way out of
483    /// the overlap hazard on [`rearm`](Self::rearm) reachable: a callback
484    /// watching a manual-reset event can reset it before re-arming, so the next
485    /// activation waits for a fresh signal instead of starting immediately
486    /// alongside this one.
487    #[must_use]
488    pub fn handle(&self) -> BorrowedHandle<'_> {
489        // SAFETY: the wait owns this handle and cannot be dropped while a
490        // callback is running, so it is open for at least this borrow.
491        unsafe { BorrowedHandle::borrow_raw(self.ctx.handle) }
492    }
493
494    /// Arm the wait again, so the next signal or timeout activates it.
495    ///
496    /// `timeout` of `None` waits indefinitely. This is the mechanism the SDK
497    /// requires for repeated waits: an activation consumes the arming, so a
498    /// callback that wants to keep watching must rearm from inside itself.
499    ///
500    /// # This can overlap the callback with itself
501    ///
502    /// Re-arming takes effect immediately, and the pool activates as soon as the
503    /// handle is signalled. If the handle is *still signalled* when this is
504    /// called -- which is the normal state of a manual-reset event -- the next
505    /// activation is queued at once and can begin before the current callback
506    /// returns. Re-arming early in a long callback therefore runs it
507    /// concurrently with itself, repeatedly: a 20ms callback that re-armed at
508    /// its start was measured entering 7529 times in 400ms, 5110 of those
509    /// overlapping an earlier entry.
510    ///
511    /// This is not the guarantee [`TimerFiring::rearm_after`] gives. A one-shot
512    /// timer's re-arm is deferred until the callback returns, precisely so
513    /// firings stay sequential; a wait's re-arm is not, because the SDK requires
514    /// the wait to be re-armed for the handle's *current* signal state to be
515    /// observed.
516    ///
517    /// Either reset the event before re-arming, so the next activation waits
518    /// for a fresh signal -- through a clone of the `win-sync-sys` [`Event`]
519    /// the wait was built from, which the callback keeps:
520    ///
521    /// ```no_run
522    /// # fn example(activation: &windows_threadpool_sys::wait::WaitActivation<'_>, event: &win_sync_sys::Event) {
523    /// // `event` is the callback's clone of the event this wait watches.
524    /// event
525    ///     .reset()
526    ///     .expect("an event created here has the access to reset it");
527    /// activation.rearm(None);
528    /// # }
529    /// ```
530    ///
531    /// or accept the concurrency and make everything the callback touches
532    /// tolerate it. An auto-reset event does not have this problem, because the
533    /// wait consumes the signal.
534    ///
535    /// [`TimerFiring::rearm_after`]: crate::timer::TimerFiring::rearm_after
536    ///
537    /// # Re-arm before the next signal
538    ///
539    /// The ordering rule described on [`ThreadpoolWait::arm`] applies to
540    /// every arming, not just the first -- `SetThreadpoolWait` says the event
541    /// must be re-registered "before signaling it each time". A producer that
542    /// signals in the window after an activation consumed the arming but
543    /// before this call re-establishes it may not get a callback for that
544    /// signal.
545    ///
546    /// Re-arming *before* draining closes that window, at the cost of
547    /// callbacks that find nothing to do; draining first and re-arming after
548    /// leaves it open. A caller who cannot order the two can drain, re-arm,
549    /// then drain again, so that anything which landed in the window is
550    /// picked up by the second pass rather than waited for.
551    ///
552    /// # Teardown
553    ///
554    /// Re-arming after the object has begun tearing down does nothing, so a
555    /// callback racing [`ThreadpoolWait`]'s `Drop` cannot leave the object armed
556    /// behind it.
557    pub fn rearm(&self, timeout: Option<Duration>) {
558        let _ = self.rearm_reporting(timeout);
559    }
560
561    /// [`rearm`](Self::rearm), reporting whether the arming actually happened.
562    ///
563    /// Returns `false` when the request was suppressed because the object is
564    /// tearing down. The public entry point discards this, because a caller
565    /// cannot act on it: by the time it could look, the object is gone. Tests
566    /// use it to observe the suppression directly, which is otherwise only
567    /// visible as the absence of undefined behaviour.
568    pub(crate) fn rearm_reporting(&self, timeout: Option<Duration>) -> bool {
569        // Recorded *before* the lock, not after: the acquisition can block on a
570        // concurrent `suppress_and_disarm`, and a re-arm parked on that mutex is
571        // one of the things a silent interval in the trace could be. Entry and
572        // exit straddling the lock is what tells the two apart.
573        let wait = self.ctx.wait.load(Ordering::Acquire);
574        crate::trace_record!("wait", "rearm-entered", wait);
575        // Taken before arming and held across it, so this either happens before
576        // a suppressing caller raises the count or is suppressed by it -- never
577        // in between.
578        let suppressed = self.ctx.suppression();
579        if *suppressed > 0 {
580            crate::trace_record!("wait", "rearm-suppressed", wait, *suppressed);
581            return false;
582        }
583        debug_assert_ne!(
584            wait, 0,
585            "the wait object must be published before callbacks"
586        );
587        // Recorded before the arming is published, not after: by here the
588        // suppression check has already returned, so the arming is certain to
589        // happen, and a dispatch it produces can settle the obligation on
590        // another thread before this one gets any further. See
591        // `CloseObligation::record_live_before`.
592        self.ctx.obligation.record_live_before(|| {
593            // SAFETY: `wait` is this object's live PTP_WAIT, published before any
594            // callback could run, and `handle` is owned by that object so it is
595            // still open. The timeout, if any, is a live stack value for the call.
596            unsafe { arm_raw(wait, self.ctx.handle, timeout) };
597        });
598        drop(suppressed);
599        crate::trace_record!("wait", "rearm-left", wait, self.ctx.handle as usize);
600        true
601    }
602}
603
604/// Stop a raw wait object.
605///
606/// SAFETY: `wait` must be a live `PTP_WAIT`.
607pub(crate) unsafe fn disarm_raw(wait: PTP_WAIT) {
608    crate::trace_call!("SetThreadpoolWait(disarm)", wait, 0, {
609        // SAFETY: forwarded; a null handle is the documented way to cancel a wait.
610        unsafe { SetThreadpoolWait(wait, ptr::null_mut(), ptr::null()) };
611    });
612    crate::trace_record!("wait", "disarmed", wait);
613}
614
615/// Arm a raw wait object against a borrowed target.
616///
617/// SAFETY: `wait` must be a live `PTP_WAIT` and `target` must stay open until
618/// the wait is disarmed or the object released.
619pub(crate) unsafe fn arm_member(wait: PTP_WAIT, target: &WaitTarget, timeout: Option<Duration>) {
620    // SAFETY: forwarded from this function's own contract.
621    unsafe { arm_raw(wait, target.raw(), timeout) };
622}
623
624/// Arm or disarm a wait object.
625///
626/// SAFETY: `wait` must be a live `PTP_WAIT` and `handle` a live waitable handle
627/// (or null to disarm).
628unsafe fn arm_raw(wait: PTP_WAIT, handle: HANDLE, timeout: Option<Duration>) {
629    crate::trace_call!("SetThreadpoolWait", wait, handle as usize, {
630        match timeout {
631            Some(timeout) => {
632                let filetime = relative_filetime(timeout);
633                // SAFETY: forwarded from this function's contract; `filetime` is
634                // read only for the duration of the call.
635                unsafe { SetThreadpoolWait(wait, handle, &filetime) };
636            }
637            // SAFETY: forwarded; a null timeout means "wait indefinitely".
638            None => unsafe { SetThreadpoolWait(wait, handle, ptr::null()) },
639        }
640    });
641    crate::trace_record!("wait", "armed", wait, handle as usize);
642}
643
644/// Trampoline from the raw `PTP_WAIT_CALLBACK` ABI into the boxed closure.
645///
646/// SAFETY: `context` must point to a live [`WaitContext`] for the entire
647/// duration of every callback invocation, which [`ThreadpoolWait`]'s `Drop`
648/// ordering guarantees.
649unsafe extern "system" fn wait_trampoline(
650    _instance: PTP_CALLBACK_INSTANCE,
651    context: *mut core::ffi::c_void,
652    _wait: PTP_WAIT,
653    wait_result: u32,
654) {
655    crate::trace_record!("wait", "trampoline-entered", _wait, wait_result);
656    // SAFETY: context is a valid *mut WaitContext for the full callback duration.
657    let ctx = unsafe { &*(context as *const WaitContext) };
658    // This activation consumed the arming: the pool is no longer watching, so
659    // nothing is owed unless the callback below arms it again.
660    ctx.obligation.record_settled();
661    // Stamped before the callback, not after: a dispatch that is still running
662    // is evidence the pool is live, and a long callback must not look like
663    // silence to the self-heal.
664    let activation = WaitActivation {
665        result: WaitResult::from_raw(wait_result),
666        ctx,
667    };
668    // Not contained: see the callback contract in the crate docs.
669    (ctx.callback)(&activation);
670    // Paired with the record above. The pair is what makes a silent interval
671    // readable: an entry with no exit says a callback is still inside the
672    // closure, which is a different finding from no entry at all.
673    crate::trace_record!("wait", "trampoline-left", _wait, wait_result);
674}
675
676/// An owned thread-pool wait object bound to one waitable handle.
677///
678/// The object owns the handle, so the handle cannot be closed while a wait is
679/// pending. A newly created wait is idle; arm it with [`ThreadpoolWait::arm`],
680/// and rearm from inside the callback with [`WaitActivation::rearm`].
681///
682/// [`Drop`] disarms before draining callbacks, then closes the object and only
683/// afterwards releases the callback context and the handle.
684///
685/// Unlike [`ThreadpoolTimer`](crate::timer::ThreadpoolTimer), **the callback can
686/// run concurrently with itself**: a wait's re-arm takes effect immediately, so
687/// re-arming while the handle is still signalled queues the next activation
688/// before the current callback returns. See [`WaitActivation::rearm`] for the
689/// measurements and the two ways to avoid it.
690///
691/// # Examples
692///
693/// Watch an event once. The wait holds the handle it is given, so keep a clone
694/// of the [`Event`] -- a second holder of the same handle -- to signal it
695/// through:
696///
697/// ```
698/// use std::sync::mpsc;
699/// use win_sync_sys::{Event, ResetMode};
700/// use windows_threadpool_sys::wait::{ThreadpoolWait, WaitResult};
701///
702/// let event = Event::new(ResetMode::Manual, false)?;
703/// let signal = event.clone();
704///
705/// let (tx, rx) = mpsc::channel();
706/// let sender = std::sync::Mutex::new(tx);
707/// let wait = ThreadpoolWait::new(event.into(), move |activation| {
708///     let _ = sender.lock().expect("send").send(activation.result());
709/// }, None)?;
710///
711/// wait.arm(None);
712/// signal.set()?;
713///
714/// assert_eq!(rx.recv().expect("activation"), WaitResult::Signalled);
715/// # Ok::<(), std::io::Error>(())
716/// ```
717///
718/// Keep watching across activations by rearming from inside the callback, which
719/// is what the SDK requires -- an activation consumes the arming:
720///
721/// ```
722/// # use std::sync::Arc;
723/// # use std::sync::atomic::{AtomicUsize, Ordering};
724/// use win_sync_sys::{Event, ResetMode};
725/// use windows_threadpool_sys::wait::ThreadpoolWait;
726///
727/// let event = Event::new(ResetMode::Auto, false)?;
728/// let signal = event.clone();
729///
730/// let seen = Arc::new(AtomicUsize::new(0));
731/// let counter = Arc::clone(&seen);
732/// let wait = ThreadpoolWait::new(event.into(), move |activation| {
733///     counter.fetch_add(1, Ordering::SeqCst);
734///     activation.rearm(None);
735/// }, None)?;
736///
737/// wait.arm(None);
738/// for _ in 0..3 {
739///     signal.set()?;
740///     std::thread::sleep(std::time::Duration::from_millis(5));
741/// }
742///
743/// // Disarms and drains in one step, discharging the drain this wait owes
744/// // rather than leaving the blocking teardown to `Drop`.
745/// wait.stop_and_drain();
746/// assert!(seen.load(Ordering::SeqCst) >= 1);
747/// # Ok::<(), std::io::Error>(())
748/// ```
749pub struct ThreadpoolWait {
750    wait: PTP_WAIT,
751    target: WaitTarget,
752    // Kept alive as a raw pointer until Drop has disarmed and drained.
753    context: *mut WaitContext,
754}
755
756// SAFETY: PTP_WAIT is a cross-thread pool object, WaitTarget is Send + Sync,
757// and the context is Send + Sync; the pointer is only read until Drop frees it
758// after all callbacks have finished.
759unsafe impl Send for ThreadpoolWait {}
760unsafe impl Sync for ThreadpoolWait {}
761
762impl ThreadpoolWait {
763    /// Create an idle wait watching `handle`.
764    ///
765    /// The object takes ownership of the handle and closes it on drop, which is
766    /// what guarantees the handle outlives any pending wait.
767    ///
768    /// Pass `Some(env)` to select a private pool or callback priority; `None`
769    /// uses the process-default pool with default priority.
770    ///
771    /// The callback runs on a shared, process-managed pool thread, must restore
772    /// any thread state it changes, and must not terminate its thread. It must
773    /// not panic: a panic unwinds to the `extern "system"` trampoline and aborts
774    /// the process.
775    ///
776    /// Taking a [`WaitableHandle`] rather than a bare handle is what keeps this
777    /// constructor safe: the thread pool does not support every waitable object,
778    /// and a mutex handle in particular is undefined rather than an error.
779    ///
780    /// # Errors
781    ///
782    /// Returns the error from `CreateThreadpoolWait`.
783    pub fn new<F>(
784        handle: WaitableHandle,
785        callback: F,
786        env: Option<&mut CallbackEnviron<'_>>,
787    ) -> io::Result<Self>
788    where
789        F: Fn(&WaitActivation<'_>) + Send + Sync + 'static,
790    {
791        let target = handle.into_target();
792        // Read before `env` is consumed below, and registered before the object
793        // exists: the entry must be able to repair this pool from the moment
794        // anything of ours can dispatch on it.
795        let registration = crate::heal::register(crate::heal::key_of(env.as_deref()));
796        let context = Box::into_raw(Box::new(WaitContext {
797            wait: AtomicIsize::new(0),
798            handle: target.raw(),
799            rearm: RearmSuppression::new(),
800            obligation: crate::obligation::CloseObligation::new(),
801            registration,
802            callback: Box::new(callback),
803        }));
804        let env_ptr = env.map_or(ptr::null_mut(), |e| e.as_mut_ptr());
805
806        // SAFETY: context is a valid heap pointer that outlives every callback,
807        // and env_ptr is valid (or null) for the duration of this call.
808        let wait = crate::trace_call!("CreateThreadpoolWait", 0, 0, {
809            // SAFETY: context is a valid heap pointer that outlives every callback,
810            // and env_ptr is valid (or null) for the duration of this call.
811            unsafe {
812                CreateThreadpoolWait(Some(wait_trampoline), context.cast(), env_ptr.cast_const())
813            }
814        });
815
816        if wait == 0 {
817            let error = io::Error::last_os_error();
818            // SAFETY: the pool never saw context; reclaim it immediately.
819            unsafe { drop(Box::from_raw(context)) };
820            return Err(error);
821        }
822
823        // Publish the object before any callback can run. No wait is armed yet,
824        // so no callback can observe the unpublished value.
825        // SAFETY: context is live and exclusively ours until the first arming.
826        unsafe { (*context).wait.store(wait, Ordering::Release) };
827        crate::trace_record!("wait", "created", wait, target.raw() as usize);
828
829        Ok(Self {
830            wait,
831            target,
832            context,
833        })
834    }
835
836    /// Borrow the watched handle, for signalling or inspecting it.
837    #[must_use]
838    pub fn handle(&self) -> BorrowedHandle<'_> {
839        self.target.borrow()
840    }
841
842    /// Arm the wait, so the next signal or timeout runs the callback once.
843    ///
844    /// `timeout` of `None` waits indefinitely. Arming replaces any previous
845    /// arming rather than adding to it, and an activation consumes the arming --
846    /// rearm from inside the callback with [`WaitActivation::rearm`] to keep
847    /// watching.
848    ///
849    /// # Arm before you signal
850    ///
851    /// `SetThreadpoolWait` documents that "you must re-register the event with
852    /// the wait object before signaling it each time to trigger the wait
853    /// callback". Signal a handle that is not currently armed -- including in
854    /// the window between constructing a [`ThreadpoolWait`] and this call --
855    /// and the callback is not guaranteed to run for that signal.
856    ///
857    /// An auto-reset event makes a dropped signal permanent rather than merely
858    /// late, because the signal is consumed and there is nothing left for a
859    /// subsequent arming to observe. If the handle is only ever signalled once,
860    /// as a wakeup for state that is already present, that lost signal is the
861    /// last one the waiter will ever get.
862    ///
863    /// Every example in this module arms first for that reason; so does every
864    /// caller in this workspace. A caller who cannot control the ordering
865    /// should watch a manual-reset event kept in agreement with the state it
866    /// reports, which is level-triggered and so has no signal to lose.
867    pub fn arm(&self, timeout: Option<Duration>) {
868        // SAFETY: the context outlives every callback and is freed only by Drop,
869        // which cannot run while this borrow of self is alive.
870        unsafe { &*self.context }.obligation.record_live_before(|| {
871            // SAFETY: `wait` is valid for the lifetime of self, and the handle is
872            // owned by self so it is still open.
873            unsafe { arm_raw(self.wait, self.target.raw(), timeout) };
874        });
875    }
876
877    /// Stop watching.
878    ///
879    /// New activations stop being queued, but a callback already queued still
880    /// runs; use `try_cancel_pending`, or
881    /// [`try_cancel_pending_no_heal_tracking`](Self::try_cancel_pending_no_heal_tracking)
882    /// in a build without the `self-heal` feature, to drop those as well.
883    pub fn disarm(&self) {
884        crate::trace_call!("SetThreadpoolWait(disarm)", self.wait, 0, {
885            // SAFETY: `wait` is valid for the lifetime of self; a null handle is the
886            // documented way to cancel a pending wait.
887            unsafe { SetThreadpoolWait(self.wait, ptr::null_mut(), ptr::null()) };
888        });
889        crate::trace_record!("wait", "disarmed", self.wait);
890    }
891
892    /// Let every queued callback run, and block until none is executing.
893    ///
894    /// This does **not** leave a self-re-arming wait idle: a callback running
895    /// during this call can [`rearm`](WaitActivation::rearm) before it returns,
896    /// so the object is watching again when this returns. Use
897    /// [`stop_and_drain`](Self::stop_and_drain) to reach quiescence.
898    pub fn wait(&self) {
899        crate::trace_call!("WaitForThreadpoolWaitCallbacks", self.wait, 0, {
900            // SAFETY: `wait` is valid for the lifetime of self.
901            unsafe { WaitForThreadpoolWaitCallbacks(self.wait, FALSE) };
902        });
903    }
904
905    /// Drop callbacks that have not started, then wait for any executing one.
906    ///
907    /// Like [`wait`](Self::wait), this does not by itself leave a self-re-arming
908    /// wait idle: it does not suppress the re-arm of a callback that is already
909    /// running. Use [`stop_and_drain`](Self::stop_and_drain) when the wait must
910    /// actually be quiescent afterwards.
911    ///
912    /// # This brings a process-wide hazard forward; it does not create it
913    ///
914    /// Cancelling asks the kernel to remove a completion packet that may already
915    /// have been delivered to the pool's completion port
916    /// (`IopCancelWaitCompletionPacket` with `RemoveSignaledPacket` set).
917    /// Measured on this workspace's reproducer, a removal that lands a few
918    /// microseconds after the packet was queued, on a port whose factory has no
919    /// threads yet, can **sever the notification from that port to the worker
920    /// factory**.
921    ///
922    /// The scope of that is one pool -- but it is the whole of that pool, not
923    /// this wait and not waits as a kind. Measured during such a stall: already
924    /// armed waits, a freshly armed and signalled wait, a fresh timer, and a
925    /// completed overlapped read all fail to dispatch. A second pool in the same
926    /// process is unaffected. Since the pool normally affected is the **default**
927    /// one, the practical reach is every component in the process that did not
928    /// explicitly create its own -- including code that has nothing to do with
929    /// the caller.
930    ///
931    /// Submitting a work item recovers it, because that reaches the factory by a
932    /// route the broken notification is not on. Nothing else observed does.
933    ///
934    /// **But this method is not the only way to reach that, and on the measured
935    /// evidence it is not even the usual one.** `CloseThreadpoolWait` performs
936    /// the same removal, through the same kernel routine with the same flag,
937    /// whenever it finds a packet still outstanding -- so a teardown that never
938    /// calls this method reaches the hazard anyway. In the reproducer, dropping
939    /// the cancel entirely changed nothing measurable: 22 failures in 20004
940    /// without it against 16 with it.
941    ///
942    /// What this method does is perform the removal *earlier*, which removes any
943    /// chance for the callback to run first.
944    ///
945    /// The fix is therefore not to avoid this method but to leave nothing to
946    /// remove: [`wait`](Self::wait) and [`stop_and_drain`](Self::stop_and_drain)
947    /// let the queued callback run, which clears the association, after which the
948    /// close has nothing to take. Draining cannot reach the removal primitive on
949    /// any path. Prefer them -- but note that doing so is what makes the
950    /// *subsequent close* safe, not merely this call.
951    /// This crate repairs the pool afterwards, which is what makes this safe to
952    /// offer -- with one exception, stated because a guarantee with an unstated
953    /// hole is worse than one that names it.
954    ///
955    /// If this object's pool could not be registered for repair when the object
956    /// was created, which happens only when creating the repair work item
957    /// failed, this call registers it **here** rather than giving up. Only when
958    /// that allocation fails too is the cancellation performed with nothing to
959    /// repair it, and that case records `cancel-untracked` on the `heal` target
960    /// -- naming the pool -- and panics under the `fail-fast` feature. See
961    /// [README-FEATURE-self-heal.md](https://docs.rs/crate/windows-threadpool-sys/latest/source/README-FEATURE-self-heal.md).
962    ///
963    /// # Availability
964    ///
965    /// Requires the `self-heal` feature, which is on by default. Without it this
966    /// method does not exist and a call to it fails to compile, naming
967    /// [`try_cancel_pending_no_heal_tracking`](Self::try_cancel_pending_no_heal_tracking)
968    /// as what to reach for instead. That is the designed behaviour, not an
969    /// oversight: a guarantee you were relying on has been removed, and a
970    /// compile error is the only way you find that out.
971    #[cfg(feature = "self-heal")]
972    pub fn try_cancel_pending(&self) {
973        // SAFETY: the obligation this transfers is discharged here, by marking
974        // the pool so the self-heal repairs it.
975        unsafe { self.try_cancel_pending_no_heal_tracking() };
976        // SAFETY: the context outlives every callback and is freed only by Drop,
977        // which cannot run while this borrow of self is alive.
978        let tracked = unsafe { &*self.context }.registration.owe_repair();
979        // Last in the function, so the panic this may raise cannot skip the
980        // marking above -- the same placement rule the drop fail-fast follows.
981        crate::obligation::fail_fast_if_untracked(tracked, "ThreadpoolWait");
982    }
983
984    /// `try_cancel_pending` without the repair.
985    ///
986    /// Not a link, deliberately: the method it would name does not exist in a
987    /// build with `self-heal` off, and this one does, so the link would dangle
988    /// in exactly the configuration this method exists for.
989    ///
990    /// Always present, including in builds with `self-heal` off, which is the
991    /// point: it is the method that still exists when the gated one does not,
992    /// and its signature says what the caller takes on.
993    ///
994    /// # Safety
995    ///
996    /// The caller must ensure the pool is repaired. Submitting any work item to
997    /// it does so; so does knowing the pool is kept live by something else.
998    /// Leaving it unrepaired can stop the pool dispatching -- which, for the
999    /// process-default pool, reaches every component in the process, including
1000    /// code with no connection to this call.
1001    ///
1002    /// This is not a memory-safety obligation, and the keyword is not claiming
1003    /// one. It is here because the obligation is statable and dischargeable by
1004    /// the caller, which is what `unsafe` marks; see
1005    /// [the self-heal decision](https://docs.rs/crate/windows-threadpool-sys/latest/source/DESIGN-NOTES.md).
1006    pub unsafe fn try_cancel_pending_no_heal_tracking(&self) {
1007        crate::trace_call!("WaitForThreadpoolWaitCallbacks(cancel)", self.wait, 1, {
1008            // SAFETY: `wait` is valid for the lifetime of self. A cancelled wait
1009            // callback owns no storage, so dropping queued callbacks orphans nothing.
1010            unsafe { WaitForThreadpoolWaitCallbacks(self.wait, TRUE) };
1011        });
1012    }
1013
1014    /// Take a repair claim on this wait's pool while the object is still live.
1015    ///
1016    /// For [`crate::cleanup_group::CleanupGroup`], which must do this *before*
1017    /// its native release: a member whose pool was never registered holds
1018    /// nothing that keeps that pool alive, and the release frees the last bound
1019    /// object. The claim this returns owns a repair work object bound to the
1020    /// pool, so holding it across the release keeps the pool alive for the
1021    /// marking pass that follows.
1022    ///
1023    /// # Safety
1024    ///
1025    /// `context` must come from [`into_parts`](Self::into_parts) on this type
1026    /// and name a still-live object whose context the caller has not yet freed.
1027    pub(crate) unsafe fn recover_repair(
1028        context: *mut core::ffi::c_void,
1029    ) -> Option<crate::heal::Registration> {
1030        // SAFETY: forwarded; the context outlives the member until the group
1031        // frees it, and this only touches that object's registration.
1032        let ctx = unsafe { &*context.cast::<WaitContext>() };
1033        Some(ctx.registration.reclaim())
1034    }
1035
1036    /// Mark this wait's pool as owing a repair, retrying the registration.
1037    ///
1038    /// Gated because its only caller is `WaitMember::try_cancel_pending`, which
1039    /// the feature gates too. It was reachable in every configuration until
1040    /// `M-T11.4` removed the `OwnedResource` hook that named it: the group's
1041    /// marking pass now goes through the claims, which is what the retry here
1042    /// must not be reached from.
1043    ///
1044    /// Reports whether the pool ended up tracked. A caller that accumulates
1045    /// this across several members must act on it only after every context is
1046    /// freed -- never mid-loop, where an unwind would skip those frees.
1047    ///
1048    /// # Safety
1049    ///
1050    /// `context` must come from [`into_parts`](Self::into_parts) on this type
1051    /// and name a still-live object whose context the caller has not yet freed.
1052    #[cfg(feature = "self-heal")]
1053    #[must_use]
1054    pub(crate) unsafe fn owe_repair(context: *mut core::ffi::c_void) -> bool {
1055        // SAFETY: forwarded; the context outlives the member until the group
1056        // frees it, and this only touches that object's registration.
1057        let ctx = unsafe { &*context.cast::<WaitContext>() };
1058        ctx.registration.owe_repair()
1059    }
1060
1061    /// Stop watching and block until the wait is idle, leaving it reusable.
1062    ///
1063    /// This exists because neither [`disarm`](Self::disarm) nor
1064    /// `try_cancel_pending` can stop a self-re-arming wait on
1065    /// its own: a callback already running can call [`WaitActivation::rearm`]
1066    /// after a disarm from outside has taken effect. This suppresses re-arming
1067    /// for the duration of the call, using the same mechanism `Drop` uses, and
1068    /// lifts the suppression before returning so the wait can be armed again.
1069    ///
1070    /// # What this guarantees
1071    ///
1072    /// On return, provided no other thread arms the wait during the call:
1073    ///
1074    /// - no callback is queued or executing, and
1075    /// - the object is not watching -- a re-arm requested by a callback that ran
1076    ///   during the call is discarded rather than deferred.
1077    ///
1078    /// **A queued callback runs; it is not discarded.** That is the difference
1079    /// between this and `try_cancel_pending`, and it is why
1080    /// this can block for as long as the callback takes. See
1081    /// [the teardown-drains decision](../../../DESIGN-NOTES.md#teardown-drains)
1082    /// for why finishing the work is preferred to abandoning it.
1083    ///
1084    /// # What it does not
1085    ///
1086    /// **A concurrent [`arm`](Self::arm) from another thread is not excluded.**
1087    /// `ThreadpoolWait` is `Sync` and `arm` takes `&self`, so it does not pass
1088    /// through the suppression this uses, and nothing in this crate orders such
1089    /// a call against this one. A caller needing the wait to be provably idle
1090    /// must ensure nothing else arms it for the duration, by owning it
1091    /// exclusively or serializing access to it.
1092    ///
1093    /// Calling this from inside the wait's own callback would deadlock, because
1094    /// it waits for that callback to finish.
1095    pub fn stop_and_drain(&self) {
1096        // SAFETY: the context outlives every callback and is freed only by Drop,
1097        // which cannot run while this borrow of self is alive.
1098        unsafe { Self::stop_and_drain_parts(self.context.cast(), self.wait) };
1099    }
1100
1101    /// [`stop_and_drain`](Self::stop_and_drain) against a detached context.
1102    ///
1103    /// One body rather than two, so this type and the cleanup-group member that
1104    /// wraps the same object cannot drift apart: the suppression discipline here
1105    /// is what makes the drain mean anything, and a second copy of it is a
1106    /// second place for a later change to reach one and miss the other.
1107    ///
1108    /// # Safety
1109    ///
1110    /// `context` must come from [`into_parts`](Self::into_parts) on this type
1111    /// and name a still-live object whose context the caller has not yet freed,
1112    /// and `wait` must be that object.
1113    pub(crate) unsafe fn stop_and_drain_parts(context: *mut core::ffi::c_void, wait: PTP_WAIT) {
1114        // SAFETY: forwarded from this function's own contract.
1115        let ctx = unsafe { &*context.cast::<WaitContext>() };
1116        ctx.suppress_and_disarm();
1117        // Drained with the lock released: a callback blocked on it would
1118        // otherwise never finish, and this would never return.
1119        crate::trace_call!("WaitForThreadpoolWaitCallbacks", wait, 0, {
1120            // SAFETY: `wait` is live, and waiting without cancelling cannot
1121            // orphan any storage.
1122            unsafe { WaitForThreadpoolWaitCallbacks(wait, FALSE) };
1123        });
1124        // Settled after the drain, not before: a callback running during it may
1125        // have asked to re-arm, and the suppression discards that request, so
1126        // the object really is idle here. Releasing the suppression below lets a
1127        // later `arm` make it live again, which sets this afresh.
1128        ctx.obligation.record_settled();
1129        ctx.release_suppression();
1130    }
1131
1132    /// Give up ownership, returning the raw object, its callback context, and
1133    /// the watched target.
1134    ///
1135    /// Used only by [`crate::cleanup_group::CleanupGroup`], which takes over all
1136    /// three. The target must go with them: the pool may still be watching it
1137    /// until the group releases the member, so it cannot be closed when the
1138    /// borrowing member goes out of scope.
1139    pub(crate) fn into_parts(self) -> (PTP_WAIT, *mut core::ffi::c_void, WaitTarget) {
1140        let this = std::mem::ManuallyDrop::new(self);
1141        // SAFETY: `this` is never dropped, so moving the target out cannot be
1142        // observed by a later drop of the original value.
1143        let target = unsafe { ptr::read(&this.target) };
1144        (this.wait, this.context.cast(), target)
1145    }
1146
1147    /// Free a context returned by [`ThreadpoolWait::into_parts`].
1148    ///
1149    /// # Safety
1150    ///
1151    /// `context` must come from `into_parts` on this type, its object must
1152    /// already have been released, and it must be freed exactly once.
1153    pub(crate) unsafe fn drop_context(context: *mut core::ffi::c_void) {
1154        // SAFETY: forwarded from this function's own contract.
1155        drop(unsafe { Box::from_raw(context.cast::<WaitContext>()) });
1156    }
1157
1158    /// Whether `Drop` would report an undischarged drain obligation right now.
1159    ///
1160    /// Exists so the obligation's wiring can be asserted without depending on
1161    /// the trace, whose filter is fixed before `main` and so cannot be narrowed
1162    /// from inside a test.
1163    #[cfg(test)]
1164    pub(crate) fn obligation_owed(&self) -> bool {
1165        // SAFETY: the context outlives every callback and is freed only by Drop.
1166        unsafe { &*self.context }.obligation.is_owed()
1167    }
1168
1169    /// Suppress this member's re-arm and disarm it, before a
1170    /// [`crate::cleanup_group::CleanupGroup`] bulk-releases its members.
1171    ///
1172    /// `CloseThreadpoolCleanupGroupMembers` waits for executing callbacks but
1173    /// does not stop one from re-arming: a callback still running can call
1174    /// [`WaitActivation::rearm`] after the bulk release has begun, which would
1175    /// re-arm an object being torn down and then free its context under a freshly
1176    /// queued callback. Raising the suppression before the release closes that
1177    /// door, exactly as this type's own `Drop` does; the suppression is never
1178    /// lifted because the member is being destroyed.
1179    ///
1180    /// # Safety
1181    ///
1182    /// `context` must come from [`into_parts`](Self::into_parts) on this type
1183    /// and name a still-live object whose context the caller has not yet freed.
1184    pub(crate) unsafe fn prepare_shutdown(context: *mut core::ffi::c_void) {
1185        // SAFETY: forwarded; the context outlives the member until the group
1186        // frees it, and `suppress_and_disarm` only touches this object.
1187        let ctx = unsafe { &*context.cast::<WaitContext>() };
1188        ctx.suppress_and_disarm();
1189    }
1190}
1191
1192impl Drop for ThreadpoolWait {
1193    fn drop(&mut self) {
1194        // Close the door on re-arming before disarming, and do both under the
1195        // same lock. Disarming alone is not enough: a callback already running
1196        // could re-arm afterwards, the drain below could then return with the
1197        // object armed, and the close and context free would race a freshly
1198        // queued callback.
1199        // SAFETY: the context outlives every callback; Drop frees it below,
1200        // after the drain.
1201        let ctx = unsafe { &*self.context };
1202        // Raised and never released: unlike `stop_and_drain`, there is no
1203        // afterwards for this object.
1204        crate::trace_record!("wait", "drop-begin", self.wait, self.target.raw() as usize);
1205        // Read before the disarm and drain, and emitted before them: the record
1206        // marks the start of the blocking interval it reports, so a reader sees
1207        // what the following gap is for rather than learning it afterwards.
1208        // Captured, not re-read later: a callback dispatched during the drain
1209        // below settles the obligation, so asking afterwards would find nothing
1210        // owed on exactly the objects that owed something. The context is also
1211        // freed before the fail-fast, so this must be a value rather than a
1212        // borrow.
1213        let owed = ctx.obligation.is_owed();
1214        if owed {
1215            crate::trace_record!("wait", crate::obligation::DROP_OBLIGATION_OWED, self.wait);
1216        }
1217        ctx.suppress_and_disarm();
1218        // The lock is released before draining: a callback blocked on it would
1219        // otherwise never finish, and this wait would never return.
1220        self.wait();
1221        crate::trace_record!("wait", "drop-drained", self.wait);
1222
1223        // SAFETY: no callback can be queued or executing, so the object can be
1224        // closed and the context freed exactly once. `target` is dropped after
1225        // this, when its field is dropped, so the handle outlives the wait
1226        // object and its close routine -- `CloseHandle` or a custom one -- runs
1227        // only once the pool has stopped watching it.
1228        crate::trace_call!("CloseThreadpoolWait", self.wait, 0, {
1229            unsafe { CloseThreadpoolWait(self.wait) };
1230        });
1231        // SAFETY: as above; the pool can no longer reach the context.
1232        unsafe {
1233            drop(Box::from_raw(self.context));
1234        }
1235        // The last record this object emits, and it carries the target handle
1236        // because nothing after it can: the close of that handle is the field
1237        // drop that runs the instant this body returns, with no code of ours in
1238        // between. So the interval from `drop-drained` to the handle's close
1239        // holds exactly `CloseThreadpoolWait`, the context free, and that field
1240        // drop -- a reader who sees a gap there is looking at one of those three
1241        // and not at something unrecorded.
1242        crate::trace_record!("wait", "drop-closed", self.wait, self.target.raw() as usize);
1243        // Last, after the drain, the close and the context free. A panic
1244        // unwinds, so anything after it would be skipped; the target handle
1245        // still closes, because field drops run even when `Drop::drop` unwinds.
1246        crate::obligation::fail_fast_if_owed(owed, "ThreadpoolWait", "stop_and_drain");
1247    }
1248}
1249
1250#[cfg(test)]
1251mod tests;