rust - 当任何任务正在运行时，如何阻止异步程序终止？

Question

假设我有这样的东西，

async fn do_update() {
    // here we store it.
    let task = task::spawn(async {
        let duration = Duration::from_millis(10);
        let mut stream = tokio::time::interval(duration);
        stream.tick().await;
        loop {
            println!("Foo");
            stream.tick().await;
        }
    });
    // and here we await it.
    task.await;
}

如果我调用它，比如 do_update().await 循环将永远运行。这就是我想要的。但我不希望它成为唯一运行的东西。我只希望它在任务解决之前阻止终止。

我想要一个事件每 5 秒运行一次而不阻塞程序的其余部分。如果我把它放在程序的末尾，它完全按我的意愿工作。似乎这曾经是通过 shutdown_on_idle 完成的，但出于同样的原因，现在提供的解决方案对我无效 as this comment ,

I cannot implement it like that because I spawn inside of another task.

在任务产生任务的系统中，我如何确定可执行文件仅在没有更多任务运行时才关闭？

Answer 1

我认为您正在寻找结构化并发，特别是全局隐式范围。可悲的是，nobody could find a good solution , 因此暂时放弃了这方面的工作。

与此同时，这里有一个可能的解决方法。调用 spawn_keep_alive 而不是 tokio::spawn 来执行任何应该使程序保持事件状态的任务:

Playground

use parking_lot::Mutex;
use std::{
    future::Future,
    sync::atomic::{AtomicU32, Ordering},
    time::Duration,
};
use tokio::{sync::oneshot, task::JoinHandle};

static KEEPALIVE_COUNT: AtomicU32 = AtomicU32::new(0);
static KEEPALIVE_SENDER: Mutex<Option<oneshot::Sender<()>>> = parking_lot::const_mutex(None);

pub fn spawn_keep_alive<T>(task: T) -> JoinHandle<T::Output>
where
    T: Future + Send + 'static,
    T::Output: Send + 'static,
{
    KEEPALIVE_COUNT.fetch_add(1, Ordering::Relaxed);
    tokio::spawn(async {
        let result = task.await;
        if KEEPALIVE_COUNT.fetch_sub(1, Ordering::Relaxed) == 1 {
            let sender = KEEPALIVE_SENDER.try_lock().unwrap().take().unwrap();
            sender.send(()).unwrap();
        }
        result
    })
}

async fn do_update() {
    let mut stream = tokio::time::interval(Duration::from_millis(100));
    stream.tick().await;
    for _ in 0..10 {
        println!("Foo");
        stream.tick().await;
    }
    spawn_keep_alive(async {
        tokio::time::sleep(Duration::from_millis(1000)).await;
        println!("I'm aliveeee!");
    });
    spawn_keep_alive(async {
        tokio::time::sleep(Duration::from_millis(2000)).await;
        println!("Don't forget about me!");
    });
}

#[tokio::main]
async fn main() {
    let (send, recv) = oneshot::channel();
    *KEEPALIVE_SENDER.try_lock().unwrap() = Some(send);
    spawn_keep_alive(do_update());
    // Wait for all keep-alive tasks to finish
    recv.await.unwrap();
}

(KEEPALIVE_SENDER 可能更有效，但它只使用了两次)。