\"没有信号驱动程序运行，必须从 Tokio 运行时的上下文中调用\" 尽管在 Runtime::block_on 中运行

Question

我正在使用 Tokio 编写命令行应用程序，它通过侦听键盘中断事件（即 ctrl + c）来控制其生命周期；但是，与此同时，它还必须监视生成的其他任务，并可能在任何任务崩溃或遇到错误时启动提前关闭。为此，我将 tokio::select 包装在 while 循环中，一旦应用程序至少有机会安全关闭，该循环就会终止。

但是，一旦 select 块轮询 tokio::signal::ctrl_c 返回的未来，主线程就会出现恐慌并显示以下消息：

thread 'main' panicked at 'there is no signal driver running, must be called from the context of Tokio runtime'

...这令人困惑，因为这一切都是在 Runtime::block_on 调用中完成的。我还没有发布此应用程序（但），但可以使用以下代码重现该问题：

use tokio::runtime::Builder;
use tokio::signal;
use tokio::sync::watch;
use tokio::task::JoinSet;

fn main() {
    let runtime = Builder::new_multi_thread().worker_threads(2).build().unwrap();
    runtime.block_on(async {
        let _rt_guard = runtime.enter();
        let (ping_tx, mut ping_rx) = watch::channel(0u32);
        let (pong_tx, mut pong_rx) = watch::channel(0u32);
        let mut tasks = JoinSet::new();
        
        let ping = tasks.spawn(async move {
            let mut val = 0u32;
            ping_tx.send(val).unwrap();
            
            while val < 10u32 {
                pong_rx.changed().await.unwrap();
                val = *pong_rx.borrow();
                ping_tx.send(val + 1).unwrap();
                println!("ping! {}", val + 1);
            }
        });
        
        let pong = tasks.spawn(async move {
            let mut val = 0u32;
            
            while val < 10u32 {
                ping_rx.changed().await.unwrap();
                val = *ping_rx.borrow();
                pong_tx.send(val + 1).unwrap();
                println!("pong! {}", val + 1);
            }
        });
        
        let mut interrupt = Box::pin(signal::ctrl_c());
        let mut interrupt_read = false;
        
        while !interrupt_read && !tasks.is_empty() {
            tokio::select! {
                biased;
                _ = &mut interrupt, if !interrupt_read => {
                    ping.abort();
                    pong.abort();
                    interrupt_read = true;
                },
                _ = tasks.join_next() => {}
            }
        }
    });
}

Rust Playground

这个例子有点做作，但重要的部分是：

• 我有意使用Runtime::block_on()而不是tokio::main，因为我想在运行时控制运行时线程的数量。
  • 不过，奇怪的是，如果重写为使用tokio::main，此示例仍然有效。
• 我添加了 let _rt_guard = runtime.enter() 以确保设置了运行时上下文，但它的存在与否似乎没有什么不同。

Answer 1

我在写完这个问题的时候发现了这个问题的答案，但是由于我无法通过搜索错误消息找到答案，所以我将在这里分享答案。

正如我所指出的，如果从带有 tokio::main 注释的函数中运行，我给出的示例将起作用。查看docs，main宏扩展为：

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Hello world");
        })
}

我给出的示例没有在构建器上调用 enable_all()（或者更具体地说，enable_io()）。将此调用添加到构建器链可阻止恐慌的发生。我进行的 runtime.enter() 调用也可以删除，因为它实际上没有做任何事情。

更新后的示例（其中一个线程仍然因 Receiver 句柄被删除而出现恐慌，但 ctrl_c await 不再恐慌，这是关键）：

use tokio::runtime::Builder;
use tokio::signal;
use tokio::sync::watch;
use tokio::task::JoinSet;

fn main() {
    Builder::new_multi_thread()
        .worker_threads(2)
        .enable_io()
        .build()
        .unwrap()
        .block_on(async {
            let (ping_tx, mut ping_rx) = watch::channel(0u32);
            let (pong_tx, mut pong_rx) = watch::channel(0u32);
            let mut tasks = JoinSet::new();
            
            let ping = tasks.spawn(async move {
                let mut val = 0u32;
                ping_tx.send(val).unwrap();
                
                while val < 10u32 {
                    pong_rx.changed().await.unwrap();
                    val = *pong_rx.borrow();
                    ping_tx.send(val + 1).unwrap();
                    println!("ping! {}", val + 1);
                }
            });
            
            let pong = tasks.spawn(async move {
                let mut val = 0u32;
                
                while val < 10u32 {
                    ping_rx.changed().await.unwrap();
                    val = *ping_rx.borrow();
                    pong_tx.send(val + 1).unwrap();
                    println!("pong! {}", val + 1);
                }
            });
            
            let mut interrupt = Box::pin(signal::ctrl_c());
            let mut interrupt_read = false;
            
            while !interrupt_read && !tasks.is_empty() {
                tokio::select! {
                    biased;
                    _ = &mut interrupt, if !interrupt_read => {
                        ping.abort();
                        pong.abort();
                        interrupt_read = true;
                    },
                    _ = tasks.join_next() => {}
                }
            }
        });
}

Rust Playground