包装 waker：空闲与排队怎么算

struct State<M> {
    /// Where metrics should be recorded
    monitor: M,

    /// Instant at which the task was instrumented. This is used to track the time to first poll.
    instrumented_at: Instant,

    /// The instant, tracked as nanoseconds since `instrumented_at`, at which the future
    /// was last woken.
    woke_at: AtomicU64,

    /// Waker to forward notifications to.
    waker: AtomicWaker,
    // ...
    log: Option<Arc<SchedulingLog>>,
}

impl<M: Send + Sync> ArcWake for State<M> {
    fn wake_by_ref(arc_self: &Arc<State<M>>) {
        arc_self.on_wake();
        arc_self.waker.wake();
    }

    fn wake(self: Arc<State<M>>) {
        self.on_wake();
        self.waker.wake();
    }
}

impl<M> State<M> {
    fn on_wake(&self) {
        let woke_at: u64 = match self.instrumented_at.elapsed().as_nanos().try_into() {
            Ok(woke_at) => woke_at,
            // ...
            Err(_) => return,
        };

        // We don't actually care about the result
        let _ = self.woke_at.compare_exchange(0, woke_at, SeqCst, SeqCst);
    }
}

    let woke_at = state.woke_at.swap(0, SeqCst);
    // The state of a future is *idling* in the interim between the instant
    // it completes a `poll`, and the instant it is next awoken.
    if *idled_at < woke_at {
        // increment the counter of how many idles occurred
        metrics.total_idled_count.fetch_add(1, SeqCst);

        // compute the duration of the idle
        let idle_ns = woke_at.saturating_sub(*idled_at);
        // ...
        metrics.local_max_idle_duration_ns.fetch_max(idle_ns, SeqCst);
        metrics.global_max_idle_duration_ns.fetch_max(idle_ns, SeqCst);
        // adjust the total elapsed time monitored tasks spent idling
        metrics.total_idle_duration_ns.fetch_add(idle_ns, SeqCst);
    }
    // if this task spent any time in the scheduled state after instrumentation,
    // and after first poll, `woke_at` will be greater than 0.
    if woke_at > 0 {
        // increment the counter of how many schedules occurred
        metrics.total_scheduled_count.fetch_add(1, SeqCst);
        // ...
        let woke_instant = instrumented_at
            .checked_add(Duration::from_nanos(woke_at))
            .unwrap_or(poll_start);
        // ...
        let scheduled_ns = poll_start
            .saturating_duration_since(woke_instant)
            // ...
        let (count_bucket, duration_bucket) = // was the scheduling delay long or short?
            if scheduled >= metrics.long_delay_threshold {
                (&metrics.total_long_delay_count, &metrics.total_long_delay_duration_ns)
            } else {
                (&metrics.total_short_delay_count, &metrics.total_short_delay_duration_ns)
            };
        // ...
    }

tokio-metrics = "=0.5.2"
tokio = { version = "=1.54.1", features = ["full"] }

use std::future::{poll_fn, Future};
use std::pin::pin;
use std::task::Poll;
use std::time::{Duration, Instant};
use tokio_metrics::{TaskMetrics, TaskMonitor};

fn spin(d: Duration) {
  let t = Instant::now();
  while t.elapsed() < d {}
}

fn show(name: &str, m: &TaskMetrics) {
  println!(
    "{name:<10} polls={} idled={} idle={:>7.1?} scheduled={} sched={:>7.1?}",
    m.total_poll_count, m.total_idled_count, m.total_idle_duration,
    m.total_scheduled_count, m.total_scheduled_duration
  );
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  // A. 两次 poll 之间被唤醒两次：10ms 一次，40ms 一次；线程被占到 50ms
  let a = TaskMonitor::new();
  let task = tokio::spawn(a.instrument(async {
    let mut started = false;
    poll_fn(|cx| {
      if started {
        return Poll::Ready(());
      }
      started = true;
      let waker = cx.waker().clone();
      std::thread::spawn(move || {
        std::thread::sleep(Duration::from_millis(10));
        waker.wake_by_ref(); // 第一次唤醒
        std::thread::sleep(Duration::from_millis(30));
        waker.wake(); // 第二次唤醒
      });
      Poll::Pending
    })
    .await
  }));
  tokio::task::yield_now().await; // 让 A 先跑第一次 poll
  spin(Duration::from_millis(50)); // 占住唯一的线程
  task.await.unwrap();
  show("two wakes", &a.cumulative());

  // B. poll 里面立刻 wake_by_ref 自己，然后返回 Pending
  let b = TaskMonitor::new();
  tokio::spawn(b.instrument(async {
    let mut yielded = false;
    poll_fn(|cx| {
      if yielded {
        return Poll::Ready(());
      }
      yielded = true;
      cx.waker().wake_by_ref();
      spin(Duration::from_millis(5)); // 唤醒之后，这次 poll 还要忙 5ms
      Poll::Pending
    })
    .await
  }))
  .await
  .unwrap();
  show("self wake", &b.cumulative());

  // C. tokio::task::yield_now：tokio 把唤醒推迟到这次 poll 之后
  let c = TaskMonitor::new();
  tokio::spawn(c.instrument(async {
    spin(Duration::from_millis(5));
    tokio::task::yield_now().await;
  }))
  .await
  .unwrap();
  show("yield_now", &c.cumulative());

  // D. 嵌套：外层每次 poll 先忙 5ms，再 poll 内层
  let outer = TaskMonitor::new();
  let inner = TaskMonitor::new();
  let inner_fut = inner.instrument(tokio::time::sleep(Duration::from_millis(10)));
  let task = tokio::spawn(outer.instrument(async move {
    let mut inner_fut = pin!(inner_fut);
    poll_fn(|cx| {
      spin(Duration::from_millis(5));
      inner_fut.as_mut().poll(cx)
    })
    .await
  }));
  tokio::task::yield_now().await;
  spin(Duration::from_millis(30)); // 定时器到期后，外层任务还要再等一会儿才被 poll
  task.await.unwrap();
  show("outer", &outer.cumulative());
  show("inner", &inner.cumulative());
}

two wakes  polls=2 idled=1 idle= 12.5ms scheduled=1 sched= 37.5ms
self wake  polls=2 idled=0 idle=  0.0ns scheduled=1 sched=  5.0ms
yield_now  polls=2 idled=1 idle= 14.3µs scheduled=1 sched=334.0ns
outer      polls=2 idled=1 idle= 30.0ms scheduled=1 sched=542.0ns
inner      polls=2 idled=1 idle= 30.0ms scheduled=1 sched=  5.0ms

        // Don't wake the task immediately, as that would push it right back
        // onto the run queue and it could be polled again before other tasks
        // or the IO/timer driver get a chance to run. Instead, hand the waker
        // to the scheduler, which wakes deferred tasks only after it has run
        // out of ready tasks and polled the driver. When polled from outside
        // a Tokio runtime, the waker is woken immediately.
        context::defer(cx.waker());

victm wall=13.5ms poll=178.0ns idle=13.5ms sched= 4.0µs  slow_polls=0 long_delays=0
victm wall=13.5ms poll=237.0ns idle=13.5ms sched= 4.6µs  slow_polls=0 long_delays=0
victm wall=13.5ms poll=279.0ns idle=13.5ms sched= 5.2µs  slow_polls=0 long_delays=0