rust_web_server/thread_pool/
mod.rs

1#[cfg(test)]
2mod tests;
3
4use std::{thread};
5use std::sync::{Arc, mpsc, Mutex};
6
7pub struct ThreadPool {
8    _workers: Vec<Worker>,
9    sender: mpsc::Sender<Job>,
10}
11
12type Job = Box<dyn FnOnce() + Send + 'static>;
13
14impl ThreadPool {
15    pub fn new(size: usize) -> ThreadPool {
16        assert!(size > 0);
17
18        let (sender, receiver) = mpsc::channel();
19
20        let receiver = Arc::new(Mutex::new(receiver));
21
22        let mut workers = Vec::with_capacity(size);
23        for id in 0..size {
24            workers.push(Worker::new(id, Arc::clone(&receiver)));
25        }
26
27        ThreadPool {
28            _workers: workers,
29            sender,
30        }
31    }
32
33    pub fn execute<F>(&self, f: F)
34        where
35            F: FnOnce() + Send  + 'static,
36    {
37        let job = Box::new(f);
38        let boxed_send = self.sender.send(job);
39        if boxed_send.is_err() {
40            eprintln!("unable to send job: {}", boxed_send.err().unwrap());
41        } else {
42            boxed_send.unwrap()
43        }
44
45    }
46
47    /// Drain the pool: stop accepting new jobs and wait for all in-flight
48    /// workers to finish. Consumes `self`.
49    pub fn join(mut self) {
50        drop(self.sender);
51        for worker in self._workers.drain(..) {
52            if let Err(e) = worker._thread.join() {
53                eprintln!("worker thread panicked: {:?}", e);
54            }
55        }
56    }
57}
58
59struct Worker {
60    _id: usize,
61    _thread: thread::JoinHandle<()>,
62}
63
64impl Worker {
65    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
66        let builder = thread::Builder::new().name(format!("{}", id));
67
68        let boxed_thread = builder.spawn(move || loop {
69
70            let boxed_lock = receiver.lock();
71            if boxed_lock.is_err() {
72                eprintln!("Worker {} -> unable to acquire lock {}", id, boxed_lock.err().unwrap());
73            } else {
74                let boxed_job = boxed_lock.unwrap().recv();
75                match boxed_job {
76                    Ok(job) => job(),
77                    Err(_) => break,
78                }
79            }
80
81        });
82
83        if boxed_thread.is_err() {
84            eprintln!("Failed while creating a thread id: {} error: {}", id, boxed_thread.as_ref().err().unwrap());
85        }
86
87        Worker { _id: id, _thread: boxed_thread.unwrap() }
88    }
89}
rust_web_server/thread_pool/mod.rs

rust_web_server/thread_pool/
mod.rs
