Rust10 Building a Multithreaded Web Server

Rust学习笔记

Rust编程语言入门教程课程笔记

参考教材: The Rust Programming Language (by Steve Klabnik and Carol Nichols, with contributions from the Rust Community)

Lecture 20: Final Project: Building a Multithreaded Web Server

src/main.rs

use std::fs;
use std::{
    io::{prelude::*, BufReader},
    net::{TcpListener, TcpStream},
};
use std::thread;
use std::time::Duration;
use hello::ThreadPool;

fn main() {
    // bind to a port
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); 
    let pool = ThreadPool::new(4);// set a limit on the number of threads in the pool

    // listen for incoming connections
    for stream in listener.incoming().take(2) {// only accept two connections
        let stream = stream.unwrap();

        // println!("Connection established!");
        
        // handle each connection
        // handle_connection(stream);

        // spawn a new thread for each connection
        // thread::spawn(|| {
        //     handle_connection(stream);//suffer from ddos attack
        // });

        pool.execute(|| {
            handle_connection(stream);
        });
    }

    println!("Shutting down.");
}

fn handle_connection(mut stream: TcpStream) {
    // let buf_reader = BufReader::new(&mut stream);
    // let http_request: Vec<_> = buf_reader
    //     .lines()
    //     .map(|result| result.unwrap())
    //     .take_while(|line| !line.is_empty())
    //     .collect();

    // println!("Request: {:#?}", http_request);
    let mut buffer = [0; 1024];
    stream.read(&mut buffer).unwrap();
    let get = b"GET / HTTP/1.1\r\n";
    let sleep = b"GET /sleep HTTP/1.1\r\n";

    //Reading a Request

    // Request Format
    // Method Request-URI HTTP-Version CRLF
    // headers CRLF
    // message-body

    // Example

    // Request: [
    //     "GET / HTTP/1.1",
    //     "Host: 127.0.0.1:7878",
    //     "Connection: keep-alive",
    //     "sec-ch-ua: \"Google Chrome\";v=\"119\", \"Chromium\";v=\"119\", \"Not?A_Brand\";v=\"24\"",
    //     "sec-ch-ua-mobile: ?0",
    //     "sec-ch-ua-platform: \"macOS\"",
    //     "Upgrade-Insecure-Requests: 1",
    //     "User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36",
    //     "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
    //     "Sec-Fetch-Site: none",
    //     "Sec-Fetch-Mode: navigate",
    //     "Sec-Fetch-User: ?1",
    //     "Sec-Fetch-Dest: document",
    //     "Accept-Encoding: gzip, deflate, br",
    //     "Accept-Language: zh-CN,zh;q=0.9,en;q=0.8",
    // ]   

    //Writing a Response

    // Response Format
    // HTTP-Version Status-Code Reason-Phrase CRLF
    // headers CRLF
    // message-body

    //let response = "HTTP/1.1 200 OK\r\n\r\n";
    // let contents = fs::read_to_string("hello.html").unwrap();
    // let response = format!("HTTP/1.1 200 OK\r\n\r\n{}", contents);

    // write the response to the stream
    // stream.write(response.as_bytes()).unwrap();
    // stream.flush().unwrap();

    // if buffer.starts_with(get) {
    //     let contents = fs::read_to_string("hello.html").unwrap();
    //     let response = format!("HTTP/1.1 200 OK\r\n\r\n{}", contents);
    //     stream.write(response.as_bytes()).unwrap();
    //     stream.flush().unwrap();
    // } else {
    //     // other request
    //     let status_line = "HTTP/1.1 404 NOT FOUND\r\n\r\n";
    //     let contents = fs::read_to_string("404.html").unwrap();
    //     let response = format!("{}{}", status_line, contents);
    //     stream.write(response.as_bytes()).unwrap();
    //     stream.flush().unwrap();
    // }

    // Refactoring
    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
    } else if buffer.starts_with(sleep) {// simulate a slow request
        // simulate a slow request
        std::thread::sleep(std::time::Duration::from_secs(5));
        ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
    } 
    else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
    };

    let contents = fs::read_to_string(filename).unwrap();
    let response = format!("{}{}", status_line, contents);
    stream.write(response.as_bytes()).unwrap();
    stream.flush().unwrap();
}

src/lib.rs

use std::thread;
use std::sync::{mpsc, Arc, Mutex};

enum Message {
    NewJob(Job),
    Terminate,
}

