concurrency - Rust 异步等待 : check if any future in a list resolves to true concurrently?

Question

我正在尝试在 Rust async-await(即将稳定)中同时(而不是按顺序)运行 futures 列表，直到它们中的任何一个解析为 true .

想象一下有一个 Vec<File> ，以及为每个文件运行的 future，产生 bool (可能是无序的)。这是一个简单的顺序实现。

async fn my_function(files: Vec<File>) -> bool {
    // Run the future on each file, return early if we received true
    for file in files {
        if long_future(file).await {
            return true;
        }
    }

    false
}

async fn long_future(file: File) -> bool {
    // Some long-running task here...
}

这可行，但我想同时运行其中一些 future 以加快该过程。我遇到了 buffer_unordered() (在 Stream 上)，但无法弄清楚如何实现这一点。

据我了解，类似于 join 假设您提供了多线程池，也可以用于并发运行 futures。但我不知道如何在这里有效地使用它。

我尝试了这样的事情，但无法让它工作:

let any_true = futures::stream::iter(files)
    .buffer_unordered(4) // Run up to 4 concurrently
    .map(|file| long_future(file).await)
    .filter(|stop| stop) // Only propagate true values
    .next() // Return early on first true
    .is_some();

除此之外，我正在寻找类似 any 的内容如迭代器中使用的，替换 if 语句或 filter().next().is_some()组合。

我该如何解决这个问题？

Answer 1

我认为您应该能够使用select_ok，正如某些人提到的那样。一个示例，其中我用一堆 u32 替换了文件以进行说明:

use futures::future::FutureExt;

async fn long_future(file: u32) -> bool {
    true
}

async fn handle_file(file: u32) -> Result<(), ()> {
    let should_stop = long_future(file).await;
    // Would be better if there were something more descriptive here
    if should_stop {
        Ok(())
    } else {
        Err(())
    }
}

async fn tims_answer(files: Vec<u32>) -> bool {
    let waits = files.into_iter().map(|f| handle_file(f).boxed());

    let any_true = futures::future::select_ok(waits).await.is_ok();

    any_true
}