captain-workflow-manager 0.1.0

Run and manage jobs that depend on each other on a variety of backends.s
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

use crate::ExpandedDescr;

use super::*;
pub struct ThreadLocalExecutorBuilder {
    pub n_workers: u16,
}

impl ExecutorBuilder for ThreadLocalExecutorBuilder {
    type Executor = ThreadLocalExecutor;

    fn init<J: JobUnit>(
        self,
    ) -> (
        Self::Executor,
        Sender<ToRun<J>>,
        Sender<ExecutorResult>,
        Receiver<ExecutorResult>,
    ) {
        let to_run_chan = crossbeam_channel::bounded::<ToRun<J>>(2 * (self.n_workers as usize));
        let done_chan = crossbeam_channel::unbounded();
        let mut worker_threads = Vec::new();
        for i in 0..self.n_workers {
            let rcvr = to_run_chan.1.clone();
            let sndr = done_chan.0.clone();
            let handle = thread::Builder::new()
                .name(format!("worlflow_manager_worker_{}", i))
                .spawn(move || {
                    for to_run in rcvr.into_iter() {
                        let j_idx = to_run.job_idx;

                        let stdout = io::stdout();
                        {
                            let mut lock = stdout.lock();
                            writeln!(lock, "Starting job: {:?}", to_run.job_unit).unwrap();
                            write!(lock, "Starting job: {:#?} interpreted as: {:#?} ", to_run.job_unit, ExpandedDescr(to_run.job_unit())).unwrap();
                            writeln!(lock, "\n").unwrap();
                        }
                        super::create_out_dir(&to_run.job_unit);
                        let done_running = Command::new("bash")
                            .arg("-c")
                            .arg(to_run.job_unit.cmd().unwrap())
                            .output()
                            .unwrap();
                        if done_running.status.success() {
                            {
                                let mut lock = stdout.lock();
                                writeln!(lock, "Finished job: {:?}\n", to_run.job_unit).unwrap();
                            }
                            sndr.send(ExecutorResult::new_ok(j_idx)).unwrap()
                        } else {
                            sndr.send(ExecutorResult {
                                job_idx: j_idx,
                                result: Err(format!(
                                    "Process exited with status {:?} and stderr:\n{}",
                                    done_running.status.code(),
                                    String::from_utf8_lossy(&done_running.stderr)
                                )),
                            })
                            .unwrap();
                        }
                    }
                })
                .unwrap();
            worker_threads.push(handle);
        }
        (
            ThreadLocalExecutor {
                threads: worker_threads,
            },
            to_run_chan.0,
            done_chan.0,
            done_chan.1,
        )
    }
}

pub struct ThreadLocalExecutor {
    threads: Vec<JoinHandle<()>>,
}

impl Executor for ThreadLocalExecutor {
    fn join(self)-> Vec<std::thread::Result<()>>  {
        self.threads.into_iter().map(|jh| jh.join()).collect()
    }
}