（unstable）time_alt

/// Enables the alternative timer implementation, which is disabled by default.
///
/// The alternative timer implementation is an unstable feature that may
/// provide better performance on multi-threaded runtimes with a large number
/// of worker threads.
///
/// This option only applies to multi-threaded runtimes. Attempting to use
/// this option with any other runtime type will have no effect.
/// ...
#[cfg(all(tokio_unstable, feature = "time", feature = "rt-multi-thread"))]
pub fn enable_alt_timer(&mut self) -> &mut Self {
    self.enable_time();
    self.timer_flavor = TimerFlavor::Alternative;
    self
}

pub(crate) struct LocalContext {
    pub(crate) wheel: Wheel,
    pub(crate) registration_queue: RegistrationQueue,
    pub(crate) canc_tx: cancellation_queue::Sender,
    pub(crate) canc_rx: cancellation_queue::Receiver,
}

struct State {
    cancelled: bool,
    woken_up: bool,
    waker: Option<Waker>,
    cancel_tx: Option<Sender>,
}

pub(crate) struct Entry {
    /// The intrusive pointer used by [`super::cancellation_queue`].
    cancel_pointers: linked_list::Pointers<Entry>,

    /// The intrusive pointer used by any of the following queues:
    ///
    /// - [`Wheel`]
    /// - [`RegistrationQueue`]
    /// - [`WakeQueue`]
    /// ...
    extra_pointers: linked_list::Pointers<Entry>,

    /// The tick when this entry is scheduled to expire.
    deadline: u64,

    state: Mutex<State>,
    // ...
}

pub(crate) fn new(handle: scheduler::Handle, deadline: u64) -> Self {
    let entry = with_current_temp_local_context(&handle, |ctx| match ctx {
        Some(TempLocalContext::Running { registration_queue }) => {
            let entry = EntryHandle::new(deadline);
            unsafe { registration_queue.push_front(entry.clone()) }
            entry
        }
        #[cfg(feature = "rt-multi-thread")]
        Some(TempLocalContext::Shutdown) => panic!("{RUNTIME_SHUTTING_DOWN_ERROR}"),

        _ => {
            let entry = EntryHandle::new(deadline);
            push_from_remote(&handle, entry.clone());
            entry
        }
    });

    Timer { entry }
}

util::time_alt::process_registration_queue(
    &mut time_cx.registration_queue,
    &mut time_cx.wheel,
    &time_cx.canc_tx,
    &mut wake_queue,
);
util::time_alt::insert_inject_timers(
    &mut time_cx.wheel,
    &time_cx.canc_tx,
    handle.take_remote_timers(),
    &mut wake_queue,
);
util::time_alt::remove_cancelled_timers(&mut time_cx.wheel, &mut time_cx.canc_rx);
let should_yield = !wake_queue.is_empty();

let next_timer = util::time_alt::next_expiration_time(&time_cx.wheel, &handle.driver);

pub(crate) fn cancel(&self) {
    let mut lock = self.entry.state.lock();
    if !lock.cancelled {
        lock.cancelled = true;
        let waker = lock.waker.take();
        let cancel_tx = lock.cancel_tx.take();
        drop(lock);

        if let Some(cancel_tx) = cancel_tx {
            // Safety: we can guarantee that `self` is not in any cancellation queue
            // because the `self.cancelled` was just set to `true`.
            unsafe {
                cancel_tx.send(self.clone());
            }
        }

        drop(waker);
    }
}

// 需要 RUSTFLAGS="--cfg tokio_unstable"
fn run(alt: bool) {
  let mut b = tokio::runtime::Builder::new_multi_thread();
  b.worker_threads(4).enable_all();
  if alt {
    b.enable_alt_timer();
  }
  let rt = b.build().unwrap();
  let name = if alt { "alt        " } else { "traditional" };

  rt.block_on(async move {
    // 1. block_on 的线程不是 worker：sleep 走 push_remote_timer
    let t = Instant::now();
    sleep(Duration::from_millis(30)).await;
    println!("{name} block_on 里 sleep(30ms): {}ms", t.elapsed().as_millis());

    // 2. 一个 worker 被卡住时，登记在它那里的定时器
    let h = tokio::spawn(async {
      let t = Instant::now();
      // 先 spawn 一个会卡住线程 300ms 的任务：它进当前 worker 的 LIFO 槽，
      // 等本任务返回 Pending 后就在同一个 worker 上运行
      tokio::spawn(async {
        std::thread::sleep(Duration::from_millis(300));
      });
      sleep(Duration::from_millis(20)).await;
      t.elapsed().as_millis()
    });
    println!("{name} 所在 worker 被卡 300ms 时 sleep(20ms): {}ms", h.await.unwrap());
  });
}

fn main() {
  run(false);
  run(true);
}

traditional block_on 里 sleep(30ms): 33ms
traditional 所在 worker 被卡 300ms 时 sleep(20ms): 23ms
alt         block_on 里 sleep(30ms): 33ms
alt         所在 worker 被卡 300ms 时 sleep(20ms): 308ms