advance-practice1/multi-threads

[giscus[bot]]

advance-practice1/multi-threads

https://course.rs/advance-practice1/multi-threads.html

[l16659]

前面学的好好的，学到这里看不懂了~~~~~~~呜呜

[li199603]

主要就是线程池的实现，可以先在你熟悉的语言看看怎么实现的，然后再回来看这个

[bxb100]

MutexGuard 实现了 Drop 然后才 unlock

[Grothendieck42]

thread '' panicked at 'called Option::unwrap() on a None value'
在这一句panic了，是我的问题吗
···
fn handle_connection(mut stream: TcpStream){
let buf_reader = BufReader::new(&mut stream);
let request_line = buf_reader.lines().next().unwrap().unwrap();
let (status_line, filename) = match &request_line[..] {
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
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}");
stream.write_all(response.as_bytes()).unwrap();
}
···
···
let request_line = buf_reader.lines().next().unwrap().unwrap();
···

[bxb100]

@Kev1n3zz 这里稍微多了一点解释: bxb100/bxb100.github.io#37 (comment)

[Kev1n3zz]

@bxb100 感谢大佬，了解了。

[hzw-the-programmer]

怀疑chrome浏览器会预先建立一些tcp链接，且不在这些链接上发数据，导致read超时后返回None

[hzw-the-programmer]

如下代码，会等待一段时间后打印出: read 0。表明读不到一行，返回了next() 返回None
fn handle_connection(mut stream: TcpStream) {
let mut buffer = [0; 1024];
match stream.read(&mut buffer) {
Ok(n) => println!("read {n}"),
Err(err) => {
println!("{err}");
return;
}
}
}

[zhanglun]

let request_line = buf_reader
        .lines()
        .next()
        .unwrap_or_else(|| Ok(String::from("")))
        .unwrap();

[cyx404]

各位大哥，小白有个地方不懂，还望大哥解答一下：

impl Worker {
    //Arc 允许多个 Worker 同时持有 receiver，而 Mutex 可以确保一次只有一个 Worker 能从 receiver 接收消息。
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(
            move || {
                loop {
                    //receiver一直持有锁，直到recv到消息，并且执行完job，那么其他的worker是不是就无法给receiver上锁了？
                    let job = receiver.lock().unwrap().recv().unwrap();
                    println!("Worker {id} got a job; executing.");
                    job();
                }
            }
        );
        // 每个 `Worker` 都拥有自己的唯一 id
        Worker { id, thread }
    }
}

let job = receiver.lock().unwrap().recv().unwrap();
这里receiver一直持有锁，直到recv到消息，并且执行完job，那么其他的worker是不是就无法给receiver上锁了，那么请求还是需要依次等待执行啊，是不是这样理解的？

[kongh]

The receiver lock is automatically unlocked when it goes out of scope. In Rust, when a variable bound by a lock guard (in this case, receiver.lock().unwrap()) goes out of scope, the lock is released.

In the provided code, the lock guard is acquired with receiver.lock().unwrap(), and then immediately used to call recv() on the receiver. After recv() returns, the lock guard is dropped as it goes out of scope, and the lock is released.

To provide a bit more context, the lock guard returned by receiver.lock().unwrap() is a smart pointer that implements the Deref and Drop traits. The Deref trait allows you to access the underlying data (in this case, the mpsc::Receiver) using the * operator, and the Drop trait guarantees that the lock will be released when the lock guard goes out of scope.

By acquiring the lock guard using receiver.lock().unwrap(), you ensure that only one worker at a time can access the receiver, preventing data races and ensuring safe concurrent access to the shared channel.

[TsGanT]

我理解这个东西的意思是，确实是依次执行，但是不存在等待，即只要有一个请求来（收到了sender send的一个Job），那么当receiver获得Job后，并未开始执行（执行时期是在下一句Job()的时候），就可以放开锁，而后后续一个worker线程（不管是谁，拿到Mutex的锁）接着等待sender发送job；因此，从最开始mpsc的时候，就注定了这是一个sender多个worker的模型；但这也符合常识，如果sender此时没有其他需要执行的任务，那么存在空闲的线程就是合理的，而一旦sender开始发送任务，则会根据receiver获取锁的情况，依次给他们派发任务；并且假如在派发任务的时候，线程池里的线程还没用完，但sender出现了阻塞，无法正确的派发任务，那么是的，在这个模型下，确实就整个阻塞了；

[araraloren]

这里lock在当前语句结束之后就会unlock了，rustc是非常擅长在合适的时候运行drop的

[limuyao1996]

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

            drop(_receiver);

            println!("Worker {id} got a job; executing.");

            job();
        });
        // 每个 `Worker` 都拥有自己的唯一 id
        Worker { id, thread }
    }
}

感觉这样也可以，更好理解

[wyman-lol]

let _receiver = receiver.lock().unwrap(); 可不行，右边的任何临时变量会在 let 语句结束后立即被 drop，因此锁会自动释放，这时候再去recv()可能锁已经不在当前线程。

[wwq1997]

21  |             workers.push(Worker::new(id,Arc::clone(&receiver)));
    |                                         ---------- ^^^^^^^^^ expected `&Arc<Mutex<Receiver<Job>>>`, found 
`&Receiver<_>`
    |                                         |
    |                                         arguments to this function are incorrect

