Repository: tokio-rs/tokio
File: tokio/tests/rt_metrics.rs (Rust)

Code before:

    assert_eq!(0, rt.metrics().blocking_queue_depth());
}

#[test]
fn num_active_tasks() {
    let rt = current_thread();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.num_active_tasks());
    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.num_active_tasks());
    }))
    .unwrap();

    assert_eq!(0, rt.metrics().num_active_tasks());

    let rt = threaded();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.num_active_tasks());
    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.num_active_tasks());
    }))
    .unwrap();

    // this is too flaky on multiple runtime threads due to relaxed orderings
    // assert_eq!(0, metrics.num_active_tasks());
}

#[test]
fn spawned_tasks_count() {
    let rt = current_thread();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.spawned_tasks_count());

    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.spawned_tasks_count());
    }))
    .unwrap();

    assert_eq!(1, rt.metrics().spawned_tasks_count());

    let rt = threaded();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.spawned_tasks_count());

    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.spawned_tasks_count());
    }))
    .unwrap();

    assert_eq!(1, rt.metrics().spawned_tasks_count());

Review comment (suggestion):
It's not because the counter uses relaxed orderings. Rather, it's because we call `wake_join()` before `self.release()` here:

https://github.com/tokio-rs/tokio/blob/6fcd9c02176bf3cd570bc7de88edaa3b95ea480a/tokio/src/runtime/task/harness.rs#L324-L333

The decrement happens somewhere inside `self.release()`.

I'm happy to merge this without the assert, but please fix the comment.

Code after:
        let _unused = ready_cloned.lock().unwrap();
    };

    let h1 = rt.spawn_blocking(wait_until_ready.clone());
    let h2 = rt.spawn_blocking(wait_until_ready);
    assert!(rt.metrics().blocking_queue_depth() > 0);

    drop(guard);

    let _ = rt.block_on(h1);
    let _ = rt.block_on(h2);

    assert_eq!(0, rt.metrics().blocking_queue_depth());
}

#[test]
fn num_active_tasks() {
    let rt = current_thread();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.num_active_tasks());
    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.num_active_tasks());
    }))
    .unwrap();

    assert_eq!(0, rt.metrics().num_active_tasks());

    let rt = threaded();
    let metrics = rt.metrics();
    assert_eq!(0, metrics.num_active_tasks());
    rt.block_on(rt.spawn(async move {
        assert_eq!(1, metrics.num_active_tasks());
    }))
    .unwrap();

    // try for 10 seconds to see if this eventually succeeds.
    // wake_join() is called before the task is released, so in multithreaded
    // code, this means we sometimes exit the block_on before the counter decrements.
    for _ in 0..100 {
        if rt.metrics().num_active_tasks() == 0 {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    assert_eq!(0, rt.metrics().num_active_tasks());
}

#[test]
fn spawned_tasks_count() {
    let rt = current_thread();
    let metrics = rt.metrics();