Instrumented 与 instrument_poll

    pub fn instrument_with<F, M: AsRef<TaskMonitorCore> + Send + Sync + 'static>(
        task: F,
        monitor: M,
    ) -> Instrumented<F, M> {
        monitor
            .as_ref()
            .metrics
            .instrumented_count
            .fetch_add(1, SeqCst);

        let log = monitor
            .as_ref()
            .record_scheduling_log
            .then(|| Arc::new(SchedulingLog::default()));

        let state: State<M> = State {
            monitor,
            instrumented_at: Instant::now(),
            woke_at: AtomicU64::new(0),
            waker: AtomicWaker::new(),
            log,
        };

        let instrumented: Instrumented<F, M> = Instrumented {
            task,
            did_poll_once: false,
            idled_at: 0,
            state: Arc::new(state),
        };

        instrumented
    }

pin_project! {
    #[derive(Debug)]
    pub struct Instrumented<T, M: AsRef<TaskMonitorCore> = TaskMonitor> {
        // The task being instrumented
        #[pin]
        task: T,

        // True when the task is polled for the first time
        did_poll_once: bool,

        // The instant, tracked as nanoseconds since `instrumented_at`, at which the future finished
        // its last poll.
        idled_at: u64,

        // State shared between the task and its instrumented waker.
        state: Arc<State<M>>,
    }

    impl<T, M: AsRef<TaskMonitorCore>> PinnedDrop for Instrumented<T, M> {
        fn drop(this: Pin<&mut Self>) {
            this.state.monitor.as_ref().metrics.dropped_count.fetch_add(1, SeqCst);
        }
    }
}

fn instrument_poll<T, M: AsRef<TaskMonitorCore> + Send + Sync + 'static, Out>(
    cx: &mut Context<'_>,
    instrumented: Pin<&mut Instrumented<T, M>>,
    poll_fn: impl FnOnce(Pin<&mut T>, &mut Context<'_>) -> Poll<Out>,
) -> Poll<Out> {

    let poll_start = Instant::now();
    let this = instrumented.project();
    let idled_at = this.idled_at;
    let state = this.state;
    let instrumented_at = state.instrumented_at;
    let metrics = &state.monitor.as_ref().metrics;
    // ...
    if !*this.did_poll_once {
        // ...
        *this.did_poll_once = true;
        // ...
        let elapsed = poll_start
            .saturating_duration_since(instrumented_at)
            .as_nanos()
            .try_into()
            .unwrap_or(u64::MAX);
        // add this duration to `time_to_first_poll_ns_total`
        metrics.total_first_poll_delay_ns.fetch_add(elapsed, SeqCst);
        // ...
        metrics.first_poll_count.fetch_add(1, SeqCst);
    }

use std::time::Duration;
use futures::StreamExt;
use tokio_metrics::TaskMonitor;

#[tokio::main]
async fn main() {
  // 1. instrument() 和 spawn 之间隔 200ms
  let early = TaskMonitor::new();
  let fut = early.instrument(async {});
  tokio::time::sleep(Duration::from_millis(200)).await;
  tokio::spawn(fut).await.unwrap();
  // 2. 在 spawn 的同一行 instrument
  let late = TaskMonitor::new();
  tokio::time::sleep(Duration::from_millis(200)).await;
  tokio::spawn(late.instrument(async {})).await.unwrap();
  println!("instrument early: first_poll_delay={:?}", early.cumulative().total_first_poll_delay);
  println!("instrument late:  first_poll_delay={:?}", late.cumulative().total_first_poll_delay);

  // 3. 包一个 Stream：每次 poll_next 都算一次 poll
  let streams = TaskMonitor::new();
  let items: Vec<u32> = streams
    .instrument(futures::stream::iter([1, 2, 3]).then(|x| async move {
      tokio::task::yield_now().await;
      x
    }))
    .collect()
    .await;
  let s = streams.cumulative();
  println!(
    "stream: items={items:?} instrumented={} first_polls={} polls={} dropped={}",
    s.instrumented_count, s.first_poll_count, s.total_poll_count, s.dropped_count
  );
}

instrument early: first_poll_delay=202.161792ms
instrument late:  first_poll_delay=12.292µs
stream: items=[1, 2, 3] instrumented=1 first_polls=1 polls=7 dropped=1

    let woke_at = state.woke_at.swap(0, SeqCst);
    // ...
    if *idled_at < woke_at {
        // ... 记一次空闲：woke_at - idled_at
    }
    // ...
    if woke_at > 0 {
        // ... 记一次排队：poll_start - woke_at，按 long_delay_threshold 分长短
    }

    // Register the waker
    state.waker.register(cx.waker());
    // Get the instrumented waker
    let waker_ref = futures_util::task::waker_ref(state);
    let mut cx = Context::from_waker(&waker_ref);

    let inner_poll_start = Instant::now();
    // ...
    let ret = if let Some(log) = &state.log {
        let _sched_guard = SchedulingLogGuard::enter(log.clone());
        poll_fn(this.task, &mut cx)
    } else {
        poll_fn(this.task, &mut cx)
    };
    let inner_poll_end = Instant::now();

    /* idle time starts now */
    *idled_at = inner_poll_end
        .saturating_duration_since(instrumented_at)
        .as_nanos()
        .try_into()
        .unwrap_or(u64::MAX);
    /* accounting for poll time */
    let inner_poll_duration = inner_poll_end.saturating_duration_since(inner_poll_start);
    // ...
    let (count_bucket, duration_bucket) = // was this a slow or fast poll?
            if inner_poll_duration >= metrics.slow_poll_threshold {
                (&metrics.total_slow_poll_count, &metrics.total_slow_poll_duration)
            } else {
                (&metrics.total_fast_poll_count, &metrics.total_fast_poll_duration_ns)
            };
    // update the appropriate bucket
    count_bucket.fetch_add(1, SeqCst);
    duration_bucket.fetch_add(inner_poll_ns, SeqCst);
    ret

use tokio_metrics::TaskMonitorCore;
static CORE: TaskMonitorCore = TaskMonitorCore::new();

fn main() {
  let s = CORE.instrument(futures::stream::iter([1, 2]));
  use futures::StreamExt;
  let _ = s.collect::<Vec<_>>();
}

error[E0599]: the method `collect` exists for struct `Instrumented<futures::stream::Iter<std::array::IntoIter<{integer}, 2>>, &TaskMonitorCore>`, but its trait bounds were not satisfied
   --> src/main.rs:7:13
    |
  7 |     let _ = s.collect::<Vec<_>>();
    |               ^^^^^^^ method cannot be called due to unsatisfied trait bounds
    |
   ::: pin-project-lite-0.2.17/src/lib.rs:745:9
    |
745 | /         $vis struct $ident $($def_generics)*
...
    | |_________- doesn't satisfy `_: Iterator`, `_: StreamExt` or `_: Stream`
    |
    = note: the following trait bounds were not satisfied:
            `Instrumented<futures::stream::Iter<std::array::IntoIter<{integer}, 2>>, &TaskMonitorCore>: Stream`
            which is required by `Instrumented<futures::stream::Iter<std::array::IntoIter<{integer}, 2>>, &TaskMonitorCore>: StreamExt`
            ...