Next, then, we’ll implement the Drop trait to call join on each of the threads in the pool so they can finish the requests they’re working on before closing. Then we’ll implement a way to tell the threads they should stop accepting new requests and shut down. To see this code in action, we’ll modify our server to accept only two requests before gracefully shutting down its thread pool.

    Let’s start with implementing Drop on our thread pool. When the pool is dropped, our threads should all join to make sure they finish their work. Listing 20-22 shows a first attempt at a Drop implementation; this code won’t quite work yet.

    Filename: src/lib.rs

    Listing 20-22: Joining each thread when the thread pool goes out of scope

    First, we loop through each of the thread pool workers. We use &mut for this because self is a mutable reference, and we also need to be able to mutate worker. For each worker, we print a message saying that this particular worker is shutting down, and then we call join on that worker’s thread. If the call to join fails, we use unwrap to make Rust panic and go into an ungraceful shutdown.

    Here is the error we get when we compile this code:

    1. $ cargo check
    2.     Checking hello v0.1.0 (file:///projects/hello)
    3. error[E0507]: cannot move out of `worker.thread` which is behind a mutable reference
    4.     --> src/lib.rs:52:13
    5.      |
    6. 52   |             worker.thread.join().unwrap();
    7.      |             ^^^^^^^^^^^^^ ------ `worker.thread` moved due to this method call
    8.      |             |
    9.      |             move occurs because `worker.thread` has type `JoinHandle<()>`, which does not implement the `Copy` trait
    10.      |
    11. note: this function takes ownership of the receiver `self`, which moves `worker.thread`
    13. For more information about this error, try `rustc --explain E0507`.
    14. error: could not compile `hello` due to previous error

    The error tells us we can’t call join because we only have a mutable borrow of each worker and join takes ownership of its argument. To solve this issue, we need to move the thread out of the Worker instance that owns thread so join can consume the thread. We did this in Listing 17-15: if Worker holds an Option> instead, we can call the take method on the Option to move the value out of the Some variant and leave a None variant in its place. In other words, a Worker that is running will have a Some variant in thread, and when we want to clean up a Worker, we’ll replace Some with None so the Worker doesn’t have a thread to run.

    So we know we want to update the definition of Worker like this:

    Filename: src/lib.rs

    1. use std::{
    2.     sync::{mpsc, Arc, Mutex},
    3.     thread,
    4. };
    6. pub struct ThreadPool {
    7.     workers: Vec<Worker>,
    8.     sender: mpsc::Sender<Job>,
    9. }
    11. type Job = Box<dyn FnOnce() + Send + 'static>;
    13. impl ThreadPool {
    14.     /// Create a new ThreadPool.
    15.     ///
    16.     /// The size is the number of threads in the pool.
    17.     ///
    18.     /// # Panics
    19.     ///
    20.     /// The `new` function will panic if the size is zero.
    21.     pub fn new(size: usize) -> ThreadPool {
    22.         assert!(size > 0);
    24.         let (sender, receiver) = mpsc::channel();
    26.         let receiver = Arc::new(Mutex::new(receiver));
    28.         let mut workers = Vec::with_capacity(size);
    30.         for id in 0..size {
    31.             workers.push(Worker::new(id, Arc::clone(&receiver)));
    32.         }
    34.         ThreadPool { workers, sender }
    35.     }
    37.     pub fn execute<F>(&self, f: F)
    38.     where
    39.         F: FnOnce() + Send + 'static,
    40.     {
    41.         let job = Box::new(f);
    43.         self.sender.send(job).unwrap();
    44.     }
    45. }
    47. impl Drop for ThreadPool {
    48.     fn drop(&mut self) {
    49.         for worker in &mut self.workers {
    50.             println!("Shutting down worker {}", worker.id);
    52.             worker.thread.join().unwrap();
    53.         }
    54.     }
    55. }
    57. struct Worker {
    58.     id: usize,
    59.     thread: Option<thread::JoinHandle<()>>,
    60. }
    62. impl Worker {
    63.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
    64.         let thread = thread::spawn(move || loop {
    65.             let job = receiver.lock().unwrap().recv().unwrap();
    67.             println!("Worker {id} got a job; executing.");
    69.             job();
    70.         });
    72.         Worker { id, thread }
    73.     }
    74. }

    Now let’s lean on the compiler to find the other places that need to change. Checking this code, we get two errors:

    1. $ cargo check
    2.     Checking hello v0.1.0 (file:///projects/hello)
    3. error[E0599]: no method named `join` found for enum `Option` in the current scope
    4.   --> src/lib.rs:52:27
    5.    |
    6. 52 |             worker.thread.join().unwrap();
    7.    |                           ^^^^ method not found in `Option<JoinHandle<()>>`
    9. error[E0308]: mismatched types
    10.   --> src/lib.rs:72:22
    11.    |
    12. 72 |         Worker { id, thread }
    13.    |                      ^^^^^^ expected enum `Option`, found struct `JoinHandle`
    14.    |
    15.    = note: expected enum `Option<JoinHandle<()>>`
    16.             found struct `JoinHandle<_>`
    17. help: try wrapping the expression in `Some`
    18.    |
    19. 72 |         Worker { id, thread: Some(thread) }
    20.    |                      +++++++++++++      +
    22. Some errors have detailed explanations: E0308, E0599.
    23. For more information about an error, try `rustc --explain E0308`.
    24. error: could not compile `hello` due to 2 previous errors

    Let’s address the second error, which points to the code at the end of Worker::new; we need to wrap the thread value in Some when we create a new Worker. Make the following changes to fix this error:

    The first error is in our Drop implementation. We mentioned earlier that we intended to call take on the Option value to move thread out of worker. The following changes will do so:

    Filename: src/lib.rs

    1. use std::{
    2.     sync::{mpsc, Arc, Mutex},
    3.     thread,
    4. };
    6. pub struct ThreadPool {
    7.     workers: Vec<Worker>,
    8.     sender: mpsc::Sender<Job>,
    9. }
    11. type Job = Box<dyn FnOnce() + Send + 'static>;
    13. impl ThreadPool {
    14.     /// Create a new ThreadPool.
    15.     ///
    16.     /// The size is the number of threads in the pool.
    17.     ///
    18.     /// # Panics
    19.     ///
    20.     /// The `new` function will panic if the size is zero.
    21.     pub fn new(size: usize) -> ThreadPool {
    22.         assert!(size > 0);
    24.         let (sender, receiver) = mpsc::channel();
    26.         let receiver = Arc::new(Mutex::new(receiver));
    28.         let mut workers = Vec::with_capacity(size);
    30.         for id in 0..size {
    31.             workers.push(Worker::new(id, Arc::clone(&receiver)));
    32.         }
    34.         ThreadPool { workers, sender }
    35.     }
    37.     pub fn execute<F>(&self, f: F)
    38.     where
    39.         F: FnOnce() + Send + 'static,
    40.     {
    41.         let job = Box::new(f);
    43.         self.sender.send(job).unwrap();
    44.     }
    45. }
    47. impl Drop for ThreadPool {
    48.     fn drop(&mut self) {
    50.             println!("Shutting down worker {}", worker.id);
    52.             if let Some(thread) = worker.thread.take() {
    53.                 thread.join().unwrap();
    54.             }
    55.         }
    56.     }
    57. }
    59. struct Worker {
    60.     id: usize,
    61.     thread: Option<thread::JoinHandle<()>>,
    62. }
    63. impl Worker {
    64.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
    65.         let thread = thread::spawn(move || loop {
    66.             let job = receiver.lock().unwrap().recv().unwrap();
    68.             println!("Worker {id} got a job; executing.");
    70.             job();
    71.         });
    73.         Worker {
    74.             id,
    75.             thread: Some(thread),
    76.         }
    77.     }
    78. }

    As discussed in Chapter 17, the take method on Option takes the Some variant out and leaves None in its place. We’re using if let to destructure the Some and get the thread; then we call join on the thread. If a worker’s thread is already None, we know that worker has already had its thread cleaned up, so nothing happens in that case.

    With all the changes we’ve made, our code compiles without any warnings. However, the bad news is this code doesn’t function the way we want it to yet. The key is the logic in the closures run by the threads of the Worker instances: at the moment, we call join, but that won’t shut down the threads because they loop forever looking for jobs. If we try to drop our ThreadPool with our current implementation of drop, the main thread will block forever waiting for the first thread to finish.

    To fix this problem, we’ll need a change in the ThreadPool drop implementation and then a change in the Worker loop.

    First, we’ll change the ThreadPool drop implementation to explicitly drop the sender before waiting for the threads to finish. Listing 20-23 shows the changes to ThreadPool to explicitly drop sender. We use the same Option and take technique as we did with the thread to be able to move sender out of ThreadPool:

    Filename: src/lib.rs

    1. use std::{
    2.     sync::{mpsc, Arc, Mutex},
    3.     thread,
    4. };
    6. pub struct ThreadPool {
    7.     workers: Vec<Worker>,
    8.     sender: Option<mpsc::Sender<Job>>,
    9. }
    10. // --snip--
    12. type Job = Box<dyn FnOnce() + Send + 'static>;
    14. impl ThreadPool {
    15.     /// Create a new ThreadPool.
    16.     ///
    17.     /// The size is the number of threads in the pool.
    18.     ///
    19.     /// # Panics
    20.     ///
    21.     /// The `new` function will panic if the size is zero.
    22.     pub fn new(size: usize) -> ThreadPool {
    23.         // --snip--
    25.         assert!(size > 0);
    27.         let (sender, receiver) = mpsc::channel();
    29.         let receiver = Arc::new(Mutex::new(receiver));
    31.         let mut workers = Vec::with_capacity(size);
    33.         for id in 0..size {
    34.             workers.push(Worker::new(id, Arc::clone(&receiver)));
    35.         }
    37.         ThreadPool {
    38.             workers,
    39.             sender: Some(sender),
    40.         }
    41.     }
    43.     pub fn execute<F>(&self, f: F)
    44.     where
    45.         F: FnOnce() + Send + 'static,
    46.     {
    47.         let job = Box::new(f);
    49.         self.sender.as_ref().unwrap().send(job).unwrap();
    50.     }
    51. }
    53. impl Drop for ThreadPool {
    54.     fn drop(&mut self) {
    55.         drop(self.sender.take());
    57.         for worker in &mut self.workers {
    58.             println!("Shutting down worker {}", worker.id);
    60.             if let Some(thread) = worker.thread.take() {
    61.                 thread.join().unwrap();
    62.             }
    63.         }
    64.     }
    65. }
    67. struct Worker {
    68.     id: usize,
    69.     thread: Option<thread::JoinHandle<()>>,
    70. }
    72. impl Worker {
    73.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
    74.         let thread = thread::spawn(move || loop {
    75.             let job = receiver.lock().unwrap().recv().unwrap();
    77.             println!("Worker {id} got a job; executing.");
    79.             job();
    80.         });
    82.         Worker {
    83.             id,
    84.             thread: Some(thread),
    85.         }
    86.     }
    87. }

    Listing 20-23: Explicitly drop sender before joining the worker threads

    Dropping sender closes the channel, which indicates no more messages will be sent. When that happens, all the calls to recv that the workers do in the infinite loop will return an error. In Listing 20-24, we change the Worker loop to gracefully exit the loop in that case, which means the threads will finish when the ThreadPool drop implementation calls join on them.

    Filename: src/lib.rs

    1. use std::{
    2.     sync::{mpsc, Arc, Mutex},
    3.     thread,
    4. };
    6. pub struct ThreadPool {
    7.     workers: Vec<Worker>,
    8.     sender: Option<mpsc::Sender<Job>>,
    9. }
    11. type Job = Box<dyn FnOnce() + Send + 'static>;
    13. impl ThreadPool {
    14.     /// Create a new ThreadPool.
    15.     ///
    16.     /// The size is the number of threads in the pool.
    17.     ///
    18.     /// # Panics
    19.     ///
    20.     /// The `new` function will panic if the size is zero.
    21.     pub fn new(size: usize) -> ThreadPool {
    22.         assert!(size > 0);
    24.         let (sender, receiver) = mpsc::channel();
    26.         let receiver = Arc::new(Mutex::new(receiver));
    28.         let mut workers = Vec::with_capacity(size);
    30.         for id in 0..size {
    31.             workers.push(Worker::new(id, Arc::clone(&receiver)));
    32.         }
    34.         ThreadPool {
    35.             workers,
    36.             sender: Some(sender),
    37.         }
    38.     }
    40.     pub fn execute<F>(&self, f: F)
    41.     where
    42.         F: FnOnce() + Send + 'static,
    43.     {
    44.         let job = Box::new(f);
    46.         self.sender.as_ref().unwrap().send(job).unwrap();
    47.     }
    48. }
    50. impl Drop for ThreadPool {
    51.     fn drop(&mut self) {
    52.         drop(self.sender.take());
    54.         for worker in &mut self.workers {
    55.             println!("Shutting down worker {}", worker.id);
    57.             if let Some(thread) = worker.thread.take() {
    59.             }
    60.         }
    61.     }
    62. }
    64. struct Worker {
    65.     id: usize,
    66.     thread: Option<thread::JoinHandle<()>>,
    67. }
    69. impl Worker {
    70.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
    71.         let thread = thread::spawn(move || loop {
    72.             match receiver.lock().unwrap().recv() {
    73.                 Ok(job) => {
    74.                     println!("Worker {id} got a job; executing.");
    75.                     job();
    76.                 }
    77.                 Err(_) => {
    78.                     println!("Worker {id} disconnected; shutting down.");
    79.                     break;
    80.                 }
    81.             }
    82.         });
    84.         Worker {
    85.             id,
    86.             thread: Some(thread),
    87.         }
    88.     }
    89. }

    Listing 20-24: Explicitly break out of the loop when recv returns an error

    Filename: src/main.rs

    Listing 20-25: Shut down the server after serving two requests by exiting the loop

    You wouldn’t want a real-world web server to shut down after serving only two requests. This code just demonstrates that the graceful shutdown and cleanup is in working order.

    The take method is defined in the Iterator trait and limits the iteration to the first two items at most. The ThreadPool will go out of scope at the end of main, and the drop implementation will run.

    Start the server with cargo run, and make three requests. The third request should error, and in your terminal you should see output similar to this: 

    1. $ cargo run
    2.    Compiling hello v0.1.0 (file:///projects/hello)
    3.     Finished dev [unoptimized + debuginfo] target(s) in 1.0s
    4.      Running `target/debug/hello`
    5. Worker 0 got a job; executing.
    6. Shutting down.
    7. Shutting down worker 0
    8. Worker 3 got a job; executing.
    9. Worker 1 disconnected; shutting down.
    10. Worker 2 disconnected; shutting down.
    11. Worker 3 disconnected; shutting down.
    12. Worker 0 disconnected; shutting down.
    13. Shutting down worker 1
    14. Shutting down worker 2
    15. Shutting down worker 3

    You might see a different ordering of workers and messages printed. We can see how this code works from the messages: workers 0 and 3 got the first two requests. The server stopped accepting connections after the second connection, and the Drop implementation on ThreadPool starts executing before worker 3 even starts its job. Dropping the sender disconnects all the workers and tells them to shut down. The workers each print a message when they disconnect, and then the thread pool calls join to wait for each worker thread to finish.

    Notice one interesting aspect of this particular execution: the ThreadPool dropped the sender, and before any worker received an error, we tried to join worker 0. Worker 0 had not yet gotten an error from recv, so the main thread blocked waiting for worker 0 to finish. In the meantime, worker 3 received a job and then all threads received an error. When worker 0 finished, the main thread waited for the rest of the workers to finish. At that point, they had all exited their loops and stopped.

    Congrats! We’ve now completed our project; we have a basic web server that uses a thread pool to respond asynchronously. We’re able to perform a graceful shutdown of the server, which cleans up all the threads in the pool.

    Here’s the full code for reference:

    Filename: src/main.rs

    1. use hello::ThreadPool;
    2. use std::fs;
    3. use std::io::prelude::*;
    4. use std::net::TcpListener;
    5. use std::net::TcpStream;
    6. use std::thread;
    7. use std::time::Duration;
    9. fn main() {
    10.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
    11.     let pool = ThreadPool::new(4);
    13.     for stream in listener.incoming().take(2) {
    14.         let stream = stream.unwrap();
    16.         pool.execute(|| {
    17.             handle_connection(stream);
    18.         });
    19.     }
    21.     println!("Shutting down.");
    22. }
    24. fn handle_connection(mut stream: TcpStream) {
    25.     let mut buffer = [0; 1024];
    26.     stream.read(&mut buffer).unwrap();
    28.     let get = b"GET / HTTP/1.1\r\n";
    29.     let sleep = b"GET /sleep HTTP/1.1\r\n";
    31.     let (status_line, filename) = if buffer.starts_with(get) {
    32.         ("HTTP/1.1 200 OK", "hello.html")
    33.     } else if buffer.starts_with(sleep) {
    34.         thread::sleep(Duration::from_secs(5));
    35.         ("HTTP/1.1 200 OK", "hello.html")
    36.     } else {
    37.         ("HTTP/1.1 404 NOT FOUND", "404.html")
    38.     };
    40.     let contents = fs::read_to_string(filename).unwrap();
    42.     let response = format!(
    43.         "{}\r\nContent-Length: {}\r\n\r\n{}",
    44.         status_line,
    45.         contents.len(),
    46.         contents
    47.     );
    49.     stream.write_all(response.as_bytes()).unwrap();
    50.     stream.flush().unwrap();
    51. }

    Filename: src/lib.rs

    1. use std::{
    2.     sync::{mpsc, Arc, Mutex},
    3.     thread,
    4. };
    6. pub struct ThreadPool {
    7.     workers: Vec<Worker>,
    8.     sender: Option<mpsc::Sender<Job>>,
    9. }
    11. type Job = Box<dyn FnOnce() + Send + 'static>;
    13. impl ThreadPool {
    14.     /// Create a new ThreadPool.
    15.     ///
    16.     /// The size is the number of threads in the pool.
    17.     ///
    18.     /// # Panics
    19.     ///
    20.     /// The `new` function will panic if the size is zero.
    21.     pub fn new(size: usize) -> ThreadPool {
    22.         assert!(size > 0);
    24.         let (sender, receiver) = mpsc::channel();
    26.         let receiver = Arc::new(Mutex::new(receiver));
    28.         let mut workers = Vec::with_capacity(size);
    30.         for id in 0..size {
    31.             workers.push(Worker::new(id, Arc::clone(&receiver)));
    32.         }
    34.         ThreadPool {
    35.             workers,
    36.             sender: Some(sender),
    37.         }
    38.     }
    40.     pub fn execute<F>(&self, f: F)
    41.     where
    42.         F: FnOnce() + Send + 'static,
    43.     {
    44.         let job = Box::new(f);
    46.         self.sender.as_ref().unwrap().send(job).unwrap();
    47.     }
    48. }
    50. impl Drop for ThreadPool {
    51.     fn drop(&mut self) {
    52.         drop(self.sender.take());
    54.         for worker in &mut self.workers {
    55.             println!("Shutting down worker {}", worker.id);
    57.             if let Some(thread) = worker.thread.take() {
    58.                 thread.join().unwrap();
    59.             }
    60.         }
    61.     }
    62. }
    64. struct Worker {
    65.     id: usize,
    66.     thread: Option<thread::JoinHandle<()>>,
    67. }
    69. impl Worker {
    70.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
    71.         let thread = thread::spawn(move || loop {
    72.             let message = receiver.lock().unwrap().recv();
    74.             match message {
    75.                 Ok(job) => {
    76.                     println!("Worker {id} got a job; executing.");
    78.                     job();
    79.                 }
    80.                 Err(_) => {
    81.                     println!("Worker {id} disconnected; shutting down.");
    82.                     break;
    83.                 }
    84.             }
    85.         });
    87.         Worker {
    88.             id,
    89.             thread: Some(thread),
    90.         }
    91.     }
    92. }
    • Add more documentation to ThreadPool and its public methods.
    • Add tests of the library’s functionality.
    • Change calls to to more robust error handling.
    • Use ThreadPool to perform some task other than serving web requests.

    Well done! You’ve made it to the end of the book! We want to thank you for joining us on this tour of Rust. You’re now ready to implement your own Rust projects and help with other peoples’ projects. Keep in mind that there is a welcoming community of other Rustaceans who would love to help you with any challenges you encounter on your Rust journey.