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    /// The shell to run commands in; `sh` when unset.
26    pub shell: Option<String>,
27}
28
29impl Settings {
30    fn workers(&self) -> usize {
31        if self.jobs > 0 {
32            return self.jobs;
33        }
34        std::thread::available_parallelism().map(|n| n.get().min(8)).unwrap_or(2)
35    }
36}
37
38fn one(root: &Path, job: &Job, shell_prog: Option<&str>) -> Outcome {
39    let missing = tools::missing_programs(root, &job.cmd);
40    if !missing.is_empty() {
41        return Outcome {
42            verdict: Verdict::CannotRun(format!("not installed: {}", missing.join(", "))),
43            tail: String::new(),
44            elapsed: Duration::ZERO,
45        };
46    }
47    shell::run(root, &job.cmd, &job.env, job.timeout, shell_prog)
48}
49
50fn drain(
51    root: &Path,
52    queue: &Mutex<VecDeque<usize>>,
53    unique: &[&Job],
54    done: &Mutex<Vec<Option<Outcome>>>,
55    shell_prog: Option<&str>,
56) {
57    loop {
58        let Some(i) = queue.lock().ok().and_then(|mut q| q.pop_front()) else { return };
59        let o = one(root, unique[i], shell_prog);
60        if let Ok(mut d) = done.lock() {
61            d[i] = Some(o);
62        }
63    }
64}
65
66pub fn run(root: &Path, jobs: &[Job], settings: &Settings) -> Vec<Outcome> {
67    let mut index: HashMap<&Job, usize> = HashMap::new();
68    let mut unique: Vec<&Job> = vec![];
69    let slots: Vec<usize> = jobs
70        .iter()
71        .map(|j| {
72            *index.entry(j).or_insert_with(|| {
73                unique.push(j);
74                unique.len() - 1
75            })
76        })
77        .collect();
78    let heavy = |j: &Job| settings.serial.iter().any(|r| r.is_match(&j.cmd));
79    let light: Mutex<VecDeque<usize>> = Mutex::new((0..unique.len()).filter(|&i| !heavy(unique[i])).collect());
80    let serial: Mutex<VecDeque<usize>> = Mutex::new((0..unique.len()).filter(|&i| heavy(unique[i])).collect());
81    let done: Mutex<Vec<Option<Outcome>>> = Mutex::new(vec![None; unique.len()]);
82    let shell_prog = settings.shell.as_deref();
83    std::thread::scope(|s| {
84        s.spawn(|| drain(root, &serial, &unique, &done, shell_prog));
85        for _ in 0..settings.workers() {
86            s.spawn(|| drain(root, &light, &unique, &done, shell_prog));
87        }
88    });
89    let done = done.into_inner().unwrap_or_default();
90    slots
91        .into_iter()
92        .map(|i| {
93            done.get(i).cloned().flatten().unwrap_or(Outcome {
94                verdict: Verdict::CannotRun("not run".into()),
95                tail: String::new(),
96                elapsed: Duration::ZERO,
97            })
98        })
99        .collect()
100}
101
102#[cfg(all(test, unix))]
103mod tests {
104    use super::*;
105
106    fn job(cmd: &str) -> Job {
107        Job { cmd: cmd.into(), env: vec![], timeout: Duration::from_secs(10) }
108    }
109
110    #[test]
111    fn identical_commands_run_once_and_results_keep_order() {
112        let dir = tempfile::tempdir().unwrap();
113        let cmd = "echo x >> count.txt";
114        let jobs = [job(cmd), job("exit 2"), job(cmd), job("true")];
115        let settings = Settings { jobs: 3, serial: vec![Regex::new("^exit").unwrap()], shell: None };
116        let out = run(dir.path(), &jobs, &settings);
117        let verdicts: Vec<&Verdict> = out.iter().map(|o| &o.verdict).collect();
118        assert_eq!(verdicts, [&Verdict::Pass, &Verdict::Fail(Some(2)), &Verdict::Pass, &Verdict::Pass]);
119        assert_eq!(std::fs::read_to_string(dir.path().join("count.txt")).unwrap(), "x\n");
120    }
121
122    #[test]
123    fn a_missing_program_is_not_run() {
124        let dir = tempfile::tempdir().unwrap();
125        let out = run(dir.path(), &[job("no-such-tool-xyz && touch ran")], &Settings::default());
126        assert!(matches!(&out[0].verdict, Verdict::CannotRun(m) if m.contains("no-such-tool-xyz")));
127        assert!(!dir.path().join("ran").exists());
128    }
129}
specfold_core/runner/pool.rs

specfold_core/runner/
pool.rs
