分层时间轮

pub(crate) struct Wheel {
    /// The number of milliseconds elapsed since the wheel started.
    elapsed: u64,

    /// Timer wheel.
    /// ...
    levels: Box<[Level; NUM_LEVELS]>,

    /// Entries queued for firing
    pending: LinkedList<TimerShared>,
}

const NUM_LEVELS: usize = 6;

const BITS_PER_LEVEL: usize = 6;

/// The maximum duration of a `Sleep`.
const MAX_DURATION: u64 = 1 << (BITS_PER_LEVEL * NUM_LEVELS);

fn slot_for(duration: u64, level: usize) -> usize {
    ((duration >> (level * super::BITS_PER_LEVEL)) % LEVEL_MULT as u64) as usize
}

fn level_for(elapsed: u64, when: u64) -> usize {
    const SLOT_MASK: u64 = (1 << BITS_PER_LEVEL) - 1;

    // Mask in the trailing bits ignored by the level calculation in order to cap
    // the possible leading zeros
    let masked = elapsed ^ when | SLOT_MASK;

    if masked >= MAX_DURATION {
        // Fudge the timer into the top level
        return NUM_LEVELS - 1;
    }

    masked.ilog2() as usize / BITS_PER_LEVEL
}

fn main() {
  for (elapsed, when) in [(0u64, 10u64), (0, 100), (60, 70), (0, 4200), (64, 100), (4096, 4200), (4160, 4200), (0, 1 << 40), (1 << 36, 1 << 40)] {
    let x = elapsed ^ when;
    let level = level_for(elapsed, when);
    println!(
      "elapsed={elapsed:>6} when={when:>14} xor={x:>14} masked.ilog2={:>2} level={level} slot={}",
      (x | 63).ilog2(),
      slot_for(when, level)
    );
  }
}

elapsed=     0 when=            10 xor=            10 masked.ilog2= 5 level=0 slot=10
elapsed=     0 when=           100 xor=           100 masked.ilog2= 6 level=1 slot=1
elapsed=    60 when=            70 xor=           122 masked.ilog2= 6 level=1 slot=1
elapsed=     0 when=          4200 xor=          4200 masked.ilog2=12 level=2 slot=1
elapsed=    64 when=           100 xor=            36 masked.ilog2= 5 level=0 slot=36
elapsed=  4096 when=          4200 xor=           104 masked.ilog2= 6 level=1 slot=1
elapsed=  4160 when=          4200 xor=            40 masked.ilog2= 5 level=0 slot=40
elapsed=     0 when= 1099511627776 xor= 1099511627776 masked.ilog2=40 level=5 slot=0
elapsed=68719476736 when= 1099511627776 xor= 1168231104512 masked.ilog2=40 level=5 slot=0

fn next_occupied_slot(&self, now: u64) -> Option<usize> {
    if self.occupied == 0 {
        return None;
    }

    // Add the +1 offset for the `now_slot` to ignore the slot that `now` fits in,
    // since it's the farthest timer that could appear from `now`.
    // ...
    let now_slot = ((now / slot_range(self.level)) % LEVEL_MULT as u64) as usize + 1;
    let occupied = self.occupied.rotate_right(now_slot as u32);
    let zeros = occupied.trailing_zeros() as usize;
    let slot = (zeros + now_slot) % LEVEL_MULT;

    Some(slot)
}

let level_range = level_range(self.level);
let slot_range = slot_range(self.level);

// Compute the start date of the current level by masking the low bits
// of `now` (`level_range` is a power of 2).
let level_start = now & !(level_range - 1);
let mut deadline = level_start + slot as u64 * slot_range;

if deadline <= now {
    // A timer is in a slot "prior" to the current time. ...
    debug_assert_eq!(self.level, super::NUM_LEVELS - 1);

    deadline += level_range;
}

let mut entries = self.take_entries(expiration);
let deadline = if now == u64::MAX {
    now
} else {
    expiration.deadline
};

while let Some(item) = entries.pop_back() {
    // ...

    // Try to expire the entry; this is cheap (doesn't synchronize) if
    // the timer is not expired, and updates registered_when.
    match unsafe { item.mark_pending(deadline) } {
        Ok(()) => {
            // Item was expired
            self.pending.push_front(item);
        }
        Err(expiration_tick) => {
            let level = level_for(expiration.deadline, expiration_tick);
            unsafe {
                self.levels[level].add_entry(item);
            }
        }
    }
}

while let Some(entry) = lock.wheel.poll(now) {
    debug_assert!(unsafe { entry.is_pending() });

    // SAFETY: We hold the driver lock, and just removed the entry from any linked lists.
    if let Some(waker) = unsafe { entry.fire(Ok(())) } {
        waker_list.push(waker);

        if !waker_list.can_push() {
            // Wake a batch of wakers. To avoid deadlock, we must do this with the lock temporarily dropped.
            drop(lock);

            waker_list.wake_all();

            lock = self.inner.lock();
        }
    }
}

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());
}

use std::sync::Mutex;
use tokio::time::{sleep, Duration, Instant};

static START: Mutex<Option<Instant>> = Mutex::new(None);

fn main() {
  for ms in [64u64, 5000] {
    let rt = tokio::runtime::Builder::new_current_thread()
      .enable_time()
      .start_paused(true)
      .on_thread_unpark(|| {
        let start = START.lock().unwrap().unwrap();
        println!("  醒来，tokio 时钟 = {}ms", (Instant::now() - start).as_millis());
      })
      .build()
      .unwrap();

    rt.block_on(async {
      *START.lock().unwrap() = Some(Instant::now());
      println!("sleep({ms}ms)");
      sleep(Duration::from_millis(ms)).await;
    });
  }
}

sleep(64ms)
  醒来，tokio 时钟 = 64ms
sleep(5000ms)
  醒来，tokio 时钟 = 4096ms
  醒来，tokio 时钟 = 4992ms
  醒来，tokio 时钟 = 5000ms