taskqueue 0.1.1

Parallel execution of the task queue with the ability to add new tasks inside the running tasks
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

use fehler::throws;
use std::collections::HashMap;

use crate::{
    errors,
    events::ControlToWorkerEvent,
    worker::{WorkerId, Workers},
    Error, Panic, TaskContainer,
};

enum Status {
    Free,
    Busy,
    Released,
}
type WorkersStatus = HashMap<WorkerId, Status>;

pub struct WorkersHolder {
    workers: Workers,
    workers_status: WorkersStatus,
}

impl WorkersHolder {
    pub fn new(workers: Workers) -> Self {
        let mut workers_status = WorkersStatus::new();
        for (id, _worker) in workers.iter() {
            workers_status.insert(*id, Status::Free);
        }

        Self {
            workers,
            workers_status,
        }
    }

    #[throws]
    pub fn free_worker(&mut self, id: WorkerId) {
        *(self
            .workers_status
            .get_mut(&id)
            .ok_or_else(|| errors::WorkerNotFound { id }.build())?) = Status::Free;
    }

    pub fn has_busy_workers(&self) -> bool {
        self.workers_status
            .iter()
            .any(|(_, status)| matches!(status, Status::Busy))
    }

    #[throws]
    pub fn run_tasks(&mut self, task_queue: &mut TaskContainer) {
        for (id, status) in self.workers_status.iter_mut() {
            if matches!(status, Status::Free) {
                if let Some(task) = task_queue.pop_back() {
                    self.workers
                        .get(id)
                        .ok_or_else(|| errors::WorkerNotFound { id: *id }.build())?
                        .channel
                        .send(ControlToWorkerEvent::RunTask(task))
                        .ok();

                    *status = Status::Busy;
                } else {
                    break;
                }
            }
        }
    }

    #[throws]
    pub fn keep_alive(&mut self) -> bool {
        let mut has_crashed_workers = false;
        for (id, status) in self.workers_status.iter_mut() {
            if matches!(status, Status::Busy) {
                let is_err = self
                    .workers
                    .get(id)
                    .ok_or_else(|| errors::WorkerNotFound { id: *id }.build())?
                    .channel
                    .send(ControlToWorkerEvent::KeepAlive)
                    .is_err();

                if is_err {
                    *status = Status::Released;
                    has_crashed_workers = true;
                }
            }
        }

        has_crashed_workers
    }

    #[throws]
    pub fn stop_workers(&mut self) {
        for (id, status) in self.workers_status.iter_mut() {
            if matches!(status, Status::Busy | Status::Free) {
                self.workers
                    .get(id)
                    .ok_or_else(|| errors::WorkerNotFound { id: *id }.build())?
                    .channel
                    .send(ControlToWorkerEvent::Stop)
                    .ok();

                *status = Status::Released;
            }
        }
    }

    #[throws]
    pub fn join(&mut self) -> Panic {
        let mut last_panic = None;

        for (_id, worker) in self.workers.iter_mut() {
            if let Some(worker_panic) = worker.join()? {
                last_panic = Some(worker_panic);
            }
        }

        last_panic
    }
}