book

book copied to clipboard

• I have searched open and closed issues and pull requests for duplicates, using these search terms:
  • ch20
  • multi thread
• I have checked the latest main branch to see if this has already been fixed, in this file:
  • ch20-02-multithreaded.md

URL to the section(s) of the book with this problem: https://doc.rust-lang.org/book/ch20-02-multithreaded.html#creating-a-finite-number-of-threads

Description of the problem: The multi-threaded server example does not work, from my testing it seems that the problem lies in TcpListener::incoming, it seems that incoming blocks until the previous stream is dropped. A snippet to demonstrate the issue:

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
    let pool = ThreadPool::new(4);

    for stream in listener.incoming() {
        // ^^^ incoming() blocks until the previous instance of
        // stream is dropped
        let stream = stream.unwrap();

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

fn handle_connection(mut stream: TcpStream) {
    // ...
    let (status_line, filename) = match &request_line[..] {
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));

            // to kinda prove dropping stream would unblock `incoming` in main
            drop(stream);
            thread::sleep(Duration::from_secs(5));
            return;
        }
    };

    // ...
}

Suggested fix:

I don't know

I've looked in the docs for TcpListener::incoming and TcpListener::accept (which is used by incoming), but there is no mentioned that it would block until the previous stream is dropped, so I am at a lost as to how to solve this issue. I am not experienced in TCP nor rust, I hope someone who has better insight can give their thoughts.

In case it matters: tested on linux 6.0.9 with rustc version 1.65.0 (897e37553 2022-11-02)

@horriblename I've noticed the same issue, I've managed to fix it with following fixes:

pub struct ThreadPool {
    workers: Vec<Worker>,
    // sender: Option<mpsc::Sender<Job>>,
    senders: Vec<Option<mpsc::Sender<Job>>>,
}

    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        let mut workers = Vec::with_capacity(size);
        let mut senders = Vec::with_capacity(size);

        for id in 0..size {
            let (sender, receiver) = mpsc::channel();
            let receiver = Arc::new(Mutex::new(receiver));
            
            workers.push(Worker::new(id, Arc::clone(&receiver)));
            senders.push(Some(sender))
        }

        ThreadPool {
            workers,
            // sender: Some(sender),
            senders,
        }
    }

    pub fn execute<F>(&self, f: F, thread_number: usize)
        where
            F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);

        self.senders[thread_number].as_ref().unwrap().send(job).unwrap();
    }

    let mut connection_thread = 0;
    let max_thread_index = pool_size - 1;

    for stream in listener.incoming() {
        let stream = stream.unwrap();

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

        connection_thread += 1;
        if connection_thread > max_thread_index {
            connection_thread = 0;
        }
    }

but I'm not rust or multithreading expert so there is probably some other fix. That code probably works as it should. The version from the book works in following manner:

• single request is handled by each thread, My version works this way:
• single request is handled by single thread,
• when the next request are send it is passed to next thread so server is able to handle multiple requests simultaneously.

Hmm. I'm not sure what you mean by “does not work”; can you elaborate on the problem you were having? I spent a while running this (I am literally running the crate that the book listings are pulled from) and it works exactly as described in the text, so I would need more detail to understand what the issue is!