等待编译时未知的未来数

Question

我想利用Tokio的运行时来处理可变数量的异步期货。由于在编译时期货的计数是未知的，因此 FuturesUnorderd 似乎是我的最佳选择（像 select！ 这样的宏需要在编译时指定你的分支;join_all可能是可能的，但是当顺序无关紧要时，文档建议“在很多情况下”FuturesUnordered）。

这个片段的逻辑是一个recv()循环，它被推送到期货桶中，应该一直运行。当新数据到达时，它的解析/处理也被推到futures桶中(而不是立即被处理)。这确保了接收器在响应新事件时保持低延迟，并且数据处理(潜在地计算昂贵的解密)与所有其他数据处理异步块(加上监听接收器)同时发生。

顺便说一下，这个线程解释了为什么期货会得到. box（）。

问题是这个神秘的错误：

error[E0277]: `dyn futures::Future<Output = ()> + std::marker::Send` cannot be shared between threads safely
  --> src/main.rs:27:8
   |
27 |     }).boxed());
   |        ^^^^^ `dyn futures::Future<Output = ()> + std::marker::Send` cannot be shared between threads safely
   |
   = help: the trait `Sync` is not implemented for `dyn futures::Future<Output = ()> + std::marker::Send`
   = note: required because of the requirements on the impl of `Sync` for `Unique<dyn futures::Future<Output = ()> + std::marker::Send>`
   = note: required because it appears within the type `Box<dyn futures::Future<Output = ()> + std::marker::Send>`
   = note: required because it appears within the type `Pin<Box<dyn futures::Future<Output = ()> + std::marker::Send>>`
   = note: required because of the requirements on the impl of `Sync` for `FuturesUnordered<Pin<Box<dyn futures::Future<Output = ()> + std::marker::Send>>>`
   = note: required because of the requirements on the impl of `std::marker::Send` for `&FuturesUnordered<Pin<Box<dyn futures::Future<Output = ()> + std::marker::Send>>>`
   = note: required because it appears within the type `[static generator@src/main.rs:16:25: 27:6 _]`
   = note: required because it appears within the type `from_generator::GenFuture<[static generator@src/main.rs:16:25: 27:6 _]>`
   = note: required because it appears within the type `impl futures::Future`

这看起来像是“递归地”推到一个无序的未来(我猜不是真的，但是你还能称它为什么？)不起作用，但我不确定为什么。此错误表示不满足盒的某些< code>Sync特征要求

use std::error::Error;
use tokio::sync::mpsc::{self, Receiver, Sender};
use futures::stream::futures_unordered::FuturesUnordered;
use futures::FutureExt;

#[tokio::main]
pub async fn main() -> Result<(), Box<dyn Error>> {
    let mut futures = FuturesUnordered::new();
    let (tx, rx) = mpsc::channel(32);
    
    tokio::spawn( foo(tx) );    // Only the receiver is relevant; its transmitter is
                                // elsewhere, occasionally sending data.
    futures.push((async {                               // <--- NOTE: futures.push()
        loop {
            match rx.recv().await {
                Some(data) => {
                    futures.push((async move {          // <--- NOTE: nested futures.push()
                        let _ = data; // TODO: replace with code that processes 'data'
                    }).boxed());
                },
                None => {}
            }
        }
    }).boxed());
    
    while let Some(_) = futures.next().await {}

    Ok(())
}

Answer 1

当您使用 boxed 方法将异步块创建的未来装箱时，您正在尝试将其强制发送到 dyn Future Send：

pub fn boxed<'a>(
    self
) -> Pin<Box<dyn Future<Output = Self::Output> + 'a + Send>>

但是，创建的未来不是“发送”。为什么？因为在它里面，你试图推动期货无序，借用它：

pub fn push(&self, future: Fut)

这意味着async块捕获一个

对于发送的引用，类型还必须是Sync：

impl<'_, T> Send for &'_ T where
    T: Sync

为了使< code > futures ordered 成为< code>Sync，存储的期货也必须是< code>Sync:

impl<Fut: Sync> Sync for FuturesUnordered<Fut> {}

但是，box返回的未来不一定是Sync：

pub fn boxed<'a>(
    self
) -> Pin<Box<dyn Future<Output = Self::Output> + 'a + Send>>

这意味着异步生成器不是< code>Send，所以您不能将它强制为< code>dyn Future Send，并且您会得到一个令人困惑的错误消息。

解决方案是添加一个绑定到未来的同步，并手动添加框：：pin：

type BoxedFuture = Pin<Box<dyn Future<Output = ()> + Send + Sync>>;

let mut futures = FuturesUnordered::<BoxedFuture>::new();

futures.push(Box::pin(async {
    loop {
        match rx.recv().await {
            Some(data) => {
                futures.push(Box::pin(async move {
                    let _ = data;
                }));
            }
            None => {}
        }
    }
}));

但是，您将遇到一系列借款问题。更好的解决方案是使用tokio::选择！而不是外部推送，正如Michael的回答所解释的那样。

Answer 2

我将把低级错误留给另一个答案，但是我认为解决这个高级问题的更惯用的方法是将< code > futures ordered 与类似于< code>tokio::select！如下:

use tokio::sync::mpsc;
use futures::stream::FuturesUnordered;
use futures::StreamExt;

#[tokio::main]
pub async fn main() {
    let mut futures = FuturesUnordered::new();
    let (tx, mut rx) = mpsc::channel(32);
    
    //turn foo into something more concrete
    tokio::spawn(async move {
        let _ = tx.send(42i32).await;
    });

    loop {
        tokio::select! {
            Some(data) = rx.recv() => {
                futures.push(async move {
                    data.to_string()
                });
            },
            Some(result) = futures.next() => {
                println!("{}", result)
            },
            else => break,
        }
    }
}

您可以在此处阅读有关选择宏的更多信息：https://tokio.rs/tokio/tutorial/select