驱动栈与 park

pub(crate) struct Driver {
    inner: TimeDriver,
}

pub(crate) fn new(cfg: Cfg) -> io::Result<(Self, Handle)> {
    let (io_stack, io_handle, signal_handle) =
        create_io_stack(cfg.enable_io, cfg.nevents, cfg.nevents_busy)?;

    let clock = create_clock(cfg.enable_pause_time, cfg.start_paused);

    let (time_driver, time_handle) =
        create_time_driver(cfg.enable_time, cfg.timer_flavor, io_stack, &clock);

    Ok((
        Self { inner: time_driver },
        Handle {
            io: io_handle,
            signal: signal_handle,
            time: time_handle,
            clock,
        },
    ))
}

let ret = if enabled {
    let (io_driver, io_handle) = crate::runtime::io::Driver::new(nevents, nevents_busy)?;

    let (signal_driver, signal_handle) = create_signal_driver(io_driver, &io_handle)?;
    let process_driver = create_process_driver(signal_driver);

    (IoStack::Enabled(process_driver), IoHandle::Enabled(io_handle), signal_handle)
} else {
    let park_thread = ParkThread::new();
    let unpark_thread = park_thread.unpark();
    (IoStack::Disabled(park_thread), IoHandle::Disabled(unpark_thread), Default::default())
};

pub(crate) enum TimeDriver {
    Enabled {
        driver: crate::runtime::time::Driver,
    },
    EnabledAlt(IoStack),
    Disabled(IoStack),
}

pub(crate) struct Handle {
    /// IO driver handle
    pub(crate) io: IoHandle,

    /// Signal driver handle
    #[cfg_attr(any(not(unix), loom), allow(dead_code))]
    pub(crate) signal: SignalHandle,

    /// Time driver handle
    pub(crate) time: TimeHandle,

    /// Source of `Instant::now()`
    #[cfg_attr(not(all(feature = "time", feature = "test-util")), allow(dead_code))]
    pub(crate) clock: Clock,
}

A Tokio 1.x context was found, but IO is disabled. Call `enable_io` on the runtime builder to enable IO.
A Tokio 1.x context was found, but timers are disabled. Call `enable_time` on the runtime builder to enable timers.

fn park_internal(&mut self, rt_handle: &driver::Handle, limit: Option<Duration>) {
    let handle = rt_handle.time();
    let mut lock = handle.inner.lock();

    assert!(!handle.is_shutdown());

    let next_wake = lock.wheel.next_expiration_time();
    lock.next_wake =
        next_wake.map(|t| NonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap()));

    drop(lock);

    match next_wake {
        Some(when) => {
            let now = handle.time_source.now(rt_handle.clock());
            // Note that we effectively round up to 1ms here - this avoids
            // very short-duration microsecond-resolution sleeps that the OS
            // might treat as zero-length.
            let mut duration = handle
                .time_source
                .tick_to_duration(when.saturating_sub(now));

            if duration > Duration::from_millis(0) {
                if let Some(limit) = limit {
                    duration = std::cmp::min(limit, duration);
                }

                self.park_thread_timeout(rt_handle, duration);
            } else {
                self.park.park_timeout(rt_handle, Duration::from_secs(0));
            }
        }
        None => {
            if let Some(duration) = limit {
                self.park_thread_timeout(rt_handle, duration);
            } else {
                self.park.park(rt_handle);
            }
        }
    }

    // Process pending timers after waking up
    handle.process(rt_handle.clock());
}

pub(crate) fn park(&mut self, handle: &driver::Handle) {
    self.park.park(handle);
    GlobalOrphanQueue::reap_orphans(&self.signal_handle);
}

pub(crate) fn park(&mut self, handle: &driver::Handle) {
    self.io.park(handle);
    self.process();
}

match self.poll.poll(events, max_wait) {

pub(crate) fn unpark(&self) {
    #[cfg(feature = "time")]
    if let Some(handle) = &self.time {
        handle.unpark();
    }

    self.io.unpark();
}

pub(crate) fn unpark(&self) {
    match self {
        IoHandle::Enabled(handle) => handle.unpark(),
        IoHandle::Disabled(handle) => handle.unpark(),
    }
}

if token == TOKEN_WAKEUP {
    // Nothing to do, the event is used to unblock the I/O driver
} else if token == TOKEN_SIGNAL {
    self.signal_ready = true;
} else {

struct Inner {
    state: AtomicUsize,
    mutex: Mutex<()>,
    condvar: Condvar,
    // ...
}

const EMPTY: usize = 0;
const PARKED: usize = 1;
const NOTIFIED: usize = 2;

fn park(&self) {
    // If we were previously notified then we consume this notification and
    // return quickly.
    if self
        .state
        .compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst)
        .is_ok()
    {
        return;
    }

    // Otherwise we need to coordinate going to sleep
    let mut m = self.mutex.lock();

    match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
        Ok(_) => {}
        Err(NOTIFIED) => {
            // ...
            let old = self.state.swap(EMPTY, SeqCst);
            debug_assert_eq!(old, NOTIFIED, "park state changed unexpectedly");

            return;
        }
        Err(actual) => panic!("inconsistent park state; actual = {actual}"),
    }

    loop {
        m = self.condvar.wait(m).unwrap();

        if self
            .state
            .compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst)
            .is_ok()
        {
            // got a notification
            return;
        }

        // spurious wakeup, go back to sleep
    }
}

