atd_runtime/
runtime.rs

1//! Tokio runtime helpers.
2//!
3//! SP-concurrency-baseline §5.1 — `default_worker_threads()` chooses a sensible
4//! default for ATD reference binaries: `min(available_parallelism, 4)`,
5//! overridable via `ATD_WORKER_THREADS` env. The 4-thread cap prevents a
6//! reference binary launched on a 32-core dev machine from spawning 32 idle
7//! worker threads per process (adopter test suites that spin up several
8//! ref-server instances would otherwise pay 100s of idle threads).
9
10/// Returns the recommended tokio `worker_threads` count for an ATD server
11/// binary. Reads `ATD_WORKER_THREADS` if set; otherwise defaults to
12/// `min(available_parallelism, 4)`, falling back to `2` if the OS query
13/// fails.
14pub fn default_worker_threads() -> usize {
15    if let Some(n) = std::env::var("ATD_WORKER_THREADS")
16        .ok()
17        .and_then(|s| s.parse::<usize>().ok())
18        .filter(|n| *n > 0)
19    {
20        return n;
21    }
22    std::thread::available_parallelism()
23        .map(|n| n.get().min(4))
24        .unwrap_or(2)
25}
26
27#[cfg(test)]
28mod tests {
29    use super::*;
30
31    /// Env mutations across these tests must not race. We don't pull in
32    /// `serial_test`; instead we serialize via a local mutex so the suite
33    /// stays dep-free and the workspace-wide nextest pool isn't poisoned.
34    fn env_lock() -> std::sync::MutexGuard<'static, ()> {
35        static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
36        LOCK.get_or_init(|| std::sync::Mutex::new(()))
37            .lock()
38            .unwrap_or_else(|p| p.into_inner())
39    }
40
41    #[test]
42    fn defaults_to_at_most_four() {
43        let _g = env_lock();
44        // Safety: env mutations serialized within this test module.
45        unsafe { std::env::remove_var("ATD_WORKER_THREADS") };
46        let n = default_worker_threads();
47        assert!((1..=4).contains(&n), "default {n} outside [1,4]");
48    }
49
50    #[test]
51    fn env_override_respected() {
52        let _g = env_lock();
53        unsafe { std::env::set_var("ATD_WORKER_THREADS", "8") };
54        let n = default_worker_threads();
55        unsafe { std::env::remove_var("ATD_WORKER_THREADS") };
56        assert_eq!(n, 8);
57    }
58
59    #[test]
60    fn env_zero_falls_back_to_default() {
61        let _g = env_lock();
62        unsafe { std::env::set_var("ATD_WORKER_THREADS", "0") };
63        let n = default_worker_threads();
64        unsafe { std::env::remove_var("ATD_WORKER_THREADS") };
65        // Zero is nonsense; we fall back to the cap-4 default.
66        assert!((1..=4).contains(&n), "fallback {n} outside [1,4]");
67    }
68
69    #[test]
70    fn env_garbage_falls_back_to_default() {
71        let _g = env_lock();
72        unsafe { std::env::set_var("ATD_WORKER_THREADS", "not-a-number") };
73        let n = default_worker_threads();
74        unsafe { std::env::remove_var("ATD_WORKER_THREADS") };
75        assert!((1..=4).contains(&n), "garbage fallback {n} outside [1,4]");
76    }
77}
atd_runtime/runtime.rs

atd_runtime/
runtime.rs
