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
86
87
88
89
90
91

//! Module containing the traits needed to be a job executor as well as provided job executors
use std::{
    io,
    io::Write,
    process::Command,
    thread::{self, JoinHandle},
};

use crossbeam_channel::{Receiver, Sender};
use petgraph::prelude::NodeIndex;

pub mod dry_run_executor;
pub use dry_run_executor::DryRunExecutorBuilder;
pub mod slurm_executor;
pub use slurm_executor::SlurmExecutorBuilder;
pub mod thread_local;
pub use thread_local::ThreadLocalExecutorBuilder;


use crate::{JobUnit, ToRun};

type JobResult = Result<(), String>;

/// The result of running the job on the executor
#[derive(Debug)]
pub struct ExecutorResult {
    pub(crate) job_idx: NodeIndex,
    pub(crate) result: JobResult,
}


impl ExecutorResult {
    /// Signals that job number `job_idx` has run succesfully.
    /// 
    /// `job_idx` must come from [`ToRun::job_idx()`].
    pub fn new_ok(job_idx: NodeIndex) -> Self {
        Self {
            job_idx,
            result: Ok(()),
        }
    }

    /// Signals that job number `job_idx` has failed.
    /// 
    /// `job_idx` must come from [`ToRun::job_idx()`], and `err` describes
    /// what went wrong.
    pub fn new_error(job_idx: NodeIndex, err: String) -> Self {
        Self {
            job_idx,
            result: Err(err)
        }
    }
}

/// Trait used to implement the function creating the communication channels and starting the worker threads.
pub trait ExecutorBuilder {
    /// The associated [`Executor`] type.
    type Executor: Executor;

    /// Creates the communication channel *ToRun* and *Results*
    /// 
    /// The executor should accept jobs to run on *ToRun*, run them, and send the executor result on *Results*.
    /// 
    /// Returns
    /// -------
    ///  1. The [`Executor`] object,
    ///  2. The send end of crossbeam communication channel *ToRun*,
    ///  3. The send end of the crossbeam channel *Results*,
    ///  4. The receive end of the crossbeam channel *Results*.
    fn init<J: JobUnit>(
        self,
    ) -> (
        Self::Executor,
        Sender<ToRun<J>>,
        Sender<ExecutorResult>,
        Receiver<ExecutorResult>,
    );
}

/// A trait representing a running executor
pub trait Executor {
    /// Similar to the standard [`join`](`std::thread::JoinHandle::join`), except that there can be several threads which must all return `()`.
    fn join(self) -> Vec<std::thread::Result<()>>;
}

fn create_out_dir<J: JobUnit>(j_u: &J) {
    std::fs::create_dir_all(j_u.out_file().parent().unwrap()).unwrap();
    if let Some(p) = j_u.log_file() {
        std::fs::create_dir_all(p.parent().unwrap()).unwrap();
    }
}