异步块创建的 future 不是 `Send`

Question

我用 Rust 做了服务器更新。它在两个二进制文件之间创建补丁，并提供静态文件

我尝试做

    let mut update_state;
    if let Some(state) = update_stream.next().await {
        if let Ok(state) = state {
            update_state = state
        } else if let Err(err) = state {
            reply = BuildOutput { error: "Update failed: ".to_string() + &err.to_string() }
        }
    } else {
        reply = BuildOutput { error: "Unreacheable".to_string() }
    }

    let state = update_state.borrow();
    let progress = state.histogram.progress();

    let res = update_stream.try_for_each(|_state| future::ready(Ok(()))).await;

但得到

note: future is not `Send` as this value is used across an await
   --> server\grpc\src\rpc.rs:260:50
    |
259 |         let mut update_state;
    |             ---------------- has type `SharedUpdateProgress` which is not `Send`
260 |         if let Some(state) = update_stream.next().await {
    |                                                  ^^^^^^ await occurs here, with `mut update_state` maybe used later
...
305 |     }
    |     - `mut update_state` is later dropped here
    = note: required for the cast from `impl futures::Future<Output = Result<tonic::Response<BuildOutput>, Status>>` to the object type `dyn futures::Future<Output = Result<tonic::Response<BuildOutput>, Status>> + std::marker::Send

共享更新进度：

#[derive(Clone)]
pub struct SharedUpdateProgress {
    state: Rc<RefCell<UpdateProgress>>,
}

impl SharedUpdateProgress {
    pub fn new(target_revision: CleanName) -> Self {
        Self { state: Rc::new(RefCell::new(UpdateProgress::new(target_revision))) }
    }

    pub fn borrow(&self) -> Ref<'_, UpdateProgress> {
        self.state.borrow()
    }

    pub(crate) fn borrow_mut(&self) -> RefMut<'_, UpdateProgress> {
        self.state.borrow_mut()
    }
}

我不知道为什么，也不知道如何解决

Answer 1

我假设您的问题的最小可重现示例如下：

use std::{cell::RefCell, rc::Rc};
use tokio::time::{sleep, Duration};

#[derive(Clone)]
pub struct SharedString {
    state: Rc<RefCell<String>>,
}

impl SharedString {
    pub fn new(initial: &str) -> Self {
        Self {
            state: Rc::new(RefCell::new(initial.into())),
        }
    }
}

async fn run() {
    let shared_string = SharedString::new("Hello,");
    sleep(Duration::from_millis(1)).await;
    *shared_string.state.borrow_mut() += " world!";
    sleep(Duration::from_millis(1)).await;
    println!("{:?}", shared_string.state.borrow());
}

#[tokio::main]
async fn main() {
    tokio::task::spawn(run()).await.unwrap();
}

error: future cannot be sent between threads safely
   --> src/main.rs:27:24
    |
27  |     tokio::task::spawn(run()).await.unwrap();
    |                        ^^^^^ future returned by `run` is not `Send`
    |
    = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `Rc<RefCell<String>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:19:36
    |
18  |     let shared_string = SharedString::new("Hello,");
    |         ------------- has type `SharedString` which is not `Send`
19  |     sleep(Duration::from_millis(1)).await;
    |                                    ^^^^^^ await occurs here, with `shared_string` maybe used later
...
23  | }
    | - `shared_string` is later dropped here
note: required by a bound in `tokio::spawn`
   --> /home/martin/.cargo/git/checkouts/tokio-dd4afa005f1f4b79/686577b/tokio/src/task/spawn.rs:163:21
    |
163 |         T: Future + Send + 'static,
    |                     ^^^^ required by this bound in `tokio::spawn`

tokio 运行时通常是多线程的，这意味着.await您的任务可以在任何时候从一个线程转移到另一个线程。这就是为什么跨越一个.await点的所有东西都必须是Send。这Rc<RefCell<>>显然不是，因为它是一个单线程引用计数器。

解决方案：替换Rc<RefCell<>>为Arc<Mutex<>>，这是线程安全的等效项。

error: future cannot be sent between threads safely
   --> src/main.rs:27:24
    |
27  |     tokio::task::spawn(run()).await.unwrap();
    |                        ^^^^^ future returned by `run` is not `Send`
    |
    = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `Rc<RefCell<String>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:19:36
    |
18  |     let shared_string = SharedString::new("Hello,");
    |         ------------- has type `SharedString` which is not `Send`
19  |     sleep(Duration::from_millis(1)).await;
    |                                    ^^^^^^ await occurs here, with `shared_string` maybe used later
...
23  | }
    | - `shared_string` is later dropped here
note: required by a bound in `tokio::spawn`
   --> /home/martin/.cargo/git/checkouts/tokio-dd4afa005f1f4b79/686577b/tokio/src/task/spawn.rs:163:21
    |
163 |         T: Future + Send + 'static,
    |                     ^^^^ required by this bound in `tokio::spawn`

"Hello, world!"