这个报错怎么解决呀

[lzh-marble]

前面的代码呢？我猜是这样？

let (sender, receiver) = mpsc::channel();
workers.push(Worker::new(id,Arc::clone(&receiver)));

应该要再加一行代码

let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver); // 这里
workers.push(Worker::new(id,Arc::clone(&receiver)));

[lzh-marble]

type Job = Box<dyn FnOnce() + Send + 'static>;

请问这里dyn分发的是FnOnce吗？为什么Send不需要dyn分发呢？

[Risuntsy]

dyn分发的应该是是特征，后面的几个特征可以看作是个整体

[lzh-marble]

感谢回复🙏。FnOnce() + Send + 'static作为一个整体的特征对象，已从官方文档了解到

[lifei6671]

这也太难了。

[TsGanT]

有一个问题呀，这看起来worker是多线程的，但是发送者其实还是只有一个线程，那是不是说明如果发送者被阻塞了，其实整体还是都会被阻塞住？也就是我在pool.execute被比如某段垃圾代码阻塞了，那后续进来的请求就都没法执行响应了？

[vicolj]

我理解也是这样。好在前面的逻辑足够简单也固定，收到消息之后就立马传给多线程处理了。

[wjfflash]

你说的对，锁虽然被释放了，但是不task不执行完 recv无法获取

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(move || loop {
            let lock =  receiver.lock();
            println!("id = {id} get lock");
            let msg = lock.unwrap().recv();
            println!("id = {id} get msg"); // 这里解锁
            match msg {
                Ok(task) => {
                    println!("id = {id} sleep");
                    thread::sleep(Duration::from_secs(10));
                    println!("Worker {id} got a job; executing.");
                    task();
                    // task执行完成后释放recv()新的消息其他线程虽然获取到锁了但是无法获取msg
                    // println!("id = {id} sleep");
                    // thread::sleep(Duration::from_secs(10));
                    // println!("Worker {id} got a job; executing.");
                }
                _ => {
                    println!("Worker {id} got a job; shutdown.");
                    break;
                }
            }

        });

        Worker { id, thread:Some(thread)}
    }
}

[ganpengcen]

打螺丝会了 该造飞机了

[vicolj]

整体逻辑大概应该是看懂了-好像很简单我还看了好几遍才梳理通：( ：ThreadPoll 在new的时候通过worker创建4个线程，和一个channel，让worker一直准备接受数据。当web界面接受点击时，用execute函数通过channel进行传输。例子中还有很多用法需要再好好研究。

[2102819372]

写出来了，但是我理解的线程池，不就是0线程阻塞，就到1线程上运行吗，我线程池开为2两个，然后阻塞10s，然后用两个页面访问/sleep，再用一个访问原始路径，这个原始路径竟然没阻塞

[KinonoChen]

“两个页面访问/sleep”，浏览器对同一种请求会排队处理

[wangzhen0518]

rust有什么推荐的线程池库吗，找了一下没找到

[zhangtgv]

hi，我有一个疑问，我测试的时候是按照（慢请求，快请求，慢请求）的顺序进行测试的。第一个慢请求按照预期再5秒后返回，第二个快请求也是按照预期瞬间返回，但是最后的这个慢请求却是要等到第一个慢请求完成后再等待5秒才会进行返回。大佬能否解释一下最后一个请求的行为，谢谢

[YdayangY]

为什么请求一次会有三个线程进行接受任务但是只有一个成功其余两个失败？看博主的仿佛不是这样，是因为调用出的代码是async的吗？
我的代码:

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

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}
type Job = Box<dyn FnOnce() + Send + 'static >;

impl ThreadPool {
    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}
    }

    /// 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 execute<F>(&self,f: F)
    where
        F: FnOnce() +Send + 'static,
    {
        let job = Box::new(f);
        self.sender.send(job).unwrap();
    }
}

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

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(move || loop {
           let job = match receiver.lock().unwrap().recv() {
                Ok(x) => x,
                Err(e) => {
                    panic!("Worker {id} got a error: {:?}", e);
                },
           };
           println!("Worker {id} got a job; executing.");
           job();
        });
        Worker {id, thread}
    }
}

输出如下：

Worker 0 got a job; executing.
thread '<unnamed>' panicked at src\lib.rs:51:21:
Worker 2 got a error: RecvError
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread '<unnamed>' panicked at src\lib.rs:48:44:
called `Result::unwrap()` on an `Err` value: PoisonError { .. }
thread '<unnamed>thread 'Worker 0 got a job; executing.
' panicked at <unnamed>src\lib.rs' panicked at :src\lib.rs51::4821::
44Worker 1 got a error: RecvError:

[zegzag]

啥时候该用 async/await，啥时候该用线程池呀。真的头大，不知道怎么抉择。感觉 async/await 多是用于已经比较明确异步函数代码该写在什么地方的场景，而线程池是不太清楚任务会在哪块代码和什么时候提交任务的场景。

[limuyao1996]

我理解 CPU 密集操作用线程，这种的一个线程上async/await还是排着队来。IO密集操作用async/await，线程执行很快，大部分时间在等IO，所以不需要多个线程。

[Dchih]

这个例子闭包这里是不是少了move
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

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

    thread::spawn(move || {  // 移入stream所有权
        handle_connection(stream);
    });
}

}