Livelock in poll_proceed when using a futures 0.1 FuturesUnordered on Tokio 0.2.16

[krallin]

Version

• Tokio 0.2.16
• Futures 0.1 for FuturesUnordered

Platform

Linux, OSX.

Subcrates

sync, coop

Description

We are seeing a livelock in poll_proceed when executing a futures 0.1 stream built using poll_fn and FuturesUnordered where the futures driven by FuturesUnordered calls Semaphore::acquire.

I have implemented a repro here: https://github.com/krallin/repro-tokio-livelock. To repro, just cargo run. It'll hang forever.

The code is:

use futures::{
    compat::Stream01CompatExt,
    future::{FutureExt, TryFutureExt},
    stream::TryStreamExt,
};
use futures_old::stream::FuturesUnordered; // <-- Futures 0.1
use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    let sem = Semaphore::new(1);
    let sem = &sem;

    let mut futs = FuturesUnordered::new();
    for _ in 0..200 { // If you do e.g. 10 it works
        futs.push(async move {
            let _ = sem.acquire().await;
            Result::<_, ()>::Ok(())
        }.boxed().compat());
    }

    let n = futs.compat().try_collect::<Vec<_>>().await.unwrap().len();
    println!("done: {}", n);
}

The livelock is here:

pub fn poll_proceed(cx: &mut Context<'_>) -> Poll<()> {
    HITS.with(|hits| {
        let n = hits.get();
        if n == UNCONSTRAINED {
            // opted out of budgeting
            Poll::Ready(())
        } else if n == 0 {
           # BUG HERE >> We constantly re-enter this code path.
            cx.waker().wake_by_ref();
            Poll::Pending
        } else {
            hits.set(n.saturating_sub(1));
            Poll::Ready(())
        }
    })
}

We'd love some help understanding troubleshooting this issue. Thanks! Let us know if there is anything we can do to help root-cause further.

[Darksonn]

Are you using old futures because it has FuturesUnordered? Just to confirm, are you aware that futures v0.3 also has an implementation of FuturesUnordered? Additionally, can you link to the poll_proceed function source, so I can see context?

[krallin]

@Darksonn,

Thanks for replying!

Are you using old futures because it has FuturesUnordered? Just to confirm, are you aware that futures v0.3 also has an implementation of FuturesUnordered?

Yep, we're indeed aware that FuturesUnordered in also in Futures 0.3. We simply still have some older code that still uses Futures 0.1 that is being driven by a Tokio 0.2 executor.

The repro I shared is definitely a contrived, reduced example (the actual callsite for this is here: https://github.com/facebookexperimental/eden/blob/master/eden/mononoke/pushrebase/src/lib.rs#L281, and goes through layers before finally getting to FuturesUnordered).

Additionally, can you link to the poll_proceed function source, so I can see context?

Here you go:

tokio/tokio/src/coop.rs

Line 225 in 58ba45a

pub fn poll_proceed(cx: &mut Context<'_>) -> Poll<()> {

[Darksonn]

The issue is almost certainly the new preemption feature (blog post), which makes IO resources return Poll::Pending once a per-task-poll budget has been used up. As you can see in poll_proceed, that IO resource will in this case immediately emit a wakeup and return pending.

This means that since the wakeup is emitted to the FuturesUnordered subscheduler, that sub scheduler decides that since it got a wakeup from that IO resource, it should be polled again immediately. Unfortunately this means that the poll of the Tokio task never completes, as the FuturesUnordered repeatedly polls that task without yielding to Tokio, and this means the budget is not reset.

[jonhoo]

Yeah, this is basically rust-lang/futures-rs#2047, which was fixed in futures-util 0.3 (rust-lang/futures-rs#2049), but the fix wasn't backported to 0.1 :/

[krallin]

@jonhoo thanks — sounds like I should submit this backport then. That said, just to make sure I'm following this properly: this means my FuturesUnordered will poll the underlying future 32 times "unsuccessfully" before actually yielding and polling it successfully, right?

[krallin]

I submitted rust-lang/futures-rs#2122 to backport the fix.

Out of curiosity, is this new cooperative scheduling mechanism configurable? (i.e. can it be turned on or off on a per-runtime basis?) Looking at the code, it looks like it isn't, but I'd love to make sure: we were somewhat lucky that we had a test that happened to catch this particular bug, but it'd be nice to be able to canary this individual mechanism in order to make sure we don't have further instances of it lying around in the codebase 😄

[jonhoo]

@krallin Yes, your observation is (sadly) correct — FuturesUnordered will end up polling the perpetually-pending future 32 times unsuccessfully before yielding. What I really want to happen is for sub-executors (like FuturesUnordered) to have a way to check if the budget has been expended, and if so, to stop polling more pending futures. It'd be a sort of heuristic like:

If two futures in a row that were marked as ready return Pending when you poll them, check the budget.

The tricky part of this is how we expose that functionality without requiring that the implementor having to depend on tokio. For many crates, that dependency wouldn't be too onerous, but for something like futures-util, it probably shouldn't. Arguably what's needed is for the coop "building blocks" to be provided by some low-level, separate library that everyone can then hook into. Though that comes with the cost of adding external dependencies, which has its own set of problems.

There isn't currently a way to turn off coop, though in theory it should be pretty straightforward. Essentially you'd want to make these calls not wrap the inner closure with coop::budget if budgeting is disabled. Or we could do it with a feature I suppose? @carllerche may have thoughts.

[aclysma]

+1 for being able to turn off coop. I know some people are going to have different opinions on this, but I explicitly don't want fairness in tasks. (For context, my usage is loading assets for games, and possibly in the future executing jobs that may wait on locks, GPU fences, or each other)

• If I have something long-lived and want fairness, I would use threads. In that case the unit of work dominates the cost of context switching threads and the OS provides stronger guarantees of fairness.
• If I have something short-lived, then it's not going to starve other tasks. So if it can make progress, I want it to continue to make progress, finish, and then pick up the next task.
• If I really did want fairness in short lived tasks, then I would prefer to inject my own yields based on heuristics that I choose that can be informed by application-specific logic.

I would probably still be of this opinion, even if I was writing something more "vanilla" like a web service, but I do understand not everyone is going to share my opinion on this! And long term, maybe my usage isn't precisely what tokio is designed for. But maybe it's not too difficult to allow disabling it (or allow modifying the heuristics to an impossibly high value to effectively disable it)

My current workaround is to use tokio 0.2.13. Otherwise my application hangs consistently with even a relatively small number of in-flight asset loads.

[taiki-e]

Fixed in futures 0.1.30