Gracefully shutdown and cleanup after handling 3 requests.

Author: polyactis

c20_web_server/src/lib.rs

@@ -3,7 +3,8 @@ use std::{sync::{mpsc, Arc, Mutex}, thread};
 
 pub struct ThreadPool{
     workers: Vec<Worker>,
-    sender: mpsc::Sender<Job>,
+    // Option<> to wrap sender so that ownership can be taken out sometime later.
+    sender: Option<mpsc::Sender<Job>>,
 }
 
 impl ThreadPool{
@@ -17,34 +18,57 @@ impl ThreadPool{
             workers.push(Worker::new(id, Arc::clone(&receiver)));
 
         }
-        ThreadPool {workers, sender}
+        ThreadPool {workers, sender:Some(sender)}
     }
 
     pub fn execute<F>(&self, f:F)
     where
         F: FnOnce() + Send + 'static,
         {
             let job = Box::new(f);
-            self.sender.send(job).unwrap();
 
+            self.sender.as_ref().expect("Fail to unwrap self.sender").send(job).unwrap();
         }
 }
 
+impl Drop for ThreadPool {
+    fn drop(&mut self) {
+        //drop() can handle Option<T>?
+        drop(self.sender.take());
+        for worker in &mut self.workers {
+            eprintln!("Shutting down work {}", worker.id);
+            if let Some(thread) = worker.thread.take(){
+                thread.join().unwrap();
+            }
+        }
+    }
+}
+
 type Job = Box<dyn FnOnce() + Send + 'static>;
 
 struct Worker {
     id: usize,
-    thread: thread::JoinHandle<()>,
+    // Option<> to wrap thread so that ownership can be taken out sometime later.
+    thread: Option<thread::JoinHandle<()>>,
 }
 
 impl Worker {
     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>> ) -> Worker {
         eprintln!("Worker {id} constructed and spawned a thread that loops forever over a receiver.");
         let thread = thread::spawn(move || loop {
-            let job = receiver.lock().unwrap().recv().unwrap();
-            eprintln!("Worker {id} got a job and is to execute it.");
-            job();
+            let mesg = receiver.lock().unwrap().recv();
+            match mesg {
+                Ok(job) => {
+                    eprintln!("Worker {id} got a job and is to execute it.");
+                    job();
+                }
+                Err(err) => {
+                    eprintln!("Worker {id} received an error {:?} and is shutting down.", err);
+                    break;
+                }
+            }
         });
-        Worker {id, thread}
+        // specify the field name for thread as it is Some(thread), not simply thread.
+        Worker {id, thread:Some(thread)}
     }
 }

c20_web_server/src/main.rs

@@ -24,19 +24,25 @@ fn handle_connection(mut stream: TcpStream) {
     };
     let contents = fs::read_to_string(filename).unwrap();
     let length = contents.len();
-    let response = format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
+    let response = format!(
+        "{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
     stream.write_all(response.as_bytes()).unwrap();
+    // flush the buffer
+    stream.flush().unwrap();
     eprintln!(" Done.");
 }
 
 fn main() {
     // [0; 3] is [0,0,0];
     //println!("{:#?}", &[0; 3]);
-    println!("Listening at TCP 127.0.0.1:7878.");
+    let num_threads = 4;
+    let num_requests = 3;
+    println!("Listening at TCP 127.0.0.1:7878 with {num_threads} threads.");
     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
-    let pool = ThreadPool::new(4);
+    let pool = ThreadPool::new(num_threads);
+    println!("Will gracefully exit after handling {num_requests} requests.");
 
-    for stream in listener.incoming() {
+    for stream in listener.incoming().take(num_requests) {
         let stream = stream.unwrap();
 
         pool.execute(|| {