Fairness Problem

[elbaro]

#[tokio::main]
async fn main() {
    let (tx, rx) = flume::bounded::<i32>(1_000_000);
    let task1 = async move {
        let rate_limiter = governor::RateLimiter::direct(governor::Quota::per_second(
            std::num::NonZeroU32::new(10_000_000).unwrap(),
        ));
        let jitter = governor::Jitter::up_to(std::time::Duration::from_millis(10));
        let mut n = 0;

        loop {
            n += 1;
            // tokio::task::yield_now().await;
            rate_limiter.until_ready_with_jitter(jitter).await;
            tx.send(n).unwrap();
        }
    };

    let task2 = async move {
        println!("task2 started"); // <- does not run

        while let Ok(item) = rx.recv_async().await {
            println!("received {}", item);
        }
    };

    futures::future::join(task1, task2).await;
}

task1 generates messages and task2 receives.
Although task1 loop has an await point, task2 does not even start (does not print "task2 started").
task2 starts running only after I add tokio::task::yield_now().await or change tx.send to tx.send_async.

I guess rate_limiter.xx.await may not yield to avoid the overhead, as long as there is work to do in the current task.
But this is a multi-threaded runtime and there are many idle cores.
Why is task2 conditionally run waiting for some event from task1?

[elbaro]

Found the answer in the doc of tokio::join: "concurrently but not in parallel"

[Darksonn]

Well, it is also a symptom of the fact that governor uses the futures-timer crate rather than the timers in tokio::time. Had it used tokio::time, then it would have yielded due to Tokio's coop system.