如何在 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 已经有一个线程池，但Tokio的文档建议：

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

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

Answer 2

Tokio 的错误消息具有误导性。该问题是由于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();
}

Answer 3

启动 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

也可以看看：

• 在 future-rs 中封装阻塞 I/O 的最佳方法是什么？

spawn_blocking 可以轻松地获取在一个且唯一的运行时中可用的所有线程，迫使其他期货等待它们

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