rust_web_server/thread_pool/
mod.rs

1#[cfg(test)]
2mod tests;
3
4use std::sync::atomic::Ordering;
5use std::sync::{Arc, mpsc, Mutex};
6use std::thread;
7
8use crate::metrics;
9
10pub struct ThreadPool {
11    _workers: Vec<Worker>,
12    sender: mpsc::Sender<Job>,
13}
14
15type Job = Box<dyn FnOnce() + Send + 'static>;
16
17impl ThreadPool {
18    pub fn new(size: usize) -> ThreadPool {
19        assert!(size > 0);
20
21        let (sender, receiver) = mpsc::channel();
22
23        let receiver = Arc::new(Mutex::new(receiver));
24
25        let mut workers = Vec::with_capacity(size);
26        for id in 0..size {
27            workers.push(Worker::new(id, Arc::clone(&receiver)));
28        }
29
30        ThreadPool {
31            _workers: workers,
32            sender,
33        }
34    }
35
36    pub fn execute<F>(&self, f: F)
37        where
38            F: FnOnce() + Send  + 'static,
39    {
40        metrics::THREAD_POOL_QUEUED.fetch_add(1, Ordering::Relaxed);
41        let job = Box::new(move || {
42            metrics::THREAD_POOL_QUEUED.fetch_sub(1, Ordering::Relaxed);
43            f();
44        });
45        let boxed_send = self.sender.send(job);
46        if boxed_send.is_err() {
47            eprintln!("unable to send job: {}", boxed_send.err().unwrap());
48        } else {
49            boxed_send.unwrap()
50        }
51
52    }
53
54    /// Drain the pool: stop accepting new jobs and wait for all in-flight
55    /// workers to finish. Consumes `self`.
56    pub fn join(mut self) {
57        drop(self.sender);
58        for worker in self._workers.drain(..) {
59            if let Err(e) = worker._thread.join() {
60                eprintln!("worker thread panicked: {:?}", e);
61            }
62        }
63    }
64}
65
66struct Worker {
67    _id: usize,
68    _thread: thread::JoinHandle<()>,
69}
70
71impl Worker {
72    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
73        let builder = thread::Builder::new().name(format!("{}", id));
74
75        let boxed_thread = builder.spawn(move || loop {
76
77            // A poisoned lock still guards a usable receiver — recover it
78            // rather than spinning on the error forever.
79            let boxed_job = receiver.lock().unwrap_or_else(|e| e.into_inner()).recv();
80            match boxed_job {
81                // A panicking job must not take the worker with it: otherwise
82                // each panic permanently shrinks the pool, and a client that
83                // can trigger one can stop the server with `size` requests.
84                Ok(job) => {
85                    if let Err(panic) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)) {
86                        let message = panic
87                            .downcast_ref::<&str>()
88                            .map(|s| s.to_string())
89                            .or_else(|| panic.downcast_ref::<String>().cloned())
90                            .unwrap_or_else(|| "non-string panic payload".to_string());
91                        eprintln!("Worker {}: request handler panicked ({}); worker continues", id, message);
92                        metrics::record_error();
93                    }
94                }
95                Err(_) => break,
96            }
97
98        });
99
100        if boxed_thread.is_err() {
101            eprintln!("Failed while creating a thread id: {} error: {}", id, boxed_thread.as_ref().err().unwrap());
102        }
103
104        Worker { _id: id, _thread: boxed_thread.unwrap() }
105    }
106}
rust_web_server/thread_pool/mod.rs

rust_web_server/thread_pool/
mod.rs
