asynchronous - 如何在启动异步操作和在循环中等待其结果之间有一个宽限期？

Question

我有一个循环:

let grace = 2usize;
for i in 0..100 {
    if i % 10 == 0 {
        expensive_function()
    } else {
        cheap_function()
    }
}

目标是当它到达 expensive_function() 时，它异步运行并允许 grace直到等待的进一步迭代次数 expensive_function() .
如 expensive_function()在迭代 10 时触发，然后它可以在需要等待 expensive_function() 之前运行迭代 11 和 12在迭代 10 上运行以完成继续。
我怎么能这样做？
就我而言 expensive_function()有效地:

fn expensive_function(&b) -> Vec<_> {
    return b.iter().map(|a| a.inner_expensive_function()).collect();
}

因此，我计划在此功能中使用多线程。

Answer 1

当您开始昂贵的计算时，将产生的 future 存储在一个变量中，以及等待结果的截止时间。在这里，我使用 Option元组:

use std::{thread, time::Duration};
use tokio::task; // 0.2.21, features = ["full"]

#[tokio::main]
async fn main() {
    let grace_period = 2usize;
    let mut pending = None;

    for i in 0..50 {
        if i % 10 == 0 {
            assert!(pending.is_none(), "Already had pending work");

            let future = expensive_function(i);
            let deadline = i + grace_period;
            pending = Some((deadline, future));
        } else {
            cheap_function(i);
        }

        if let Some((deadline, future)) = pending.take() {
            if i == deadline {
                future.await.unwrap();
            } else {
                pending = Some((deadline, future));
            }
        }
    }
}

fn expensive_function(n: usize) -> task::JoinHandle<()> {
    task::spawn_blocking(move || {
        println!("expensive_function {} start", n);

        thread::sleep(Duration::from_millis(500));

        println!("expensive_function {} done", n);
    })
}

fn cheap_function(n: usize) {
    println!("cheap_function {}", n);
    thread::sleep(Duration::from_millis(1));
}

这将生成输出

cheap_function 1
expensive_function 0 start
cheap_function 2
expensive_function 0 done
cheap_function 3
cheap_function 4
cheap_function 5

由于您没有提供 expensive_function 的定义和 cheap_function ，我已经提供了合适的。
这里一件棘手的事情是我需要添加 sleep调用 cheap_function .没有它，我的操作系统永远不会安排昂贵的线程，直到轮询它为止，从而有效地删除了任何并行工作。在更大的程序中，操作系统很可能只是因为 cheap_function 将完成更多的工作而调度线程。 .您也可以使用 thread::yield_now 达到同样的效果。
也可以看看:

How to create a dedicated threadpool for CPU-intensive work in Tokio?

How do I synchronously return a value calculated in an asynchronous Future in stable Rust?

What is the best approach to encapsulate blocking I/O in future-rs?