有什么办法可以关闭`tokio :: runtime :: current_thread :: Runtime`吗? [英] Is there any way to shutdown `tokio::runtime::current_thread::Runtime`?

问题描述

我正在使用 tokio::runtime::current_thread::Runtime 我希望能够运行将来并在同一线程中停止反应堆.页面上的示例未显示如何停止运行时.有什么办法可以做到吗?

I'm using tokio::runtime::current_thread::Runtime and I want to able to run a future and stop the reactor in the same thread. The example on the page doesn't show how to stop the runtime. Is there any way I can do that?

推荐答案

当将来完成时，运行时将自动关闭:

The runtime will automatically shut down when when the future is complete:

use std::time::Duration;
use tokio::time; // 0.2.21

#[tokio::main]
async fn main() {
    time::delay_for(Duration::from_secs(2)).await;
    eprintln!("future complete");
}

请参见如何在稳定的Rust中同步返回在异步Future中计算出的值?运行时.

如果您需要取消未来，则可以创建将导致未来poll成功的内容.我可能会使用频道和select:

If you need to cancel a future, you can create something that will cause future polls to succeed. I'd probably use channels and select:

use futures::{channel::oneshot, future, FutureExt}; // 0.3.5
use std::time::Duration;
use tokio::{task, time}; // 0.2.21

#[tokio::main]
async fn main() {
    let future = async {
        time::delay_for(Duration::from_secs(3600)).await;
        eprintln!("future complete");
    };

    let (cancel_tx, cancel_rx) = oneshot::channel();

    let another_task = task::spawn(async {
        eprintln!("Another task started");
        time::delay_for(Duration::from_secs(2)).await;
        eprintln!("Another task canceling the future");
        cancel_tx.send(()).expect("Unable to cancel");
        eprintln!("Another task exiting");
    });

    future::select(future.boxed(), cancel_rx).await;

    another_task.await.expect("The other task panicked");
}

这是一个替代的手动解决方案，它非常简单，蛮力，而且性能可能不太好:

Here's an alternate manual solution that's very simple, brute force, and probably not-very-performant:

use pin_project::pin_project; // 0.4.17
use std::{
    future::Future,
    pin::Pin,
    sync::{Arc, Mutex},
    task::{self, Context, Poll},
    thread,
    time::Duration,
};
use tokio::time; // 0.2.21 

#[tokio::main]
async fn main() {
    let future = async {
        time::delay_for(Duration::from_secs(3600)).await;
        eprintln!("future complete");
    };

    let (future, cancel) = Cancelable::new(future);

    let another_thread = thread::spawn(|| {
        eprintln!("Another thread started");
        thread::sleep(Duration::from_secs(2));
        eprintln!("Another thread canceling the future");
        cancel();
        eprintln!("Another thread exiting");
    });

    future.await;

    another_thread.join().expect("The other thread panicked");
}

#[pin_project]
#[derive(Debug)]
struct Cancelable<F> {
    #[pin]
    inner: F,
    info: Arc<Mutex<CancelInfo>>,
}

#[derive(Debug, Default)]
struct CancelInfo {
    cancelled: bool,
    task: Option<task::Waker>,
}

impl<F> Cancelable<F> {
    fn new(inner: F) -> (Self, impl FnOnce()) {
        let info = Arc::new(Mutex::new(CancelInfo::default()));
        let cancel = {
            let info = info.clone();
            move || {
                let mut info = info.lock().unwrap();
                info.cancelled = true;
                if let Some(waker) = info.task.take() {
                    waker.wake();
                }
            }
        };
        let me = Cancelable { inner, info };
        (me, cancel)
    }
}

impl<F> Future for Cancelable<F>
where
    F: Future<Output = ()>,
{
    type Output = ();

    fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        let mut info = this.info.lock().unwrap();

        if info.cancelled {
            Poll::Ready(())
        } else {
            let r = this.inner.poll(ctx);

            if r.is_pending() {
                info.task = Some(ctx.waker().clone());
            }

            r
        }
    }
}

另请参阅:

• 什么时候可以安全地将会员价值移出固定的未来?

这篇关于有什么办法可以关闭`tokio :: runtime :: current_thread :: Runtime`吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！