vastly-improved lifetime management

src/http.rs

@@ -1,9 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::HashMap, io::{BufRead, BufReader, Error, ErrorKind, Result, Write}, net::{IpAddr, SocketAddr, TcpListener, TcpStream}, str::FromStr, sync::Arc, thread,
-    io::{BufRead, BufReader, Error, ErrorKind, Result, Write},
-    net::{IpAddr, SocketAddr, TcpListener, TcpStream},
-    str::FromStr,
-    sync::Arc,
 };
 
 use chrono::Local;
@@ -35,27 +31,28 @@ impl StatusCode {
             StatusCode::OK => "OK",
             StatusCode::BadRequest => "Bad Request",
             StatusCode::NotFound => "Not Found",
-            // StatusCode::ImATeapot => "I'm a teapot",
+            StatusCode::ImATeapot => "I'm a teapot",
             StatusCode::InternalServerError => "Internal Server Error",
+            StatusCode::BadGateway => "Bad Gateway",
         }
     }
 }
 
 type HttpHandlerFunc = fn(&Request, Response, bool) -> Result<StatusCode>;
 
-pub struct Request<'a> {
+pub struct Request<'r> {
-    stream: &'a TcpStream,
+    stream: &'r TcpStream,
-    path: &'a str,
+    path: &'r str,
-    method: &'a str,
+    method: &'r str,
-    version: &'a str,
+    version: &'r str,
     headers: HashMap<String, String>,
     query: HashMap<String, String>,
     body: Option<String>,
     real_address: IpAddr,
 }
 
-impl<'a> Request<'a> {
+impl<'r> Request<'r> {
-    pub fn new(stream: &'a TcpStream, lines: &'a Vec<String>, trusted_proxies: Vec<IpAddr>) -> Result<Request<'a>> {
+    pub fn new(stream: &'r TcpStream, lines: &'r Vec<String>, trusted_proxies: Vec<IpAddr>) -> Result<Request<'r>> {
         let request_line = lines[0].as_str();
         let request_line_split: Vec<&str> = request_line.split(" ").collect();
         if request_line_split.len() < 3 {
@@ -139,13 +136,13 @@ impl<'a> Request<'a> {
     pub fn real_address(&self) -> &IpAddr {
         &self.real_address
     }
-    pub fn path(&self) -> &'a str {
+    pub fn path(&self) -> &'r str {
         self.path
     }
-    pub fn method(&self) -> &'a str {
+    pub fn method(&self) -> &'r str {
         self.method
     }
-    pub fn version(&self) -> &'a str {
+    pub fn version(&self) -> &'r str {
         self.version
     }
     pub fn body(&self) -> &Option<String> {
@@ -159,15 +156,15 @@ impl<'a> Request<'a> {
     }
 }
 
-pub struct Response<'a> {
+pub struct Response<'r> {
-    stream: &'a TcpStream,
+    stream: &'r TcpStream,
     status: StatusCode,
-    headers: HashMap<&'a str, String>,
+    headers: HashMap<&'r str, String>,
     body: Option<String>,
 }
 
