rt(threaded): cap LIFO slot polls

[carllerche]

As an optimization to improve locality, the multi-threaded scheduler maintains a single slot (LIFO slot). When a task is scheduled, it goes into the LIFO slot. The scheduler will run tasks in the LIFO slot first before checking the local queue.

Ping-ping style workloads where task A notifies task B, which notifies task A again, can cause starvation as these two tasks repeatedly schedule the other in the LIFO slot. #5686, a first attempt at solving this problem, consumes a unit of budget each time a task is scheduled from the LIFO slot. However, at the time of this commit, the scheduler allocates 128 units of budget for each chunk of work. This is relatively high in situations where tasks do not perform many async operations yet have meaningful poll times (even 5-10 microsecond poll times can have an outsized impact on the scheduler).

In an ideal world, the scheduler would adapt to the workload it is executing. However, as a stopgap, this commit limits the number of times the LIFO slot is prioritized per scheduler tick.

Benchmarks

In a benchmark crafted to simulate injecting tasks while the runtime is under load, this change sped things up 30x (68s -> 2s)

This is the benchmark I used to measure this change's improvements. I am still waiting to include it because, even with this change, it causes the scheduler benchmarks to run for a very long time.

    const STALL_DUR: Duration = Duration::from_micros(10);
    const NUM_SPAWN: usize = 1_000;
    static TICKS: AtomicUsize = AtomicUsize::new(0);

    let rt_handle = rt.handle();
    let mut handles = Vec::with_capacity(NUM_SPAWN);
    let stall = Arc::new(AtomicBool::new(true));

    // Spawn some tasks to keep the runtimes busy
    for _ in 0..NUM_WORKERS {
        let stall = stall.clone();
        fn iter(stall: Arc<AtomicBool>) {
            tokio::spawn(async {
                if stall.load(Relaxed) {
                    TICKS.fetch_add(1, Relaxed);
                    let now = Instant::now();

                    while now.elapsed() < STALL_DUR {}

                    iter(stall);
                }
            });
        }
        rt.spawn(async { iter(stall) });
    }

    for _ in 0..NUM_SPAWN {
        handles.push(rt_handle.spawn(async {}));
    }

    rt.block_on(async {
        for handle in handles.drain(..) {
            handle.await.unwrap();
        }
    });
    stall.store(false, Relaxed);
    println!(" TICKS = {}", TICKS.load(Relaxed));

[carllerche]

All other scheduler benchmark results are unchanged (within the margin of error).

[Noah-Kennedy]

Might not be a bad idea to add an internal highly unstable runtime metric for this.

[hawkw]

This makes sense to me! I agree with @Noah-Kennedy's suggestion of adding a new unstable runtime metric to track this behavior, though.

[Darksonn]

Would it not be less error-prone to reset lifo_enabled after calling task.run()? For example, right now you don't reset it in the Err(()) branch. That's probably fine because the worker thread shuts down there, but it is non-obvious.

[carllerche]

Yeah, in the error case, the core is stolen. We cannot reset it either way. That said, it brings up a point that we need to reset it if it is stolen.

[Darksonn]

Maybe just reset before polling a task ...

[carllerche]

Reset before polling a task isn't robust either because we can schedule tasks outside of the context of running a task. e.g. from polling the I/O driver.

[carllerche]

Instead, what I suggest is we add debug_asserts around to ensure the value is correct.