如何在 Tokio 中为 CPU 密集型工作创建专用线程池？

Question

我有一个基于 Tokio 运行时的 Rust 异步服务器。它必须同时处理对延迟敏感的 I/O 绑定请求和繁重的 CPU 绑定请求。

我不想让 CPU 密集型任务独占 Tokio 运行时并饿死 I/O 密集型任务，因此我想将 CPU 密集型任务卸载到一个专用的隔离线程池（隔离是关键在这里，所以一个共享线程池上的spawn_blocking/block_in_place 是不够的）。如何在 Tokio 中创建这样的线程池？

启动两个运行时的幼稚方法会遇到错误：

线程“tokio-runtime-worker”在“无法从运行时内启动运行时”出现恐慌。发生这种情况是因为一个函数（如block_on）在线程被用于驱动异步任务时试图阻塞当前线程。'

use tokio; // 0.2.20

fn main() {
    let mut main_runtime = tokio::runtime::Runtime::new().unwrap();
    let cpu_pool = tokio::runtime::Builder::new().threaded_scheduler().build().unwrap();
    let cpu_pool = cpu_pool.handle().clone(); // this is the fix/workaround!

    main_runtime.block_on(main_runtime.spawn(async move {
        cpu_pool.spawn(async {}).await
    }))
    .unwrap().unwrap();
}

Tokio 可以允许两个独立的运行时吗？有没有更好的方法在 Tokio 中创建一个隔离的 CPU 池？

Answer 1

虽然 Tokio 已经有一个线程池，但 documentation of Tokio advises：

如果您的代码受 CPU 限制并且您希望限制用于运行它的线程数，您应该在另一个线程池上运行它，例如 rayon。当rayon 任务完成时，您可以使用oneshot 通道将结果发送回Tokio。

所以，如果你想创建一个线程池来大量使用 CPU，一个好方法是使用像 Rayon 这样的 crate 并将结果发送回 Tokio 任务。

Answer 2

启动 Tokio 运行时已经创建了一个线程池。相关选项是

• Builder::core_threads（0.2.20中默认是CPU核数）
• Builder::max_threads（0.2.20 中默认为 512）

粗略地说，core_threads 控制将使用多少线程来处理异步代码。 max_threads - core_threads 是多少线程将用于阻塞工作（强调我的）：

否则，由于core_threads 始终处于活动状态，它会将其他线程（例如用于阻塞注释）限制为max_threads - core_threads。

您还可以通过tokio::main 属性指定这些选项。

然后您可以使用以下任一方式注释阻塞代码：

• task::spawn_blocking
• task::block_in_place

另见：

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

spawn_blocking 可以轻松获取唯一运行时中的所有可用线程，从而迫使其他未来等待它们

在这种情况下，您可以使用Semaphore 之类的技术来限制最大并行度。

Answer 3

Tokio 的错误消息was misleading。问题是由于 Runtime 对象在异步上下文中被丢弃。

解决方法是使用Handle，而不是直接使用Runtime，在另一个运行时生成任务。

fn main() {
    let mut main_runtime = tokio::runtime::Runtime::new().unwrap();
    let cpu_pool = tokio::runtime::Builder::new().threaded_scheduler().build().unwrap();

    // this is the fix/workaround:
    let cpu_pool = cpu_pool.handle().clone(); 

    main_runtime.block_on(main_runtime.spawn(async move {
        cpu_pool.spawn(async {}).await
    }))
    .unwrap().unwrap();
}