当将 Arc<Mutex> 传递到 tokio::spawn 时，“未来无法在线程之间安全地发送”

Question

我使用 tokio 实现了 TCP 客户端。但是，我的代码无法编译，因为出现错误：

error: future cannot be sent between threads safely
   --> src/main.rs:81:9
    |
81  |         tokio::spawn(async move {
    |         ^^^^^^^^^^^^ future created by async block is not `Send`
    |
    = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, Option<tokio::net::TcpStream>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:90:42
    |
82  |             match stream.lock().unwrap().as_mut() {
    |                   ---------------------- has type `std::sync::MutexGuard<'_, Option<tokio::net::TcpStream>>` which is not `Send`
...
90  |                     stream.write(&packet).await.unwrap();
    |                                          ^^^^^^ await occurs here, with `stream.lock().unwrap()` maybe used later
...
94  |             };
    |              - `stream.lock().unwrap()` is later dropped here
help: consider moving this into a `let` binding to create a shorter lived borrow
   --> src/main.rs:82:19
    |
82  |             match stream.lock().unwrap().as_mut() {
    |                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: required by a bound in `tokio::spawn`
   --> /playground/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.19.2/src/task/spawn.rs:127:21
    |
127 |         T: Future + Send + 'static,
    |                     ^^^^ required by this bound in `tokio::spawn`

这是我出现问题的代码：

async fn handle_write(&mut self) -> JoinHandle<()> {
    let stream = Arc::clone(&self.stream);
    let session = Arc::clone(&self.session);
    let queue = Arc::clone(&self.queue);

    tokio::spawn(async move {
        match stream.lock().unwrap().as_mut() {
            Some(stream) => {
                let packet: Vec<u8> = queue.lock().unwrap().pop_front().unwrap();
                let packet = match session.lock().unwrap().header_crypt.as_mut() {
                    Some(header_crypt) => header_crypt.encrypt(&packet),
                    _ => packet,
                };

                stream.write(&packet).await.unwrap();
                stream.flush().await.unwrap();
            },
            _ => {},
        };
    })
}

和这里同样的问题：

async fn handle_read(&mut self) -> JoinHandle<()> {
    let queue = Arc::clone(&self.queue);
    let stream = Arc::clone(&self.stream);
    let session = Arc::clone(&self.session);

    tokio::spawn(async move {
        match stream.lock().unwrap().as_mut() {
            Some(stream) => {
                let mut buffer = [0u8; 4096];

                match stream.read(&mut buffer).await {
                    Ok(bytes_count) => {
                        let raw_data = match session.lock().unwrap().header_crypt.as_mut() {
                            Some(header_crypt) => header_crypt.decrypt(&buffer[..bytes_count]),
                            _ => buffer[..bytes_count].to_vec(),
                        };

                        queue.lock().unwrap().push_back(raw_data);
                    },
                    _ => {},
                };
            },
            _ => {},
        };
    })
}

游乐场。

有人可以解释一下，我做错了什么吗？

PS以防万一：我正在使用std::sync::{Arc, Mutex};

Answer 1

最后，正如对我的问题的评论中所决定的，我使用tokio::sync::Mutex而不是std::sync::Mutex. 所以，现在代码可以正确编译了。

游乐场。