ch20 multi-threaded example does not work

[horriblename]

• 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)

[michalSolarz]

@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.

[chriskrycho]

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!

[dillongardner]

I am confused as to how the multi-threading is supposed to work and believe the OP may be speaking to the same confusion.

I am running the code as from the book, with an additional println! in the worker:

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(move || loop {
            let message = receiver.lock().unwrap().recv();

            match message {
                Ok(job) => {
                    println!("Worker {id} got a job; executing.");

                    job();
                    println!("Worker {id} finished a job.");
                }
                Err(_) => {
                    println!("Worker {id} disconnected; shutting down.");
                    break;
                }
            }
        });

• When the server is up I load three pages as quickly as I can
  1 http://127.0.0.1:7878/sleep
  2 http://127.0.0.1:7878/sleep
  3 http://127.0.0.1:7878/
• The pages load as
  • The first page loads in ~5 seconds
  • The second page loads in ~10 seconds (5 seconds after the first page loads)
  • The third page loads almost instantaneously (before the first page even loads)

In my terminal, I see the following

Worker 0 got a job; executing.
Worker 3 got a job; executing.
Worker 3 finished a job.
Worker 0 finished a job.
Worker 1 got a job; executing.
Worker 1 finished a job.

From the book, I would have expected the second page should also load in 5 seconds and that Worker 1 should print Worker 1 got a job; executing before Worker 0 finishes its job. This suggests that some part of the code is blocking and that the second request is not being passed to thread 1 until after thread 0 has completed its job.

I am further confused as this behavior is different when calling http://127.0.0.1:7878/