apm_core/
work.rs

1use anyhow::Result;
2use std::path::Path;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7pub fn run_engine_loop(
8    root: &Path,
9    cancel: Arc<AtomicBool>,
10    interval_secs: u64,
11    max_concurrent: usize,
12    skip_permissions: bool,
13    epic_filter: Option<String>,
14) -> Result<()> {
15    let mut workers: Vec<(String, Option<String>, crate::start::ManagedChild, std::path::PathBuf)> = Vec::new();
16    let mut no_more = false;
17    let mut next_poll = Instant::now();
18
19    loop {
20        if cancel.load(Ordering::Relaxed) {
21            break;
22        }
23
24        let mut reaped = false;
25        workers.retain_mut(|(_, _epic_id, child, _pid_path)| {
26            let done = matches!(child.try_wait(), Ok(Some(_)));
27            if done {
28                reaped = true;
29            }
30            !done
31        });
32
33        if reaped {
34            next_poll = Instant::now();
35            no_more = false;
36        }
37
38        if no_more {
39            let now = Instant::now();
40            if now < next_poll {
41                std::thread::sleep(Duration::from_millis(500));
42                continue;
43            }
44            no_more = false;
45        }
46
47        if !no_more && workers.len() < max_concurrent {
48            let (blocked_epics, default_blocked) = crate::config::Config::load(root)
49                .map(|config| {
50                    let epic_ids: Vec<Option<String>> = workers.iter()
51                        .map(|(_, eid, _, _)| eid.clone())
52                        .collect();
53                    let blocked = config.blocked_epics(&epic_ids);
54                    let def_blocked = config.is_default_branch_blocked(&epic_ids);
55                    (blocked, def_blocked)
56                })
57                .unwrap_or_default();
58            let mut _messages = Vec::new();
59            let mut _warnings = Vec::new();
60            match crate::start::spawn_next_worker(root, true, skip_permissions, epic_filter.as_deref(), &blocked_epics, default_blocked, &mut _messages, &mut _warnings) {
61                Ok(None) => {
62                    next_poll = Instant::now() + Duration::from_secs(interval_secs);
63                    no_more = true;
64                }
65                Ok(Some((id, epic_id, child, pid_path))) => {
66                    workers.push((id, epic_id, child, pid_path));
67                    no_more = false;
68                }
69                Err(_) => {
70                    no_more = true;
71                    std::thread::sleep(Duration::from_secs(30));
72                }
73            }
74        } else {
75            std::thread::sleep(Duration::from_millis(500));
76        }
77    }
78
79    Ok(())
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85
86    #[test]
87    fn cancel_flag_stops_loop_immediately() {
88        let cancel = Arc::new(AtomicBool::new(true));
89        let root = std::path::Path::new("/nonexistent");
90        let result = run_engine_loop(root, cancel, 30, 1, false, None);
91        assert!(result.is_ok());
92    }
93}
apm_core/work.rs

apm_core/
work.rs
