Add unstable on_thread_park_id() to runtime Builder (for stuck task watchdog)

[theron-eg]

Motivation

Fixes #6353. I'm looking for a way to detect when a task is not yielding back to tokio properly. Currently such a task can cause other random tasks to not get run. We had an incident where a buggy task went into a busy loop and put a service into a zombie state: it still was responding to most requests, but some background tasks were not running as expected.

If there is an existing way to detect a stuck task, I'd be delighted to be enlightened.

Solution

Add a new method on_thread_park_id() that includes the worker id being parked. This allows a watchdog process to determine which workers are not parked and also not polling (available already from RuntimeMetrics::worker_poll_count()). The watchdog code would look something like this:

fn main() {
    const WORKERS: usize = 4;
    const UNPARKED: AtomicBool = AtomicBool::new(false);
    static PARKED: [AtomicBool; WORKERS] = [UNPARKED; WORKERS];

    let runtime = runtime::Builder::new_multi_thread()
        .worker_threads(WORKERS)
        .on_thread_park_id(|id| PARKED[id].store(true, Ordering::Release))
        .on_thread_unpark_id(|id| PARKED[id].store(false, Ordering::Release))
        .build()
        .unwrap();

    let metrics = runtime.handle().metrics();
    thread::spawn(move || {
        let mut stuck_secs = [0; WORKERS];
        let mut prev_poll_counts = [None; WORKERS];
        loop {
            thread::sleep(time::Duration::from_secs(1));
            for ii in 0..WORKERS {
                if PARKED[ii].load(Ordering::Acquire) {
                    prev_poll_counts[ii] = None;
                } else {
                    let poll_count = metrics.worker_poll_count(ii);
                    if Some(poll_count) == prev_poll_counts[ii] {
                        stuck_secs[ii] += 1;
                        println!("*** worker {} is stuck >= {} secs ***", ii, stuck_secs[ii]);
                    } else {
                        prev_poll_counts[ii] = Some(poll_count);
                        stuck_secs[ii] = 0;
                    }
                }
            }
        }
    });

    runtime.spawn(async { loop {} });
    runtime.block_on(async {
        loop {
            tokio::task::yield_now().await
        }
    });
}

Rather than printing, a real watchdog implementation could call std::process::abort() if the task stays stuck past some duration (e.g. 10 seconds).

[Darksonn]

Maybe a better option would be add a new metric for this. It could count the number of parks/unparks. When the number is odd, the worker is parked. When it is even, the worker is active.

[Darksonn]

Note that you can push an empty commit when you want to trigger a new build.

You will probably need to merge in master to get the build to work. There were some changes to the CI setup.

[theron-eg]

You will probably need to merge in master to get the build to work. There were some changes to the CI setup.

Ah thanks, I was puzzled what was going wrong. I hope to get back to making more progress on this in the next couple of days.

Maybe a better option would be add a new metric for this. It could count the number of parks/unparks. When the number is odd, the worker is parked. When it is even, the worker is active.

It did occur to me to add a metric for "is currently parked" or something similar but it seemed more useful to have the generic callback mechanism (with the proper worker id). With that it's possible to for anyone to collect their own set of metrics, instead of relying on WorkerMetrics having exactly the right set of info (and incurring the overhead to accurately / atomically count things that 99.9% of people don't care about). But if you'd prefer a change to WorkerMetrics instead I'm happy to go down that route. Let me at least get this ready for official review and then I'm happy to take feedback.

[Darksonn]

I think a new worker metric makes sense.