specfold_core/runner/
pool.rs

1//! Runs a batch of commands: identical ones once, light ones on a worker pool, heavy ones one at a
2//! time in their own lane alongside it. Results come back in the order the jobs were given.
3
4use super::shell::{self, Outcome, Verdict};
5use super::tools;
6use regex::Regex;
7use std::collections::{HashMap, VecDeque};
8use std::path::Path;
9use std::sync::Mutex;
10use std::time::Duration;
11
12#[derive(Debug, Clone, PartialEq, Eq, Hash)]
13pub struct Job {
14    pub cmd: String,
15    pub env: Vec<(String, String)>,
16    pub timeout: Duration,
17}
18
19#[derive(Debug, Clone, Default)]
20pub struct Settings {
21    /// Workers for light commands; 0 means the number of CPUs, at most 8.
22    pub jobs: usize,
23    /// Commands matching any of these run one at a time.
24    pub serial: Vec<Regex>,
25}
26
27impl Settings {
28    fn workers(&self) -> usize {
29        if self.jobs > 0 {
30            return self.jobs;
31        }
32        std::thread::available_parallelism().map(|n| n.get().min(8)).unwrap_or(2)
33    }
34}
35
36fn one(root: &Path, job: &Job) -> Outcome {
37    let missing = tools::missing_programs(root, &job.cmd);
38    if !missing.is_empty() {
39        return Outcome {
40            verdict: Verdict::CannotRun(format!("not installed: {}", missing.join(", "))),
41            tail: String::new(),
42            elapsed: Duration::ZERO,
43        };
44    }
45    shell::run(root, &job.cmd, &job.env, job.timeout)
46}
47
48fn drain(root: &Path, queue: &Mutex<VecDeque<usize>>, unique: &[&Job], done: &Mutex<Vec<Option<Outcome>>>) {
49    loop {
50        let Some(i) = queue.lock().ok().and_then(|mut q| q.pop_front()) else { return };
51        let o = one(root, unique[i]);
52        if let Ok(mut d) = done.lock() {
53            d[i] = Some(o);
54        }
55    }
56}
57
58pub fn run(root: &Path, jobs: &[Job], settings: &Settings) -> Vec<Outcome> {
59    let mut index: HashMap<&Job, usize> = HashMap::new();
60    let mut unique: Vec<&Job> = vec![];
61    let slots: Vec<usize> = jobs
62        .iter()
63        .map(|j| {
64            *index.entry(j).or_insert_with(|| {
65                unique.push(j);
66                unique.len() - 1
67            })
68        })
69        .collect();
70    let heavy = |j: &Job| settings.serial.iter().any(|r| r.is_match(&j.cmd));
71    let light: Mutex<VecDeque<usize>> = Mutex::new((0..unique.len()).filter(|&i| !heavy(unique[i])).collect());
72    let serial: Mutex<VecDeque<usize>> = Mutex::new((0..unique.len()).filter(|&i| heavy(unique[i])).collect());
73    let done: Mutex<Vec<Option<Outcome>>> = Mutex::new(vec![None; unique.len()]);
74    std::thread::scope(|s| {
75        s.spawn(|| drain(root, &serial, &unique, &done));
76        for _ in 0..settings.workers() {
77            s.spawn(|| drain(root, &light, &unique, &done));
78        }
79    });
80    let done = done.into_inner().unwrap_or_default();
81    slots
82        .into_iter()
83        .map(|i| {
84            done.get(i).cloned().flatten().unwrap_or(Outcome {
85                verdict: Verdict::CannotRun("not run".into()),
86                tail: String::new(),
87                elapsed: Duration::ZERO,
88            })
89        })
90        .collect()
91}
92
93#[cfg(all(test, unix))]
94mod tests {
95    use super::*;
96
97    fn job(cmd: &str) -> Job {
98        Job { cmd: cmd.into(), env: vec![], timeout: Duration::from_secs(10) }
99    }
100
101    #[test]
102    fn identical_commands_run_once_and_results_keep_order() {
103        let dir = tempfile::tempdir().unwrap();
104        let cmd = "echo x >> count.txt";
105        let jobs = [job(cmd), job("exit 2"), job(cmd), job("true")];
106        let settings = Settings { jobs: 3, serial: vec![Regex::new("^exit").unwrap()] };
107        let out = run(dir.path(), &jobs, &settings);
108        let verdicts: Vec<&Verdict> = out.iter().map(|o| &o.verdict).collect();
109        assert_eq!(verdicts, [&Verdict::Pass, &Verdict::Fail(Some(2)), &Verdict::Pass, &Verdict::Pass]);
110        assert_eq!(std::fs::read_to_string(dir.path().join("count.txt")).unwrap(), "x\n");
111    }
112
113    #[test]
114    fn a_missing_program_is_not_run() {
115        let dir = tempfile::tempdir().unwrap();
116        let out = run(dir.path(), &[job("no-such-tool-xyz && touch ran")], &Settings::default());
117        assert!(matches!(&out[0].verdict, Verdict::CannotRun(m) if m.contains("no-such-tool-xyz")));
118        assert!(!dir.path().join("ran").exists());
119    }
120}
specfold_core/runner/pool.rs

specfold_core/runner/
pool.rs
