traverse_runtime/executor/
thread_pool.rs

1//! Bounded native capability executor.
2//!
3//! Governed by spec `047-thread-pool-executor`.
4
5use std::panic::{AssertUnwindSafe, catch_unwind};
6
7use rayon::{ThreadPool, ThreadPoolBuildError, ThreadPoolBuilder};
8use serde_json::Value;
9
10use super::{ArtifactType, CapabilityExecutor, ExecutorCapability, ExecutorError};
11
12const MIN_CAPACITY: usize = 1;
13const MAX_CAPACITY: usize = 256;
14
15/// Configuration for [`ThreadPoolExecutor`].
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub struct ThreadPoolExecutorConfig {
18    /// Number of worker threads available to native capability execution.
19    pub capacity: usize,
20}
21
22/// Construction-time configuration error.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub enum ConfigError {
25    /// The configured capacity is outside the supported inclusive range.
26    InvalidCapacity {
27        /// Requested capacity.
28        given: usize,
29        /// Minimum supported capacity.
30        min: usize,
31        /// Maximum supported capacity.
32        max: usize,
33    },
34    /// Rayon could not build the worker pool.
35    PoolBuildFailed(String),
36}
37
38impl std::fmt::Display for ConfigError {
39    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40        match self {
41            Self::InvalidCapacity { given, min, max } => {
42                write!(
43                    f,
44                    "invalid thread pool capacity {given}; expected {min}..={max}"
45                )
46            }
47            Self::PoolBuildFailed(msg) => write!(f, "thread pool build failed: {msg}"),
48        }
49    }
50}
51
52impl std::error::Error for ConfigError {}
53
54impl From<ThreadPoolBuildError> for ConfigError {
55    fn from(value: ThreadPoolBuildError) -> Self {
56        Self::PoolBuildFailed(value.to_string())
57    }
58}
59
60/// Dispatches native capability execution onto a bounded worker pool.
61pub struct ThreadPoolExecutor {
62    pool: ThreadPool,
63    inner: Box<dyn CapabilityExecutor>,
64}
65
66impl std::fmt::Debug for ThreadPoolExecutor {
67    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
68        f.debug_struct("ThreadPoolExecutor").finish_non_exhaustive()
69    }
70}
71
72impl ThreadPoolExecutor {
73    /// Create a new bounded thread-pool executor.
74    ///
75    /// # Errors
76    ///
77    /// Returns [`ConfigError`] when capacity is outside `1..=256` or the
78    /// underlying pool cannot be built.
79    pub fn new(
80        config: ThreadPoolExecutorConfig,
81        inner: Box<dyn CapabilityExecutor>,
82    ) -> Result<Self, ConfigError> {
83        if !(MIN_CAPACITY..=MAX_CAPACITY).contains(&config.capacity) {
84            return Err(ConfigError::InvalidCapacity {
85                given: config.capacity,
86                min: MIN_CAPACITY,
87                max: MAX_CAPACITY,
88            });
89        }
90
91        let pool = ThreadPoolBuilder::new()
92            .num_threads(config.capacity)
93            .build()?;
94
95        Ok(Self { pool, inner })
96    }
97}
98
99impl CapabilityExecutor for ThreadPoolExecutor {
100    fn execute(
101        &self,
102        capability: &ExecutorCapability,
103        input: &Value,
104    ) -> Result<Value, ExecutorError> {
105        if capability.artifact_type == ArtifactType::Wasm {
106            return Err(ExecutorError::UnsupportedArtifactType);
107        }
108
109        let capability = capability.clone();
110        let input = input.clone();
111        let result = self
112            .pool
113            .install(|| catch_unwind(AssertUnwindSafe(|| self.inner.execute(&capability, &input))));
114
115        match result {
116            Ok(inner_result) => inner_result,
117            Err(_) => Err(ExecutorError::ExecutionFailed(
118                "capability panicked".to_string(),
119            )),
120        }
121    }
122}
123
124#[cfg(test)]
125mod tests {
126    use std::io;
127    use std::sync::{
128        Arc,
129        atomic::{AtomicUsize, Ordering},
130    };
131    use std::thread;
132    use std::time::Duration;
133
134    use serde_json::json;
135
136    use super::*;
137    use crate::executor::NativeExecutor;
138
139    fn native_capability() -> ExecutorCapability {
140        ExecutorCapability {
141            capability_id: "test.native".to_string(),
142            artifact_type: ArtifactType::Native,
143            wasm_binary_path: None,
144            wasm_checksum: None,
145            host_abi_version: None,
146        }
147    }
148
149    fn wasm_capability() -> ExecutorCapability {
150        ExecutorCapability {
151            artifact_type: ArtifactType::Wasm,
152            ..native_capability()
153        }
154    }
155
156    fn new_executor(
157        capacity: usize,
158        inner: Box<dyn CapabilityExecutor>,
159    ) -> Result<ThreadPoolExecutor, ConfigError> {
160        ThreadPoolExecutor::new(ThreadPoolExecutorConfig { capacity }, inner)
161    }
162
163    fn clone_input(input: &Value) -> Result<Value, String> {
164        if input.get("__thread_pool_test_error").is_some() {
165            return Err("requested test error".to_string());
166        }
167        Ok(input.clone())
168    }
169
170    fn result_debug<T, E: std::fmt::Debug>(result: Result<T, E>) -> Result<T, String> {
171        result.map_err(|err| format!("{err:?}"))
172    }
173
174    fn executor(capacity: usize) -> Result<ThreadPoolExecutor, String> {
175        result_debug(new_executor(
176            capacity,
177            Box::new(NativeExecutor::new(clone_input)),
178        ))
179    }
180
181    fn execute_json(executor: &ThreadPoolExecutor, input: &Value) -> Result<Value, ExecutorError> {
182        executor.execute(&native_capability(), input)
183    }
184
185    #[test]
186    fn clone_input_helper_can_return_error() {
187        let result = clone_input(&json!({ "__thread_pool_test_error": true }));
188
189        assert_eq!(result, Err("requested test error".to_string()));
190    }
191
192    #[test]
193    fn config_capacity_zero_returns_error() {
194        let result = new_executor(0, Box::new(NativeExecutor::new(clone_input)));
195
196        assert_eq!(
197            result.err(),
198            Some(ConfigError::InvalidCapacity {
199                given: 0,
200                min: 1,
201                max: 256
202            })
203        );
204    }
205
206    #[test]
207    fn config_capacity_max_valid() {
208        let result = new_executor(256, Box::new(NativeExecutor::new(clone_input)));
209
210        assert!(result.is_ok());
211    }
212
213    #[test]
214    fn config_capacity_over_max_returns_error() {
215        let result = new_executor(257, Box::new(NativeExecutor::new(clone_input)));
216
217        assert_eq!(
218            result.err(),
219            Some(ConfigError::InvalidCapacity {
220                given: 257,
221                min: 1,
222                max: 256
223            })
224        );
225    }
226
227    #[test]
228    fn config_capacity_one_valid() {
229        let result = new_executor(1, Box::new(NativeExecutor::new(clone_input)));
230
231        assert!(result.is_ok());
232    }
233
234    #[test]
235    fn config_error_display_and_from_cover_failure_shapes() {
236        let invalid = ConfigError::InvalidCapacity {
237            given: 0,
238            min: 1,
239            max: 256,
240        };
241
242        assert_eq!(
243            invalid.to_string(),
244            "invalid thread pool capacity 0; expected 1..=256"
245        );
246
247        let build_error = ThreadPoolBuilder::new()
248            .num_threads(1)
249            .spawn_handler(|_| Err(io::Error::other("spawn denied")))
250            .build()
251            .map_err(ConfigError::from);
252
253        assert!(matches!(build_error, Err(ConfigError::PoolBuildFailed(_))));
254
255        let display = build_error.map_err(|err| err.to_string()).err();
256        assert_eq!(
257            display,
258            Some("thread pool build failed: spawn denied".to_string())
259        );
260    }
261
262    #[test]
263    fn thread_pool_executor_debug_is_non_exhaustive() -> Result<(), String> {
264        let debug = executor(1).map(|executor| format!("{executor:?}"))?;
265
266        assert_eq!(debug, "ThreadPoolExecutor { .. }");
267        Ok(())
268    }
269
270    #[test]
271    fn execute_native_returns_correct_output() -> Result<(), String> {
272        let executor = executor(2)?;
273        let result = execute_json(&executor, &json!({ "value": 42 }));
274
275        assert_eq!(result, Ok(json!({ "value": 42 })));
276        Ok(())
277    }
278
279    #[test]
280    fn execute_native_error_propagates() {
281        let result = new_executor(
282            2,
283            Box::new(NativeExecutor::new(|_| Err("inner failed".to_string()))),
284        )
285        .map(|executor| executor.execute(&native_capability(), &json!({})));
286
287        assert_eq!(
288            result,
289            Ok(Err(ExecutorError::ExecutionFailed(
290                "inner failed".to_string()
291            )))
292        );
293    }
294
295    #[test]
296    fn execute_wasm_artifact_type_returns_unsupported() {
297        let result = new_executor(2, Box::new(PanickingExecutor::new(1)))
298            .map(|executor| executor.execute(&wasm_capability(), &json!({})));
299
300        assert_eq!(result, Ok(Err(ExecutorError::UnsupportedArtifactType)));
301    }
302
303    #[test]
304    fn concurrent_calls_run_in_parallel() {
305        let active_calls = Arc::new(AtomicUsize::new(0));
306        let max_active_calls = Arc::new(AtomicUsize::new(0));
307        let active_for_handler = Arc::clone(&active_calls);
308        let max_for_handler = Arc::clone(&max_active_calls);
309
310        let result = new_executor(
311            2,
312            Box::new(NativeExecutor::new(move |input| {
313                let current = active_for_handler.fetch_add(1, Ordering::SeqCst) + 1;
314                max_for_handler.fetch_max(current, Ordering::SeqCst);
315                thread::sleep(Duration::from_millis(100));
316                active_for_handler.fetch_sub(1, Ordering::SeqCst);
317                Ok(input.clone())
318            })),
319        )
320        .map(|executor| {
321            let executor = Arc::new(executor);
322            let first = {
323                let executor = Arc::clone(&executor);
324                thread::spawn(move || execute_json(&executor, &json!({ "call": 1 })))
325            };
326            let second = {
327                let executor = Arc::clone(&executor);
328                thread::spawn(move || execute_json(&executor, &json!({ "call": 2 })))
329            };
330            (result_debug(first.join()), result_debug(second.join()))
331        });
332
333        let first_result = result
334            .as_ref()
335            .map(|(first, _)| first.as_ref().map_err(String::as_str));
336        let second_result = result
337            .as_ref()
338            .map(|(_, second)| second.as_ref().map_err(String::as_str));
339
340        assert_eq!(first_result, Ok(Ok(&Ok(json!({ "call": 1 })))));
341        assert_eq!(second_result, Ok(Ok(&Ok(json!({ "call": 2 })))));
342        assert!(
343            max_active_calls.load(Ordering::SeqCst) >= 2,
344            "expected two active calls to overlap"
345        );
346    }
347
348    #[test]
349    fn pool_size_one_serialises_calls() {
350        let result = new_executor(
351            1,
352            Box::new(NativeExecutor::new(|input| {
353                thread::sleep(Duration::from_millis(50));
354                Ok(input.clone())
355            })),
356        )
357        .map(|executor| {
358            let executor = Arc::new(executor);
359            let first = {
360                let executor = Arc::clone(&executor);
361                thread::spawn(move || execute_json(&executor, &json!({ "call": 1 })))
362            };
363            let second = {
364                let executor = Arc::clone(&executor);
365                thread::spawn(move || execute_json(&executor, &json!({ "call": 2 })))
366            };
367            (result_debug(first.join()), result_debug(second.join()))
368        });
369        let first_result = result
370            .as_ref()
371            .map(|(first, _)| first.as_ref().map_err(String::as_str));
372        let second_result = result
373            .as_ref()
374            .map(|(_, second)| second.as_ref().map_err(String::as_str));
375
376        assert_eq!(first_result, Ok(Ok(&Ok(json!({ "call": 1 })))));
377        assert_eq!(second_result, Ok(Ok(&Ok(json!({ "call": 2 })))));
378    }
379
380    #[test]
381    fn independent_calls_do_not_share_state() -> Result<(), String> {
382        let executor = Arc::new(executor(4)?);
383        let mut handles = Vec::new();
384
385        for value in 0..10 {
386            let executor = Arc::clone(&executor);
387            handles.push(thread::spawn(move || {
388                execute_json(&executor, &json!({ "value": value }))
389            }));
390        }
391
392        for (value, handle) in handles.into_iter().enumerate() {
393            let result = result_debug(handle.join())?;
394            assert_eq!(result, Ok(json!({ "value": value })));
395        }
396        Ok(())
397    }
398
399    struct PanickingExecutor {
400        remaining_panics: AtomicUsize,
401    }
402
403    impl PanickingExecutor {
404        fn new(remaining_panics: usize) -> Self {
405            Self {
406                remaining_panics: AtomicUsize::new(remaining_panics),
407            }
408        }
409    }
410
411    impl CapabilityExecutor for PanickingExecutor {
412        fn execute(
413            &self,
414            _capability: &ExecutorCapability,
415            input: &Value,
416        ) -> Result<Value, ExecutorError> {
417            let remaining = self.remaining_panics.load(Ordering::SeqCst);
418            if remaining > 0 {
419                self.remaining_panics.fetch_sub(1, Ordering::SeqCst);
420                std::panic::resume_unwind(Box::new("boom"));
421            }
422            Ok(input.clone())
423        }
424    }
425
426    #[test]
427    fn panicking_handler_returns_execution_failed() -> Result<(), String> {
428        let executor = result_debug(new_executor(1, Box::new(PanickingExecutor::new(1))))?;
429        let result = executor.execute(&native_capability(), &json!({}));
430
431        assert_eq!(
432            result,
433            Err(ExecutorError::ExecutionFailed(
434                "capability panicked".to_string()
435            ))
436        );
437        Ok(())
438    }
439
440    #[test]
441    fn pool_usable_after_panic() -> Result<(), String> {
442        let executor = result_debug(new_executor(1, Box::new(PanickingExecutor::new(1))))?;
443        let failed = executor.execute(&native_capability(), &json!({ "first": true }));
444        let recovered = executor.execute(&native_capability(), &json!({ "second": true }));
445
446        assert_eq!(
447            failed,
448            Err(ExecutorError::ExecutionFailed(
449                "capability panicked".to_string()
450            ))
451        );
452        assert_eq!(recovered, Ok(json!({ "second": true })));
453        Ok(())
454    }
455
456    #[test]
457    fn multiple_sequential_panics_do_not_exhaust_pool() -> Result<(), String> {
458        let executor = result_debug(new_executor(2, Box::new(PanickingExecutor::new(5))))?;
459
460        for _ in 0..5 {
461            let result = executor.execute(&native_capability(), &json!({}));
462            assert_eq!(
463                result,
464                Err(ExecutorError::ExecutionFailed(
465                    "capability panicked".to_string()
466                ))
467            );
468        }
469
470        let recovered = executor.execute(&native_capability(), &json!({ "ok": true }));
471
472        assert_eq!(recovered, Ok(json!({ "ok": true })));
473        Ok(())
474    }
475
476    #[test]
477    fn executor_is_send_sync() {
478        fn assert_send_sync<T: Send + Sync>() {}
479
480        assert_send_sync::<ThreadPoolExecutor>();
481    }
482
483    #[test]
484    fn arc_wrapped_executor_callable_from_multiple_threads() -> Result<(), String> {
485        let executor = Arc::new(executor(4)?);
486        let mut handles = Vec::new();
487
488        for value in 0..4 {
489            let executor = Arc::clone(&executor);
490            handles.push(thread::spawn(move || {
491                execute_json(&executor, &json!({ "thread": value }))
492            }));
493        }
494
495        for (value, handle) in handles.into_iter().enumerate() {
496            let result = result_debug(handle.join())?;
497            assert_eq!(result, Ok(json!({ "thread": value })));
498        }
499        Ok(())
500    }
501
502    #[test]
503    fn executor_drops_cleanly_after_use() -> Result<(), String> {
504        let executor = executor(1)?;
505        let result = execute_json(&executor, &json!({ "used": true }));
506
507        assert_eq!(result, Ok(json!({ "used": true })));
508        drop(executor);
509        Ok(())
510    }
511
512    #[test]
513    fn executor_drops_cleanly_with_no_calls() -> Result<(), String> {
514        let executor = executor(1)?;
515
516        drop(executor);
517        Ok(())
518    }
519}
traverse_runtime/executor/thread_pool.rs

traverse_runtime/executor/
thread_pool.rs
