beekeeper 0.3.0

A full-featured worker pool library for parallelizing 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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use super::Task;
use crate::bee::{TaskId, TaskMeta, Worker};
use crate::hive::{Outcome, OutcomeSender};

/// The type of input to a task for a given `Worker` type. This changes depending on the features
/// that are enabled.
pub use task_impl::TaskInput;

impl<W: Worker> Task<W> {
    /// Returns the ID of the task.
    #[inline]
    pub fn id(&self) -> TaskId {
        self.meta.id()
    }

    /// Returns a reference to the task metadata.
    #[inline]
    pub fn meta(&self) -> &TaskMeta {
        &self.meta
    }

    /// Consumes this `Task` and returns its input, metadata, and outcome sender.
    pub fn into_parts(self) -> (W::Input, TaskMeta, Option<OutcomeSender<W>>) {
        (self.input, self.meta, self.outcome_tx)
    }

    /// Consumes this `Task` and returns an `Outcome::Unprocessed` outcome with the input and ID,
    /// and the outcome sender.
    pub fn into_unprocessed(self) -> (Outcome<W>, Option<OutcomeSender<W>>) {
        let outcome = Outcome::Unprocessed {
            input: self.input,
            task_id: self.meta.id(),
        };
        (outcome, self.outcome_tx)
    }

    /// Creates a new `Task` with the given metadata, and increments the attempt number.
    #[cfg(feature = "retry")]
    pub fn next_retry_attempt(
        input: W::Input,
        mut meta: TaskMeta,
        outcome_tx: Option<OutcomeSender<W>>,
    ) -> Self {
        meta.inc_attempt();
        Self {
            input,
            meta,
            outcome_tx,
        }
    }
}

#[cfg(not(feature = "local-batch"))]
mod task_impl {
    use super::Task;
    use crate::bee::{TaskId, TaskMeta, Worker};
    use crate::hive::OutcomeSender;

    pub type TaskInput<W> = <W as Worker>::Input;

    impl<W: Worker> Task<W> {
        /// Creates a new `Task` with the given `task_id`.
        pub fn new(
            task_id: TaskId,
            input: TaskInput<W>,
            outcome_tx: Option<OutcomeSender<W>>,
        ) -> Self {
            Task {
                input,
                meta: TaskMeta::new(task_id),
                outcome_tx,
            }
        }
    }
}

#[cfg(feature = "local-batch")]
mod task_impl {
    use super::Task;
    use crate::bee::{TaskId, TaskMeta, Worker};
    use crate::hive::{Outcome, OutcomeSender, Weighted};

    pub type TaskInput<W> = Weighted<<W as Worker>::Input>;

    impl<W: Worker> Task<W> {
        /// Creates a new `Task` with the given `task_id`.
        pub fn new(
            task_id: TaskId,
            input: TaskInput<W>,
            outcome_tx: Option<OutcomeSender<W>>,
        ) -> Self {
            let (input, weight) = input.into_parts();
            Task {
                input,
                meta: TaskMeta::with_weight(task_id, weight),
                outcome_tx,
            }
        }

        /// Consumes this `Task` and returns a `Outcome::WeightLimitExceeded` outcome with the
        /// input, weight, and ID, and the outcome sender.
        pub fn into_overweight(self) -> (Outcome<W>, Option<OutcomeSender<W>>) {
            let outcome = Outcome::WeightLimitExceeded {
                input: self.input,
                weight: self.meta.weight(),
                task_id: self.meta.id(),
            };
            (outcome, self.outcome_tx)
        }
    }
}

impl<I: Clone, W: Worker<Input = I>> Clone for Task<W> {
    fn clone(&self) -> Self {
        Self {
            input: self.input.clone(),
            meta: self.meta.clone(),
            outcome_tx: self.outcome_tx.clone(),
        }
    }
}

impl<W: Worker> PartialEq for Task<W> {
    fn eq(&self, other: &Self) -> bool {
        self.meta.id() == other.meta.id()
    }
}

impl<W: Worker> Eq for Task<W> {}

impl<W: Worker> PartialOrd for Task<W> {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl<W: Worker> Ord for Task<W> {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.meta.id().cmp(&other.meta.id())
    }
}