Higher Scheduling Latency When Spawning Task on Separate Runtime Thread

[BlackLuny]

I’ve created two Tokio runtimes: rt1 and rt2.

In my code, a task is spawned from rt1 onto rt2 like this:

let mut h2 = rt2.spawn(
    task_monitor.instrument(async move {
        ...
    })
);

However, I’ve noticed a performance issue:

After running multiple tests, I observed that if rt2 is just a handle to the current runtime (i.e., let rt2 = tokio::runtime::Handle::current();, effectively using rt1), the scheduling latency measured by tokio-metrics is significantly lower.

In contrast, when rt2 is a separate current-thread runtime running on a different system thread or multi-thread runtime, the scheduling latency increases noticeably.

Is this expected behavior? Or could this indicate an inefficiency in how Tokio handles cross-runtime task scheduling?

Any insights would be appreciated!

metric:

[BlackLuny]

The tokio-metric metric is mean_scheduled_duration.

[Darksonn]

Tasks sent to external runtimes go through the global queue, whereas it goes to the local queue when called on a runtime thread.

[BlackLuny]

I know what you mean.But tokio-metric measures the mean schedule duration have skipped the first poll.So It is the local queue schedule latency?

[BlackLuny]

Update / Further Investigation:

I’ve made some additional observations.

In the production environment, rt2 is a current-thread runtime. However, I noticed that the number of remote schedules is significantly higher than local schedules, which seems unusual.

• Local schedule count: only around 1,300 in total
  

• Remote schedule count: around 120,000 in total
  

To investigate further, I created a minimal demo to try and reproduce the issue. However, in the demo environment, everything behaves as expected: the number of local schedules is much higher than remote schedules.

This discrepancy between the demo and production environment is puzzling. Any ideas on what might be causing this behavior?

use anyhow::Result;
use tokio::io::AsyncWriteExt;

fn main() -> Result<()> {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    let handle = rt.handle().clone();
    let (tx, rx) = tokio::sync::oneshot::channel::<()>();
    let thread = std::thread::spawn(move || {
        let _ = rt.block_on(rx);
    });
    let main_rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()?;
    // spawn some tasks

    main_rt.block_on(async move {
        tokio::spawn(async move {
            let mut handlers = Vec::new();
            for i in 0..10 {
                let mut new_tcp = tokio::net::TcpStream::connect("127.0.0.1:40048")
                    .await
                    .unwrap();
                let h = handle.spawn(async move {
                    loop {
                        println!("task {} is running", i);
                        // simulate a data transfer with the tcp established by main_rt
                        let _ = new_tcp.write_all(&[0; 1024]).await;
                        tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
                        println!(
                            "remote_schedule_count: {}",
                            tokio::runtime::Handle::current()
                                .metrics()
                                .remote_schedule_count()
                        );
                        for i in 0..tokio::runtime::Handle::current().metrics().num_workers() {
                            println!(
                                "local_schedule_count worker {}: {}",
                                i,
                                tokio::runtime::Handle::current()
                                    .metrics()
                                    .worker_local_schedule_count(i)
                            );
                        }
                    }
                    return Ok::<_, ()>(new_tcp);
                });
                handlers.push(h);
            }
            for h in handlers {
                let _ = h.await;
            }
        });
        tokio::time::sleep(tokio::time::Duration::from_secs(100)).await;
    });

    let _ = thread.join();
    Ok(())
}

output:

task 0 is running
task 1 is running
task 2 is running
task 3 is running
task 4 is running
task 5 is running
task 6 is running
task 7 is running
task 8 is running
task 9 is running
remote_schedule_count: 10
local_schedule_count worker 0: 5
task 0 is running
remote_schedule_count: 10
local_schedule_count worker 0: 5
task 1 is running
remote_schedule_count: 10
local_schedule_count worker 0: 5
task 2 is running
remote_schedule_count: 10
local_schedule_count worker 0: 5
task 3 is running
remote_schedule_count: 10
local_schedule_count worker 0: 5
task 4 is running
remote_schedule_count: 10
local_schedule_count worker 0: 10
task 5 is running
remote_schedule_count: 10
local_schedule_count worker 0: 10
task 6 is running
remote_schedule_count: 10
local_schedule_count worker 0: 10
task 7 is running
remote_schedule_count: 10
local_schedule_count worker 0: 10
task 8 is running
remote_schedule_count: 10
local_schedule_count worker 0: 10
task 9 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 0 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 1 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 2 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 3 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 4 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 5 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 6 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 7 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 8 is running
remote_schedule_count: 10
local_schedule_count worker 0: 20
task 9 is running