fn unpark(&self) {
    // ...
    match self.state.swap(NOTIFIED, SeqCst) {
        EMPTY => return,    // no one was waiting
        NOTIFIED => return, // already unparked
        PARKED => {}        // gotta go wake someone up
        _ => panic!("inconsistent state in unpark"),
    }

    // ...
    drop(self.mutex.lock());

    self.condvar.notify_one();
}

use std::panic;
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
use std::sync::Arc;
use std::sync::mpsc as std_mpsc;
use std::thread;
use std::time::{Duration, Instant};
use tokio::runtime::Builder;
use tokio::sync::mpsc;

const ROUNDS: u32 = 100_000;

// 普通线程发消息给 current_thread 运行时里的任务，任务再用 std channel 回一条。
// 每一轮运行时线程都会睡下去，再被普通线程叫醒。
fn ping_pong(io: bool) -> (Duration, usize) {
  let parks = Arc::new(AtomicUsize::new(0));
  let p2 = parks.clone();
  let mut b = Builder::new_current_thread();
  b.on_thread_park(move || {
    p2.fetch_add(1, Relaxed);
  });
  if io {
    b.enable_io();
  }
  let rt = b.build().unwrap();
  let (tx, mut rx) = mpsc::unbounded_channel::<u32>();
  let (back_tx, back_rx) = std_mpsc::channel::<u32>();

  let t = thread::spawn(move || {
    let start = Instant::now();
    for i in 0..ROUNDS {
      tx.send(i).unwrap();
      back_rx.recv().unwrap();
    }
    start.elapsed()
  });
  rt.block_on(async move {
    while let Some(i) = rx.recv().await {
      back_tx.send(i).unwrap();
    }
  });
  (t.join().unwrap(), parks.load(Relaxed))
}

fn main() {
  for io in [false, true, false, true] {
    let (d, parks) = ping_pong(io);
    println!(
      "enable_io = {io:<5}  {ROUNDS} rounds  parks {parks}  {:>4} ms  {:.2} us/round",
      d.as_millis(),
      d.as_secs_f64() * 1e6 / ROUNDS as f64
    );
  }

  panic::set_hook(Box::new(|info| {
    let msg = info.payload().downcast_ref::<&str>().map(|s| s.to_string())
      .or_else(|| info.payload().downcast_ref::<String>().cloned()).unwrap_or_default();
    println!("panic: {msg}");
  }));
  let rt = Builder::new_current_thread().build().unwrap();
  let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    rt.block_on(async { tokio::time::sleep(Duration::from_millis(1)).await });
  }));
  let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    rt.block_on(async { tokio::net::TcpListener::bind("127.0.0.1:0").await.map(|_| ()) }).unwrap();
  }));
}

enable_io = false  100000 rounds  parks 99904   430 ms  4.30 us/round
enable_io = true   100000 rounds  parks 99886   324 ms  3.24 us/round
enable_io = false  100000 rounds  parks 99876   432 ms  4.33 us/round
enable_io = true   100000 rounds  parks 99900   319 ms  3.19 us/round
panic: A Tokio 1.x context was found, but timers are disabled. Call `enable_time` on the runtime builder to enable timers.
panic: A Tokio 1.x context was found, but IO is disabled. Call `enable_io` on the runtime builder to enable IO.

use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
use std::sync::Arc;
use std::time::Duration;
use tokio::runtime::{Builder, Runtime};

fn rt(parks: Arc<AtomicUsize>) -> Runtime {
  Builder::new_current_thread()
    .enable_time()
    .on_thread_park(move || {
      parks.fetch_add(1, Relaxed);
    })
    .build()
    .unwrap()
}

fn main() {
  let parks = Arc::new(AtomicUsize::new(0));

  let r = rt(parks.clone());
  r.block_on(async {
    for _ in 0..1000 {
      tokio::task::yield_now().await;
    }
  });
  println!("1000 x yield_now     -> parks {}", parks.swap(0, Relaxed));

  let r = rt(parks.clone());
  r.block_on(async {
    for _ in 0..5 {
      tokio::time::sleep(Duration::from_millis(10)).await;
    }
  });
  println!("5 x sleep(10ms)      -> parks {}", parks.swap(0, Relaxed));

  let r = rt(parks.clone());
  r.block_on(async {
    let mut handles = Vec::new();
    for _ in 0..5 {
      handles.push(tokio::spawn(tokio::time::sleep(Duration::from_millis(10))));
    }
    for h in handles {
      h.await.unwrap();
    }
  });
  println!("5 spawned sleep(10ms) -> parks {}", parks.swap(0, Relaxed));
}

1000 x yield_now     -> parks 0
5 x sleep(10ms)      -> parks 10
5 spawned sleep(10ms) -> parks 2