-impl<'a> Response<'a> {
+impl<'r> Response<'r> {
-    pub fn new(stream: &'a TcpStream) -> Response<'a> {
+    pub fn new(stream: &'r TcpStream) -> Response<'r> {
         Response {
             stream,
             status: StatusCode::OK,
@@ -182,10 +179,10 @@ impl<'a> Response<'a> {
     pub fn status(&mut self, status: StatusCode) {
         self.status = status;
     }
-    pub fn headers(&self) -> &HashMap<&'a str, String> {
+    pub fn headers(&self) -> &HashMap<&'r str, String> {
         &self.headers
     }
-    pub fn set_header(&mut self, name: &'a str, value: String) {
+    pub fn set_header(&mut self, name: &'r str, value: String) {
         self.headers.insert(name, value);
     }
     pub fn body(&mut self, body: String) {
@@ -213,16 +210,16 @@ impl<'a> Response<'a> {
     }
 }
 
-pub struct HttpServer<'a> {
+pub struct HttpServer<'s> {
-    address: &'a str,
+    address: &'s str,
     port: u16,
     trusted_proxies: Arc<Vec<IpAddr>>,
     max_connections: usize,
     verbose: bool,
 }
 
-impl HttpServer <'_> {
+impl<'s> HttpServer <'s> {
-    pub fn new(address: &'_ str, max_connections: usize, trusted_proxies: Vec<IpAddr>, verbose: bool) -> HttpServer<'_> {
+    pub fn new(address: &'s str, max_connections: usize, trusted_proxies: Vec<IpAddr>, verbose: bool) -> HttpServer<'s> {
         let mut _address = address;
         let mut _port: u16 = 8080;
         match address.split_once(":") {
@@ -242,7 +239,8 @@ impl HttpServer <'_> {
     }
 
     pub fn start(&self, handler: HttpHandlerFunc) -> Result<()> {
-        let pool = ThreadPool::new(self.max_connections);
+        thread::scope(|scope| {
+            let pool = ThreadPool::new(scope, self.max_connections);
             let listener = TcpListener::bind(format!("{}:{}", self.address, self.port)).expect("Failed to bind to port");
             let verbose = self.verbose;
 
@@ -261,6 +259,7 @@ impl HttpServer <'_> {
                     }
                 }
             }
+        });
 
         Ok(())
     }

src/thread.rs

@@ -1,19 +1,19 @@
-use std::{sync::{mpsc, Arc, Mutex}, thread};
+use std::{marker::PhantomData, sync::{Arc, Mutex, mpsc}, thread};
 
-pub struct ThreadPool {
+pub struct ThreadPool<'s> {
-    workers: Vec<ThreadWorker>,
+    workers: Vec<ThreadWorker<'s>>,
-    sender: Option<mpsc::Sender<Job>>,
+    sender: mpsc::Sender<Job<'s>>,
 }
 
-type Job = Box<dyn FnOnce() + Send + 'static>;
+type Job<'s> = Box<dyn FnOnce() + Send + 's>;
 
-impl ThreadPool {
+impl<'s> ThreadPool<'s> {
     // Create a new ThreadPool with `size` available threads.
     //
     // # Panics
     //
     // `new` will panic if `size` is zero.
-    pub fn new(size: usize) -> ThreadPool {
+    pub fn new(scope: &'s thread::Scope<'s, '_>, size: usize) -> ThreadPool<'s> {
         assert!(size > 0);
 
         let (sender, receiver) = mpsc::channel();
@@ -23,59 +23,64 @@ impl ThreadPool {
         let mut workers = Vec::with_capacity(size);
 
         for id in 0..size {
-            workers.push(ThreadWorker::new(id, Arc::clone(&receiver)));
+            let worker = ThreadWorker::new(id, receiver.clone(), scope);
+            workers.push(worker);
         }
 
         ThreadPool {
             workers,
-            sender: Some(sender)
+            sender: sender,
         }
     }
 
     pub fn execute<F>(&self, f: F)
     where
-        F: FnOnce() + Send + 'static
+        F: FnOnce() + Send + 's
     {
         let job = Box::new(f);
 
-        self.sender.as_ref().unwrap().send(job).unwrap();
+        self.sender.send(job).expect("Failed to execute worker job")
     }
 }
 
-impl Drop for ThreadPool {
+impl<'s> Drop for ThreadPool<'s> {
     fn drop(&mut self) {
-        drop(self.sender.take());
+        for worker in self.workers.drain(..) {
-
+            worker.close();
-        for worker in &mut self.workers.drain(..) {
-            worker.thread.join().expect(
-                format!("Error in worker {}", worker.id).as_str());
         }
     }
 }
 
-struct ThreadWorker {
+struct ThreadWorker<'s> {
-    id: usize,
+    _lifetime: PhantomData<&'s ()>,
-    thread: thread::JoinHandle<()>,
+    _id: usize,
+    thread: thread::ScopedJoinHandle<'s, ()>,
 }
 
-impl ThreadWorker {
+impl<'s> ThreadWorker<'s> {
-    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> ThreadWorker {
+    fn new(_id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job<'s>>>>, scope: &'s thread::Scope<'s, '_>) -> ThreadWorker<'s> {
-        let thread = thread::spawn(move || loop {
+        let thread = scope.spawn(move || loop {
-            let msg = receiver.lock().unwrap().recv();
+            let receiver_locked = receiver.lock().expect("Failed to acquire lock");
+            let msg = receiver_locked.recv();
 
             match msg {
                 Ok(job) => {
-                    // println!("Job received by worker {id}");
+                    // println!("Job received by worker {_id}");
                     job();
                 }
                 Err(_) => {
-                    // println!("Worker {id} disconnected. Shutting down...");
+                    // println!("Worker {_id} disconnected. Shutting down...");
                     break;
                 }
             }
         });
 
-        ThreadWorker { id, thread }
+        ThreadWorker { _lifetime: PhantomData, _id, thread }
+    }
+
+    fn close(self) {
+        self.thread.join().unwrap_or_else(|err| {
+            println!("Failed to properly close thread: {err:?}")
+        })
     }
 }
-