在 async 里阻塞：一个任务卡住整个 worker

use std::time::{Duration, Instant};
use tokio::runtime::{Builder, Runtime};

// A task that wants to wake every 10ms; returns the largest gap it saw.
async fn ticker(total: Duration) -> Duration {
  let start = Instant::now();
  let mut last = Instant::now();
  let mut worst = Duration::ZERO;
  while start.elapsed() < total {
    tokio::time::sleep(Duration::from_millis(10)).await;
    worst = worst.max(last.elapsed());
    last = Instant::now();
  }
  worst
}

fn blocker() {
  std::thread::sleep(Duration::from_millis(300)); // stands for any blocking call
}

fn independent(rt: Runtime, label: &str) {
  let worst = rt.block_on(async {
    let t = tokio::spawn(ticker(Duration::from_millis(500)));
    tokio::time::sleep(Duration::from_millis(50)).await;
    tokio::spawn(async { blocker() });
    t.await.unwrap()
  });
  println!("{label:<40} worst ticker gap = {worst:?}");
}

fn main() {
  independent(Builder::new_current_thread().enable_all().build().unwrap(), "current_thread");
  for n in [1, 2] {
    let rt = Builder::new_multi_thread().worker_threads(n).enable_all().build().unwrap();
    independent(rt, &format!("multi_thread, {n} worker(s)"));
  }

  // The LIFO trap: the blocker spawns a task, then blocks.
  let rt = Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap();
  rt.block_on(async {
    tokio::spawn(async {
      let spawned = Instant::now();
      let child = tokio::spawn(async move { spawned.elapsed() });
      blocker();
      println!("{:<40} child first polled after {:?}", "multi_thread, 2 workers, spawn then block", child.await.unwrap());
    })
    .await
    .unwrap();
  });
}

current_thread                           worst ticker gap = 309.941916ms
multi_thread, 1 worker(s)                worst ticker gap = 301.353ms
multi_thread, 2 worker(s)                worst ticker gap = 13.279167ms
multi_thread, 2 workers, spawn then block child first polled after 310.077375ms

let should_notify = if is_yield || !core.lifo_enabled {
    core.run_queue
        .push_back_or_overflow(task, self, &mut core.stats);
    true
} else {
    // Push to the LIFO slot
    let prev = core.lifo_slot.take();
    let ret = prev.is_some();

    if let Some(prev) = prev {
        core.run_queue
            .push_back_or_overflow(prev, self, &mut core.stats);
    }

    core.lifo_slot = Some(task);

    ret
};

// Only notify if not currently parked. If `park` is `None`, then the
// scheduling is from a resource driver. As notifications often come in
// batches, the notification is delayed until the park is complete.
if should_notify && core.park.is_some() {
    self.notify_parked_local();
}

for disable in [false, true] {
  let mut b = tokio::runtime::Builder::new_multi_thread();
  b.worker_threads(2).enable_all();
  if disable {
    b.disable_lifo_slot();
  }
  let rt = b.build().unwrap();
  rt.block_on(async move {
    tokio::spawn(async move {
      let spawned = Instant::now();
      let child = tokio::spawn(async move { spawned.elapsed() });
      std::thread::sleep(Duration::from_millis(300));
      println!("disable_lifo_slot = {disable:<5} child first polled after {:?}", child.await.unwrap());
    })
    .await
    .unwrap();
  });
}

disable_lifo_slot = false child first polled after 310.028167ms
disable_lifo_slot = true  child first polled after 6.834µs

// ~300ms of pure CPU work, split into steps so we can yield between them
fn step(x: u64) -> u64 {
  let mut h = x;
  for _ in 0..1_000 {
    h = h.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
  }
  h
}

const STEPS: u64 = 300_000;

fn cpu_job() -> u64 {
  (0..STEPS).fold(0, |acc, i| acc ^ step(i))
}

// 0. inline: the loop runs on the worker
tokio::spawn(async { cpu_job() });

// 1. yield_now every 1000 steps
tokio::spawn(async {
  let mut acc = 0;
  for i in 0..STEPS {
    acc ^= step(i);
    if i % 1000 == 0 {
      tokio::task::yield_now().await;
    }
  }
  acc
});

// 2. spawn_blocking
tokio::spawn(async { tokio::task::spawn_blocking(cpu_job).await.unwrap() });

// 3. rayon + oneshot
tokio::spawn(async {
  let (tx, rx) = tokio::sync::oneshot::channel();
  rayon::spawn(move || {
    let _ = tx.send(cpu_job());
  });
  rx.await.unwrap()
});

// 4. block_in_place (multi_thread, 2 workers, child sits in the LIFO slot)
let child = tokio::spawn(async move { spawned.elapsed() });
tokio::task::block_in_place(cpu_job);

cpu_job alone takes 336.336583ms (result 7c4c40d855acad00)
current_thread, inline loop          worst ticker gap = 352.78625ms
current_thread, yield_now / 1000     worst ticker gap = 13.028125ms
current_thread, spawn_blocking       worst ticker gap = 13.686166ms
current_thread, rayon + oneshot      worst ticker gap = 13.237416ms
multi_thread 2w, block_in_place      child first polled after 60.417µs