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