asynchronous - 如何限制 async_trait 函数的返回值来实现同步？

Question

我正在使用 async_trait crate 定义一个 async特性的方法。我打算改造Future它返回到 Stream , 包裹 Stream在 Arc并发送 Arc到多个线程。最小的可重现代码如下:

use async_trait::async_trait;
use futures::stream::{unfold, Stream};
use std::sync::Arc;
use tokio::runtime::Runtime;

#[async_trait]
trait Trait: Send + Sync {
    async fn do_something(&self) -> i32;
}

async fn use_trait<T: Trait>(x: &T) {
    let boxed: Arc<Box<dyn Stream<Item = i32> + Send + Sync>>;
    let lazy_poller = unfold(None, move |state| async move {
        if let Some(value) = state {
            Some((value, Some(value)))
        } else {
            let value = x.do_something().await;
            Some((value, Some(value)))
        }
    });
    boxed = Arc::new(Box::new(lazy_poller));
    let boxed_clone = boxed.clone();
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        let _moved = boxed_clone;
        // Do something with `_moved.next()`
    });
}

但是它编译时出现以下错误:

error: future cannot be shared between threads safely
  --> src/main.rs:21:22
   |
21 |     boxed = Arc::new(Box::new(lazy_poller));
   |                      ^^^^^^^^^^^^^^^^^^^^^ future created by async block is not `Sync`
   |
   = help: the trait `Sync` is not implemented for `dyn futures::Future<Output = i32> + std::marker::Send`
note: future is not `Sync` as it awaits another future which is not `Sync`
  --> src/main.rs:17:25
   |
17 |             let value = x.do_something().await;
   |                         ^^^^^^^^^^^^^^^^ await occurs here on type `Pin<Box<dyn futures::Future<Output = i32> + std::marker::Send>>`, which is not `Sync`
   = note: required for the cast to the object type `dyn Stream<Item = i32> + Sync + std::marker::Send`

看来async_trait将异步方法的返回类型脱糖为 Pin<Box<dyn Future<...> + Send>> , 不指定 Sync .但是我认为要求 Future 也是相当普遍的。 s 是 Sync , 除了 Send .
我的问题是:

我应该如何将异步方法的返回类型指定为 Sync , 和

为什么async_trait 不是 指定 Sync自动返回类型？

Answer 1

正如评论中已经提到的，Future s 不需要是 Sync对于大多数用例；当它可能从不同的任务轮询时， FutureExt::shared 是要走的路，这正是我的用例所需要的。我不必将其转换为 Stream .
至于我的例子，它将是:

use async_trait::async_trait;
use futures::future::FutureExt;
use tokio::runtime::Runtime;

#[async_trait]
trait Trait: Send + Sync {
    async fn do_something(&self) -> i32;
}

async fn use_trait<T: Trait>(x: &T) {
    let shared = x.do_something().shared();
    let shared_clone = shared.clone();
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        let _moved = shared_clone;
        // Do something with `_moved.await`
    });
    println!("{}", shared.await)
}