jtp 0.1.1

A simple implementation of thread pool, similar to the ThreadPoolExecutor in java.
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

use std::{
    sync::Arc,
    thread::{self, JoinHandle},
    time::Duration,
};

use crossbeam_channel::Receiver;

use crate::{task::Task, ThreadFactory};

/// A worker holds a thread handler and a receiver.
pub(crate) struct Worker {
    pub(crate) handle: JoinHandle<()>,
    pub(crate) receiver: Receiver<Task>,
    pub(crate) keep_alive_time: Duration,
    pub(crate) thread_factory: Arc<ThreadFactory>,
}

fn create_core_work_thread(
    thread_factory: &Arc<ThreadFactory>,
    receiver: Receiver<Task>,
    task: Task,
) -> JoinHandle<()> {
    let builder = thread_factory();
    builder
        .spawn(move || {
            task.run();
            while let Ok(task) = receiver.recv() {
                task.run();
            }
        })
        .expect("failed to spawn a thread.")
}

fn create_work_thread(
    thread_factory: &Arc<ThreadFactory>,
    receiver: Receiver<Task>,
    keep_alive_time: Duration,
    task: Task,
) -> JoinHandle<()> {
    let builder = thread_factory();
    builder
        .spawn(move || {
            task.run();
            while let Ok(task) = receiver.recv_timeout(keep_alive_time) {
                task.run();
            }
        })
        .expect("failed to spawn a thread.")
}

impl Worker {
    #[must_use]
    pub fn new(
        is_core: bool,
        keep_alive_time: Duration,
        thread_factory: Arc<ThreadFactory>,
        receiver: Receiver<Task>,
        task: Task,
    ) -> Self {
        Worker {
            keep_alive_time,
            receiver: receiver.clone(),
            handle: if is_core {
                create_core_work_thread(&thread_factory, receiver, task)
            } else {
                create_work_thread(&thread_factory, receiver, keep_alive_time, task)
            },
            thread_factory,
        }
    }

    /// Creates a new thread for the worker. This will make the `self`
    /// worker available again.
    pub(crate) fn restart(&mut self, task: Task) {
        debug_assert!(self.is_finished());
        self.handle = create_work_thread(
            &self.thread_factory,
            self.receiver.clone(),
            self.keep_alive_time,
            task,
        );
    }

    #[inline]
    pub fn is_finished(&self) -> bool {
        self.handle.is_finished()
    }

    #[inline]
    pub(crate) fn join(self) -> thread::Result<()> {
        if self.handle.thread().id() != thread::current().id() {
            self.handle.join()?;
        }
        Ok(())
    }
}