struct Worker {
    id: usize,
    // thread: thread::JoinHandle<()>,
    thread: Option<thread::JoinHandle<()>>,
}

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Message>>>) -> Worker {
        // let thread = thread::spawn(|| {});
        // let thread = thread::spawn(|| {
        //     receiver;
        // });
        // let thread = thread::spawn(move || loop {
        //     while let Ok(_) = receiver.lock().unwrap().recv().unwrap(){
        //         println!("Worker {} got a job; executing.", id);
        //         job.call_box();
        //     }
        // });


        let thread = thread::spawn(move || loop {
            let message = receiver.lock().unwrap().recv().unwrap();
            match message {
                Message::NewJob(job) => {
                    println!("Worker {} got a job; executing.", id);
                    job.call_box();
                },
                Message::Terminate => {
                    println!("Worker {} was told to terminate.", id);
                    break;
                },
            }
        });
        Worker { id, thread: Some(thread) }
    }

    // fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
    //     let thread = thread::spawn(|| {
    //         // loop {
    //         //     receiver.lock().unwrap().recv().unwrap();
    //         // }
    //         loop {
    //             let job = receiver.lock().unwrap().recv().unwrap();
    //             println!("Worker {} got a job; executing.", id);
    //             job.call_box();
    //         }
    //     });
    //     Worker { id, thread }
    // }
}

pub struct ThreadPool {
    // threads: Vec<thread::JoinHandle<()>>,
    workers: Vec<Worker>,
    // sender: mpsc::Sender<Job>,
    sender: mpsc::Sender<Message>,
}

impl ThreadPool {
    /// Create a new ThreadPool
    /// 
    /// The size is the number of threads in the pool
    /// 
    /// # Panics
    /// 
    /// The `new` function will panic if the size is zero
    // 
    
    // pub fn new(size: usize) -> ThreadPool {
    //     assert!(size > 0);
    //     let (sender, receiver) = mpsc::channel();
    //     let mut workers = Vec::with_capacity(size);
    //     for id in 0..size {
    //         workers.push(Worker::new(id, receiver));
    //     }

    //     ThreadPool { workers, sender }
    // }

    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);
        let (sender, receiver) = mpsc::channel();
        let receiver = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }

        ThreadPool { workers, sender }
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);//Box<dyn FnOnce() + Send + 'static>
        // self.sender.send(job).unwrap();
        self.sender.send(Message::NewJob(job)).unwrap();
    }

    // the execute method should be similar with thread::spawn

    // pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    // where
    //     F: FnOnce() -> T + Send + 'static,
    //     T: Send + 'static,
    // {
    //     Builder::new().spawn(f).expect("failed to spawn thread")
    // }

}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        println!("Sending terminate message to all workers.");
        // for worker in &mut self.workers {
        //     println!("Shutting down worker {}", worker.id);
        //     worker.thread.join().unwrap();
        // }
        // for worker in &mut self.workers {
        //     println!("Shutting down worker {}", worker.id);
        //     // worker.thread.join().unwrap();
        //     if let Ok(_) = worker.thread.join() {
        //         println!("Worker {} shut down successfully.", worker.id);
        //     } else {
        //         println!("Worker {} failed to shut down.", worker.id);
        //     }
        // }

        for _ in &mut self.workers {
            self.sender.send(Message::Terminate).unwrap();
        }

        println!("Shutting down all workers.");

        for worker in &mut self.workers {
            // println!("Shutting down worker {}", worker.id);
            // worker.thread.join().unwrap();
            if let Some(thread) = worker.thread.take() {
                if let Ok(_) = thread.join() {
                    println!("Worker {} shut down successfully.", worker.id);
                } else {
                    println!("Worker {} failed to shut down.", worker.id);
                }
            }
        }
    }
}

// struct Job;
type Job = Box<dyn FnBox + Send + 'static>;

trait FnBox {
    fn call_box(self: Box<Self>);
}

impl <F: FnOnce()> FnBox for F {
    fn call_box(self: Box<F>) {
        (*self)()
    }
}

hello.html

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <title>Hello World</title>
</head>

<body>
    <h1>Hello World</h1>
    <p>
        <?php
        echo "Hello World";
        ?>
    </p>
</body>

</html>

404.html

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <title>404</title>
</head>

<body>
    <h1>Oops!</h1>
    <p>Page not found</p>
</body>

</html>