tokio之如何观测Rust异步任务的调度

newbmiao

发布于 2024-02-26 09:45:40

1410

发布于 2024-02-26 09:45:40

文章被收录于专栏：学点Rust学点Rust

通过之前的《Rust 异步编程之 Future 初探》我们知道Rust的异步是以task的调度来构建的。task作为抽象在语言层面的调度单元。

那如果想要观测其的调度执行有没有办法呢？

有的！异步runtime库tokio就有个tokio-console可以实现对异步调度的观测和分析，其对了解调度的机制和性能分析都很有帮助。

环境设置

其调度的观测是需要依赖程序段添加 console-subscriber 来上报runtime的调度信息，然后由命令行程序tokio-console进行数据的统计展示。

这个功能还属于 unstable，需要引入时做一些设置。

具体来说，需要在引入tokio时启用tracing, 如

[dependencies]
console-subscriber = "0.2.0"
tokio = { version = "1.35.1", features = ["full", "tracing"] }

异步代码main中也需要初始化console_subscriber

#[tokio::main]
async fn main() {
    console_subscriber::init();
    // ...
}

代码运行时需要加上编译参数：RUSTFLAGS="--cfg tokio_unstable" cargo run

也可以在项目根目录用.cargo/config.toml全局配置, 如

[build]
rustflags = ["--cfg", "tokio_unstable"]

这样同时运行tokio-console就能观测异步任务的调度了。

（cargo install tokio-console可以安装）

代码改造

以之前《Rust 并发控制之 Semaphore-两线程交替打印》代码来观测为例

修改部分详见代码注释:

use std::{sync::Arc, time::Duration};
use tokio::{sync::Semaphore, task, time::sleep};

#[tokio::main]
async fn main() {
    // 注意. 初始化tracing收集
    console_subscriber::init();
    // 线程1的令牌桶1初始一个令牌，可以先打印1
    let semaphore = Arc::new(Semaphore::new(1));
    let cnt = 3;
    let semaphore2 = semaphore.clone();

    // 线程2的令牌桶2初始没有令牌，直到1打印后增加令牌
    let semaphore_wait = Arc::new(Semaphore::new(0));
    let semaphore_wait2 = semaphore_wait.clone();

    // 注意. 使用task::Builder来增加task名字，否则等同tokio::spawn
    let t1 = task::Builder::default()
        .name("t1")
        .spawn(async move {
            for i in 0..cnt {
                let permit = semaphore.acquire().await.unwrap();
                print!("1 ");
                 // 注意. 增加等待时间，便于观测
                sleep(Duration::from_secs(i)).await;
                // 消耗令牌，不放回令牌桶1
                permit.forget();
                // 令牌桶2增加令牌，可以打印2
                semaphore_wait2.add_permits(1);
            }
        })
        .unwrap();

    let t2 = task::Builder::default()
        .name("t2")
        .spawn(async move {
            for i in 0..cnt {
                let permit = semaphore_wait.acquire().await.unwrap();
                print!("2 ");
                // 注意. 增加等待时间，便于观测
                sleep(Duration::from_secs(i)).await;
                // 消耗令牌，不放回令牌桶2
                permit.forget();
                // 令牌桶1增加令牌，可以打印1
                semaphore2.add_permits(1);
            }
        })
        .unwrap();

    tokio::try_join!(t1, t2).unwrap();
}