coordinator 0.1.1

Coordinator is a library to load balance tasks into task runners
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

use std::{
    fmt::Debug,
    hash::Hash,
    panic::UnwindSafe,
    sync::{atomic::Ordering, Arc, OnceLock},
};

use crate::{
    inner::CoordinatorInner,
    signal::ImplJoinHandle,
    task::{TaskPrefs, TaskProcessErr, TaskProcessor},
};

#[derive(Debug)]
pub struct Coordinator<TIn, TOut, TIdent>(
    OnceLock<Arc<CoordinatorInner<TIn, TOut, TIdent>>>,
    usize,
)
where
    TIdent: Hash + Eq + Ord + Debug + Clone + Send + Sync + 'static,
    TIn: Send + UnwindSafe + 'static,
    TOut: Send + UnwindSafe + 'static;

impl<TIn, TOut, TIdent> Coordinator<TIn, TOut, TIdent>
where
    TIdent: Hash + Eq + Ord + Debug + Clone + Send + Sync + 'static,
    TIn: Send + UnwindSafe + 'static,
    TOut: Send + UnwindSafe + 'static,
{
    fn get_inner(&self) -> &Arc<CoordinatorInner<TIn, TOut, TIdent>> {
        self.0.get_or_init(|| CoordinatorInner::new(self.1))
    }

    /// Create a new coordinator
    ///
    /// # Arguments
    /// * `queue_threshold` - Specify the threshold when the number of pending tasks in a worker queue
    /// prevents [`TaskPrefs::Preferred`] from adding new task to this queue
    pub fn new(queue_threshold: usize) -> Self {
        Self(OnceLock::new(), queue_threshold)
    }

    pub async fn run(
        &self,
        input: TIn,
        prefs: TaskPrefs<TIdent>,
    ) -> Result<ImplJoinHandle<Result<(TOut, TIdent), TaskProcessErr>>, TIn> {
        self.get_inner().run(input, prefs).await
    }

    pub async fn add_worker<TP>(&self, ident: TIdent, worker: TP) -> bool
    where
        TP: TaskProcessor<TIn, Output = TOut> + UnwindSafe + Send + Sync + 'static,
        TIn: UnwindSafe,
        TOut: UnwindSafe,
        TIdent: Debug,
    {
        let arc = self.get_inner().clone();
        arc.add_worker(worker, ident).await
    }

    pub async fn remove_worker(&self, ident: &TIdent) -> bool {
        self.get_inner().remove_worker(ident).await
    }
}

impl<TIn, TOut, TIdent> Drop for Coordinator<TIn, TOut, TIdent>
where
    TIdent: Hash + Eq + Ord + Debug + Clone + Send + Sync + 'static,
    TIn: Send + UnwindSafe + 'static,
    TOut: Send + UnwindSafe + 'static,
{
    fn drop(&mut self) {
        let prev = self
            .get_inner()
            .consumer_count
            .fetch_sub(1, Ordering::Release);
        if prev == 1 {
            self.get_inner().event_coordinator_abort.notify(usize::MAX);
        }
    }
}

impl<TIn, TOut, TIdent> Clone for Coordinator<TIn, TOut, TIdent>
where
    TIdent: Hash + Eq + Ord + Debug + Clone + Send + Sync + 'static,
    TIn: Send + UnwindSafe + 'static,
    TOut: Send + UnwindSafe + 'static,
{
    /// Create another consumer of this [`Coordinator`]. Note that this only increase the reference count
    /// on a internal structure, not creating a new coordinator
    fn clone(&self) -> Self {
        self.get_inner()
            .consumer_count
            .fetch_add(1, Ordering::Release);
        Self(self.0.clone(), self.1)
    }
}