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
#[cfg(test)]
mod tests;
use std::{thread};
use std::sync::{Arc, mpsc, Mutex};
pub struct ThreadPool {
_workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool {
_workers: workers,
sender,
}
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
let boxed_send = self.sender.send(job);
if boxed_send.is_err() {
eprintln!("unable to send job: {}", boxed_send.err().unwrap());
} else {
boxed_send.unwrap()
}
}
}
struct Worker {
_id: usize,
_thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || loop {
let boxed_lock = receiver.lock();
if boxed_lock.is_err() {
eprintln!("Worker {} -> unable to acquire lock {}", id, boxed_lock.err().unwrap());
} else {
let boxed_job = boxed_lock.unwrap().recv();
if boxed_job.is_err() {
eprintln!("Worker {} -> unable to get job to execute {}", id, boxed_job.err().unwrap());
} else {
let job = boxed_job.unwrap();
println!("Worker {} got a job; executing.", id);
job();
}
}
});
Worker { _id: id, _thread: thread }
}
}