scoped_threadpool_std/
thread_pool.rs

1use std::sync::{mpsc, Arc, Mutex};
2use std::thread;
3
4// https://doc.rust-lang.org/book/ch20-02-multithreaded.html
5pub struct ThreadPool<'scope, 'env> {
6    pub workers: Vec<Worker<'scope>>,
7    sender: mpsc::Sender<Job<'env>>,
8}
9
10// impl<'scope, 'env> Clone for ThreadPool<'scope, 'env> {
11//     fn clone(&self) -> Self {
12//         Self {
13//             workers: self.workers.clone(),
14//             sender: self.sender.clone(),
15//         }
16//     }
17// }
18
19type Job<'env> = Box<dyn FnOnce() + Send + 'env>;
20
21impl<'scope, 'env> ThreadPool<'scope, 'env> {
22    /* *
23     * Creates a new thread pool with the given number of worker threads.
24     * All worker threads should start immediately so they can perform work
25     * as soon as self.execute() is called.
26     *
27     * @param num_worker_threads the number of threads in the pool
28     * @param scope the thread pool's variable lifetime scope.
29     * @return the new thread pool
30     * Lifetimes 'scope and 'env are documented here
31     * https://doc.rust-lang.org/nightly/std/thread/fn.scope.html#lifetimes
32     */
33    pub fn new(
34        num_worker_threads: usize,
35        scope: &'scope thread::Scope<'scope, 'env>,
36    ) -> ThreadPool<'scope, 'env> {
37        assert!(num_worker_threads > 0);
38
39        let (sender, receiver) = mpsc::channel();
40        let receiver = Arc::new(Mutex::new(receiver));
41
42        let mut workers = Vec::with_capacity(num_worker_threads);
43
44        for id in 0..num_worker_threads {
45            workers.push(Worker::new(id, scope, Arc::clone(&receiver)));
46        }
47
48        ThreadPool { workers, sender }
49    }
50    /* *
51     * Adds work to a thread pool.
52     * The work will be performed by a worker thread as soon as all previous work is finished.
53     *
54     * @param f the function to call on a thread in the thread pool
55     * Note the lifetime 'env on function f in accordance to
56     * https://doc.rust-lang.org/nightly/std/thread/fn.scope.html#lifetimes
57     */
58    pub fn execute<F>(&self, f: F)
59    where
60        F: FnOnce() + Send + 'env,
61    {
62        let job = Box::new(f);
63
64        self.sender.send(job).unwrap();
65    }
66}
67
68pub struct Worker<'scope> {
69    pub id: usize,
70    pub thread: thread::ScopedJoinHandle<'scope, ()>,
71}
72
73impl<'scope, 'env> Worker<'scope> {
74    /* *
75     * Worker that processes jobs added to the threadpool as they are added.
76     *
77     * @param id the worker (thread's) identifying number.
78     * @param scope the variable lifetime scope to use.
79     * @param receiver is the queue that is waited on for new jobs to execute.
80     * @return the the worker
81     */
82    fn new(
83        id: usize,
84        scope: &'scope thread::Scope<'scope, 'env>,
85        receiver: Arc<Mutex<mpsc::Receiver<Job<'env>>>>,
86    ) -> Worker<'scope> {
87        let thread = scope.spawn(move || loop {
88            let job = {
89                let lock = receiver
90                    .lock()
91                    .expect("Worker thread unable to lock the job receiver");
92                lock.recv()
93            };
94
95            // println!("Worker {} got a job; executing.", id);
96
97            let job = match job {
98                Ok(job) => job,
99                // ThreadPool has been dropped
100                Err(..) => break,
101            };
102
103            job();
104
105            // println!("Worker {} completed its job; executing.", id);
106        });
107
108        Worker { id, thread }
109    }
110}
scoped_threadpool_std/thread_pool.rs

scoped_threadpool_std/
thread_pool.rs
