phasesmith_execution/
lib.rs

1//! Explicit bounded execution contexts for native `PhaseSmith` kernels.
2//!
3//! This crate owns reusable Rayon pools instead of installing or modifying the
4//! process-global pool. It gives desktop applications, services, Python
5//! adapters, and tests the same deterministic worker-budget contract.
6//! Applications normally use it through
7//! [`phasesmith::execution`](https://docs.rs/phasesmith/latest/phasesmith/).
8//!
9//! # Example
10//!
11//! ```
12//! use phasesmith_execution::ExecutionPolicy;
13//!
14//! let policy = ExecutionPolicy::new(Some(4), 2)?;
15//! let results = policy.context().map_ordered(4, 2, |index| index * index);
16//! assert_eq!(results, [0, 1, 4, 9]);
17//! assert!(policy.resolved_budget() >= 1);
18//! # Ok::<(), Box<dyn std::error::Error>>(())
19//! ```
20//!
21//! `requested_threads = None` selects available logical parallelism. Fixed
22//! requests are capped to what the process reports as available. Work below
23//! `minimum_parallel_tasks` remains serial, and ordered mapping always returns
24//! results in input order.
25
26use std::error::Error;
27use std::fmt::{Display, Formatter};
28use std::sync::Arc;
29
30use rayon::iter::{IntoParallelIterator, ParallelIterator};
31use rayon::{ThreadPool, ThreadPoolBuildError, ThreadPoolBuilder};
32
33/// Reusable, operation-owned worker pool with deterministic ordered mapping.
34///
35/// The context never installs or mutates Rayon's global pool. A one-thread
36/// context executes closures directly, which also prevents nested pools when
37/// Python already schedules independent native batches.
38#[derive(Clone)]
39pub struct ExecutionContext {
40    threads: usize,
41    pool: Option<Arc<ThreadPool>>,
42}
43
44impl ExecutionContext {
45    /// Construct a context with an exact positive worker budget.
46    ///
47    /// # Errors
48    ///
49    /// Returns [`ThreadPoolBuildError`] when Rayon cannot build the bounded
50    /// worker pool.
51    pub fn new(threads: usize) -> Result<Self, ThreadPoolBuildError> {
52        let pool = if threads <= 1 {
53            None
54        } else {
55            Some(Arc::new(
56                ThreadPoolBuilder::new()
57                    .num_threads(threads)
58                    .thread_name(|index| format!("phasesmith-native-{index}"))
59                    .build()?,
60            ))
61        };
62        Ok(Self {
63            threads: threads.max(1),
64            pool,
65        })
66    }
67
68    /// Return a zero-allocation serial context.
69    #[must_use]
70    pub const fn serial() -> Self {
71        Self {
72            threads: 1,
73            pool: None,
74        }
75    }
76
77    /// Return the exact worker budget owned by this context.
78    #[must_use]
79    pub const fn threads(&self) -> usize {
80        self.threads
81    }
82
83    /// Map independent indices while collecting results in increasing order.
84    ///
85    /// Work below `minimum_parallel_items` remains serial. Indexed Rayon
86    /// collection preserves the input order independently of completion order.
87    pub fn map_ordered<R, F>(
88        &self,
89        item_count: usize,
90        minimum_parallel_items: usize,
91        operation: F,
92    ) -> Vec<R>
93    where
94        R: Send,
95        F: Fn(usize) -> R + Send + Sync,
96    {
97        if let Some(pool) = &self.pool
98            && item_count >= minimum_parallel_items
99        {
100            return pool.install(|| (0..item_count).into_par_iter().map(operation).collect());
101        }
102        (0..item_count).map(operation).collect()
103    }
104}
105
106/// Default fixed worker budget used by scripts and applications.
107pub const DEFAULT_EXECUTION_THREADS: usize = 2;
108/// Default number of independent tasks required before parallel scheduling.
109pub const DEFAULT_MINIMUM_PARALLEL_TASKS: usize = 2;
110
111/// Invalid or unconstructable native execution policy.
112#[derive(Debug)]
113pub enum ExecutionPolicyError {
114    /// A fixed worker budget was zero.
115    InvalidThreadCount,
116    /// The parallel scheduling threshold was zero.
117    InvalidMinimumParallelTasks,
118    /// Rayon could not construct the bounded worker pool.
119    ThreadPool(ThreadPoolBuildError),
120}
121
122impl Display for ExecutionPolicyError {
123    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
124        match self {
125            Self::InvalidThreadCount => {
126                formatter.write_str("threads must be automatic or a positive integer")
127            }
128            Self::InvalidMinimumParallelTasks => {
129                formatter.write_str("minimum_parallel_tasks must be a positive integer")
130            }
131            Self::ThreadPool(error) => Display::fmt(error, formatter),
132        }
133    }
134}
135
136impl Error for ExecutionPolicyError {
137    fn source(&self) -> Option<&(dyn Error + 'static)> {
138        match self {
139            Self::ThreadPool(error) => Some(error),
140            Self::InvalidThreadCount | Self::InvalidMinimumParallelTasks => None,
141        }
142    }
143}
144
145/// Immutable worker-budget policy with one persistent bounded execution pool.
146///
147/// `requested_threads = None` selects the available logical CPU count. Fixed
148/// budgets are capped to that count. Cloned execution contexts share the same
149/// Rayon pool, so prepared phases and workflow operations do not rebuild or
150/// nest worker pools.
151#[derive(Clone)]
152pub struct ExecutionPolicy {
153    requested_threads: Option<usize>,
154    minimum_parallel_tasks: usize,
155    context: ExecutionContext,
156}
157
158impl std::fmt::Debug for ExecutionPolicy {
159    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
160        formatter
161            .debug_struct("ExecutionPolicy")
162            .field("requested_threads", &self.requested_threads)
163            .field("minimum_parallel_tasks", &self.minimum_parallel_tasks)
164            .field(
165                "context",
166                &format_args!("{} threads", self.resolved_budget()),
167            )
168            .finish()
169    }
170}
171
172impl PartialEq for ExecutionPolicy {
173    fn eq(&self, other: &Self) -> bool {
174        self.requested_threads == other.requested_threads
175            && self.minimum_parallel_tasks == other.minimum_parallel_tasks
176    }
177}
178
179impl Eq for ExecutionPolicy {}
180
181impl ExecutionPolicy {
182    /// Construct and retain the bounded execution context for this policy.
183    ///
184    /// # Errors
185    ///
186    /// Returns [`ExecutionPolicyError`] for zero budgets/thresholds or when
187    /// Rayon cannot build the resolved worker pool.
188    pub fn new(
189        requested_threads: Option<usize>,
190        minimum_parallel_tasks: usize,
191    ) -> Result<Self, ExecutionPolicyError> {
192        if requested_threads == Some(0) {
193            return Err(ExecutionPolicyError::InvalidThreadCount);
194        }
195        if minimum_parallel_tasks == 0 {
196            return Err(ExecutionPolicyError::InvalidMinimumParallelTasks);
197        }
198        let available_threads = std::thread::available_parallelism().map_or(1, usize::from);
199        let resolved_threads = requested_threads
200            .unwrap_or(available_threads)
201            .min(available_threads)
202            .max(1);
203        let context =
204            ExecutionContext::new(resolved_threads).map_err(ExecutionPolicyError::ThreadPool)?;
205        Ok(Self {
206            requested_threads,
207            minimum_parallel_tasks,
208            context,
209        })
210    }
211
212    /// Construct the bounded two-thread default policy.
213    ///
214    /// # Errors
215    ///
216    /// Returns [`ExecutionPolicyError`] when Rayon cannot build the pool.
217    pub fn bounded_default() -> Result<Self, ExecutionPolicyError> {
218        Self::new(
219            Some(DEFAULT_EXECUTION_THREADS),
220            DEFAULT_MINIMUM_PARALLEL_TASKS,
221        )
222    }
223
224    /// Return the requested fixed worker count, or `None` for automatic mode.
225    #[must_use]
226    pub const fn requested_threads(&self) -> Option<usize> {
227        self.requested_threads
228    }
229
230    /// Return the independent-task threshold for parallel scheduling.
231    #[must_use]
232    pub const fn minimum_parallel_tasks(&self) -> usize {
233        self.minimum_parallel_tasks
234    }
235
236    /// Return the resolved logical-CPU budget retained by this policy.
237    #[must_use]
238    pub const fn resolved_budget(&self) -> usize {
239        self.context.threads()
240    }
241
242    /// Return the worker count for a known number of independent tasks.
243    #[must_use]
244    pub fn worker_count(&self, task_count: usize) -> usize {
245        if task_count < self.minimum_parallel_tasks {
246            return 1;
247        }
248        self.resolved_budget().min(task_count).max(1)
249    }
250
251    /// Borrow the persistent bounded execution context.
252    #[must_use]
253    pub const fn context(&self) -> &ExecutionContext {
254        &self.context
255    }
256}
257
258#[cfg(test)]
259mod tests {
260    use super::*;
261    use std::thread;
262    use std::time::Duration;
263
264    #[test]
265    fn ordered_mapping_is_identical_across_worker_counts() {
266        let serial = ExecutionContext::serial();
267        let parallel = ExecutionContext::new(3).expect("pool");
268        let operation = |index| {
269            thread::sleep(Duration::from_micros(((7 - index) % 4) as u64));
270            index * index
271        };
272        assert_eq!(
273            serial.map_ordered(8, 2, operation),
274            parallel.map_ordered(8, 2, operation)
275        );
276        assert_eq!(parallel.threads(), 3);
277    }
278
279    #[test]
280    fn threshold_keeps_small_work_serial_and_ordered() {
281        let context = ExecutionContext::new(2).expect("pool");
282        assert_eq!(context.map_ordered(3, 4, |index| index + 1), [1, 2, 3]);
283    }
284
285    #[test]
286    fn policy_validates_resolves_and_reuses_its_context() {
287        let available = thread::available_parallelism().map_or(1, usize::from);
288        let policy = ExecutionPolicy::new(Some(available + 3), 3).expect("policy");
289        assert_eq!(policy.requested_threads(), Some(available + 3));
290        assert_eq!(policy.minimum_parallel_tasks(), 3);
291        assert_eq!(policy.resolved_budget(), available);
292        assert_eq!(policy.worker_count(2), 1);
293        assert_eq!(policy.worker_count(3), available.min(3));
294        assert_eq!(policy.context().threads(), available);
295
296        let shared_context = policy.context().clone();
297        assert_eq!(
298            policy.context().map_ordered(4, 2, |index| index * 2),
299            shared_context.map_ordered(4, 2, |index| index * 2)
300        );
301    }
302
303    #[test]
304    fn policy_rejects_zero_configuration_values() {
305        assert!(matches!(
306            ExecutionPolicy::new(Some(0), 2),
307            Err(ExecutionPolicyError::InvalidThreadCount)
308        ));
309        assert!(matches!(
310            ExecutionPolicy::new(Some(1), 0),
311            Err(ExecutionPolicyError::InvalidMinimumParallelTasks)
312        ));
313    }
314}
phasesmith_execution/lib.rs

phasesmith_execution/
lib.rs
