windows_threadpool_sys/pool.rs
1// Copyright (c) 2026 Mike Grier
2//! Owned private thread pools: `CreateThreadpool` / `CloseThreadpool`.
3//!
4//! Callbacks run on the process-default pool unless a [`CallbackEnviron`] names
5//! a private one. A private pool lets an application bound the threads a
6//! subsystem may consume, or isolate long-running callbacks from unrelated work,
7//! without affecting the rest of the process.
8//!
9//! [`CallbackEnviron`]: crate::callback_env::CallbackEnviron
10
11use std::io;
12use std::ptr;
13use std::sync::Mutex;
14
15use windows_sys::Win32::System::Threading::{
16 CloseThreadpool, CreateThreadpool, PTP_POOL, SetThreadpoolThreadMaximum,
17 SetThreadpoolThreadMinimum,
18};
19
20/// The thread limits this wrapper has been told about.
21///
22/// Each field is `None` until the corresponding setter succeeds, because Win32
23/// offers no way to read a pool's current limits back. A limit we were never
24/// given cannot be used to reject its counterpart.
25#[derive(Debug, Default)]
26struct Limits {
27 minimum: Option<u32>,
28 maximum: Option<u32>,
29}
30
31/// An owned private thread pool.
32///
33/// Pass it to [`CallbackEnviron::set_pool`] to run an object's callbacks on this
34/// pool instead of the process-default one. The environment borrows the pool, so
35/// the pool cannot be closed while an environment still names it.
36///
37/// # Ordering and teardown
38///
39/// Creating a callback object from an environment that names this pool **binds**
40/// the object to the pool. `CloseThreadpool` -- which this type's `Drop` calls --
41/// frees the pool immediately only when no object is bound; otherwise it defers
42/// the release until every bound object has been freed. A live object can
43/// therefore never observe a freed pool, whatever order the pool and its objects
44/// are dropped in, so this is not a memory-safety obligation on the caller. (The
45/// `CallbackEnviron` borrow of the pool covers the one case binding does not: a
46/// freshly created pool with no bound object yet is freed at once, so an
47/// environment must not outlive it.)
48///
49/// What the order *does* control is when teardown blocks. Each object's `Drop`
50/// waits for its in-flight callbacks; the pool's deferred release then completes
51/// once the last object is gone. Declare the pool before the objects that use
52/// it, so it is dropped last and that blocking happens where you expect.
53///
54/// # Examples
55///
56/// ```
57/// use windows_threadpool_sys::callback_env::CallbackEnviron;
58/// use windows_threadpool_sys::pool::ThreadpoolPool;
59/// use windows_threadpool_sys::timer::ThreadpoolTimer;
60/// use std::time::Duration;
61///
62/// // Declared first, so it outlives the objects that use it.
63/// let pool = ThreadpoolPool::new()?;
64/// pool.set_min_threads(1)?;
65/// pool.set_max_threads(4)?;
66///
67/// let mut env = CallbackEnviron::new();
68/// env.set_pool(&pool);
69///
70/// let timer = ThreadpoolTimer::new(|_firing| {}, Some(&mut env))?;
71/// timer.set_after(Duration::from_millis(1));
72/// timer.wait();
73/// # Ok::<(), std::io::Error>(())
74/// ```
75///
76/// [`CallbackEnviron::set_pool`]: crate::callback_env::CallbackEnviron::set_pool
77#[derive(Debug)]
78pub struct ThreadpoolPool {
79 pool: PTP_POOL,
80 limits: Mutex<Limits>,
81}
82
83// SAFETY: PTP_POOL is a kernel-managed object usable from any thread; this type
84// only owns the handle and hands it to the pool APIs, which are thread-safe.
85unsafe impl Send for ThreadpoolPool {}
86unsafe impl Sync for ThreadpoolPool {}
87
88impl ThreadpoolPool {
89 /// Create a new private thread pool.
90 ///
91 /// # Errors
92 ///
93 /// Returns the error from `CreateThreadpool`, which fails when the process
94 /// cannot allocate the pool.
95 pub fn new() -> io::Result<Self> {
96 // SAFETY: the reserved parameter must be null; no other input is read.
97 let pool = unsafe { CreateThreadpool(ptr::null()) };
98 if pool == 0 {
99 return Err(io::Error::last_os_error());
100 }
101 Ok(Self {
102 pool,
103 limits: Mutex::new(Limits::default()),
104 })
105 }
106
107 /// Set the maximum number of threads this pool may allocate.
108 ///
109 /// # Conflicting limits
110 ///
111 /// Win32 lets the two limits contradict each other and resolves the conflict
112 /// by *last call wins*, silently and unreportably. A pool given a maximum of
113 /// 2 and then a minimum of 4 was measured running **4** callbacks
114 /// concurrently, and it did not settle back to 2. This wrapper therefore
115 /// tracks the limits it has set and rejects a pair that cannot both hold,
116 /// rather than letting one quietly annul the other.
117 ///
118 /// # The maximum is a steady-state target, not an instantaneous ceiling
119 ///
120 /// Even where the maximum is the effective limit, it bounds the pool once it
121 /// has settled, not every instant. Raising the minimum creates threads
122 /// eagerly, and those surplus threads are not retired the moment a lower
123 /// maximum is applied: with a minimum of 4 then a maximum of 2, a third
124 /// callback was observed running concurrently in roughly 1 trial in 240 when
125 /// many pools were being created at once. Do not rely on the maximum as a
126 /// mutual-exclusion mechanism; use it to bound resource consumption.
127 ///
128 /// # Errors
129 ///
130 /// Returns [`io::ErrorKind::InvalidInput`] if `maximum` is zero. Such a pool
131 /// runs no callbacks at all -- work submitted to it is queued and never
132 /// executed -- and `SetThreadpoolThreadMaximum` returns void, so nothing
133 /// else could report the mistake. Use
134 /// [`CleanupGroup`](crate::cleanup_group::CleanupGroup) or the objects' own
135 /// teardown to stop callbacks, rather than starving the pool that runs them.
136 ///
137 /// Also returns [`io::ErrorKind::InvalidInput`] if `maximum` is below a
138 /// minimum previously set through [`set_min_threads`](Self::set_min_threads).
139 ///
140 /// # Examples
141 ///
142 /// A maximum below an established minimum is refused instead of silently
143 /// overriding it:
144 ///
145 /// ```
146 /// use windows_threadpool_sys::pool::ThreadpoolPool;
147 ///
148 /// let pool = ThreadpoolPool::new()?;
149 /// pool.set_min_threads(4)?;
150 /// assert!(pool.set_max_threads(2).is_err());
151 /// # Ok::<(), std::io::Error>(())
152 /// ```
153 pub fn set_max_threads(&self, maximum: u32) -> io::Result<()> {
154 if maximum == 0 {
155 return Err(io::Error::new(
156 io::ErrorKind::InvalidInput,
157 "a thread pool needs a maximum of at least one thread; a maximum of zero runs no \
158 callbacks at all and the native call cannot report it",
159 ));
160 }
161 let mut limits = self.limits.lock().unwrap_or_else(|e| e.into_inner());
162 if let Some(minimum) = limits.minimum
163 && maximum < minimum
164 {
165 return Err(io::Error::new(
166 io::ErrorKind::InvalidInput,
167 format!(
168 "a maximum of {maximum} is below this pool's minimum of {minimum}; Win32 \
169 would let the minimum win silently, so the conflict is refused instead"
170 ),
171 ));
172 }
173 // SAFETY: pool is valid for the lifetime of self.
174 unsafe { SetThreadpoolThreadMaximum(self.pool, maximum) };
175 limits.maximum = Some(maximum);
176 Ok(())
177 }
178
179 /// Set the minimum number of threads this pool keeps available.
180 ///
181 /// Raising the minimum makes the pool create threads eagerly, which is what
182 /// guarantees forward progress for callbacks that block on one another.
183 ///
184 /// # Errors
185 ///
186 /// Returns the error from `SetThreadpoolThreadMinimum`, which fails when the
187 /// pool cannot create the requested threads.
188 ///
189 /// Returns [`io::ErrorKind::InvalidInput`] if `minimum` exceeds a maximum
190 /// previously set through [`set_max_threads`](Self::set_max_threads). Win32
191 /// would accept it and run up to `minimum` callbacks concurrently, annulling
192 /// the maximum without reporting anything; see
193 /// [`set_max_threads`](Self::set_max_threads) for the measurements.
194 ///
195 /// # Examples
196 ///
197 /// ```
198 /// use windows_threadpool_sys::pool::ThreadpoolPool;
199 ///
200 /// let pool = ThreadpoolPool::new()?;
201 /// pool.set_max_threads(2)?;
202 /// assert!(pool.set_min_threads(4).is_err());
203 /// pool.set_min_threads(2)?;
204 /// # Ok::<(), std::io::Error>(())
205 /// ```
206 pub fn set_min_threads(&self, minimum: u32) -> io::Result<()> {
207 let mut limits = self.limits.lock().unwrap_or_else(|e| e.into_inner());
208 if let Some(maximum) = limits.maximum
209 && minimum > maximum
210 {
211 return Err(io::Error::new(
212 io::ErrorKind::InvalidInput,
213 format!(
214 "a minimum of {minimum} exceeds this pool's maximum of {maximum}; Win32 \
215 would honour the minimum and annul the maximum silently, so the conflict \
216 is refused instead"
217 ),
218 ));
219 }
220 // SAFETY: pool is valid for the lifetime of self.
221 let ok = unsafe { SetThreadpoolThreadMinimum(self.pool, minimum) };
222 if ok == 0 {
223 return Err(io::Error::last_os_error());
224 }
225 limits.minimum = Some(minimum);
226 Ok(())
227 }
228
229 /// The raw pool value, for storing in a callback environment.
230 pub(crate) fn as_raw(&self) -> PTP_POOL {
231 self.pool
232 }
233}
234
235impl Drop for ThreadpoolPool {
236 fn drop(&mut self) {
237 // SAFETY: pool is valid and owned; the OS releases it once its last
238 // member object is released.
239 unsafe { CloseThreadpool(self.pool) };
240 }
241}
242
243#[cfg(test)]
244mod tests;