Instrument 与异步

pub trait Instrument: Sized {
    fn instrument(self, span: Span) -> Instrumented<Self> {
        Instrumented {
            inner: ManuallyDrop::new(self),
            span,
        }
    }

    #[inline]
    fn in_current_span(self) -> Instrumented<Self> {
        self.instrument(Span::current())
    }
}

pin_project! {
    #[project = InstrumentedProj]
    #[project_ref = InstrumentedProjRef]
    #[derive(Debug, Clone)]
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    pub struct Instrumented<T> {
        // `ManuallyDrop` is used here to to enter instrument `Drop` by entering
        // `Span` and executing `ManuallyDrop::drop`.
        #[pin]
        inner: ManuallyDrop<T>,
        span: Span,
    }

    impl<T> PinnedDrop for Instrumented<T> {
        fn drop(this: Pin<&mut Self>) {
            let this = this.project();
            let _enter = this.span.enter();
            // SAFETY: 1. `Pin::get_unchecked_mut()` is safe, because this isn't
            //             different from wrapping `T` in `Option` and calling
            //             `Pin::set(&mut this.inner, None)`, except avoiding
            //             additional memory overhead.
            //         2. `ManuallyDrop::drop()` is safe, because
            //            `PinnedDrop::drop()` is guaranteed to be called only
            //            once.
            unsafe { ManuallyDrop::drop(this.inner.get_unchecked_mut()) }
        }
    }
}

impl<T: Future> Future for Instrumented<T> {
    type Output = T::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let (span, inner) = self.project().span_and_inner_pin_mut();
        let _enter = span.enter();
        inner.poll(cx)
    }
}

    /// The drop guard returned by `Span::enter` exits the span when it is
    /// dropped. When an async function or async block yields at an `.await`
    /// point, the current scope is _exited_, but values in that scope are
    /// **not** dropped (because the async block will eventually resume
    /// execution from that await point). This means that _another_ task will
    /// begin executing while _remaining_ in the entered span. This results in
    /// an incorrect trace.

tracing = "=0.1.44"
tracing-subscriber = "=0.3.23"
tokio = { version = "=1.54.1", features = ["macros", "rt"] }

use tracing::{Instrument, info, info_span};

// 错误写法：守卫跨 .await
async fn wrong(id: u32) {
  let span = info_span!("task", id);
  let _g = span.enter();
  info!(who = id, "start");
  tokio::task::yield_now().await;
  info!(who = id, "after yield");
}

// 正确写法：span 交给 Instrumented，每次 poll 进出
async fn right(id: u32) {
  async {
    info!(who = id, "start");
    tokio::task::yield_now().await;
    info!(who = id, "after yield");
  }
  .instrument(info_span!("task", id))
  .await
}

// 旁观者：自己没有任何 span
async fn bystander() {
  info!(who = "c", "no span here");
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  tracing_subscriber::fmt()
    .with_ansi(false)
    .without_time()
    .with_target(false)
    .init();

  info!("--- enter() 跨 .await");
  let tasks = [
    tokio::spawn(wrong(1)),
    tokio::spawn(wrong(2)),
    tokio::spawn(bystander()),
  ];
  for t in tasks {
    t.await.unwrap();
  }
  info!("--- .instrument()");
  let tasks = [
    tokio::spawn(right(1)),
    tokio::spawn(right(2)),
    tokio::spawn(bystander()),
  ];
  for t in tasks {
    t.await.unwrap();
  }
  info!("--- 结束");
}

 INFO --- enter() 跨 .await
 INFO task{id=1}: start who=1
 INFO task{id=1}:task{id=2}: start who=2
 INFO task{id=1}:task{id=2}: no span here who="c"
 INFO task{id=1}:task{id=2}: after yield who=2
 INFO task{id=1}: after yield who=1
 INFO --- .instrument()
 INFO task{id=1}: start who=1
 INFO task{id=2}: start who=2
 INFO no span here who="c"
 INFO task{id=2}: after yield who=2
 INFO task{id=1}: after yield who=1
 INFO --- 结束

use tracing::info_span;

async fn work() {}

async fn hold_entered() {
  let span = info_span!("a");
  let _g = span.enter();
  work().await;
}

async fn hold_entered_span() {
  let _g = info_span!("b").entered();
  work().await;
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  hold_entered().await;
  hold_entered_span().await;
}

warning[E0602]: unknown lint: `clippy::await_holding_span_guard`
  |
  = help: did you mean: `clippy::await_holding_lock`
  = note: requested on the command line with `-W clippy::await_holding_span_guard`
  = note: `#[warn(unknown_lints)]` on by default

await-holding-invalid-types = [
  { path = "tracing::span::Entered", reason = "用 Instrument::instrument 代替" },
  { path = "tracing::span::EnteredSpan", reason = "用 Instrument::instrument 代替" },
]

warning: holding a disallowed type across an await point `tracing::span::Entered`
 --> src/bin/clippy_guard.rs:7:7
  |
7 |   let _g = span.enter();
  |       ^^
  |
  = note: 用 Instrument::instrument 代替
  = help: for further information visit https://rust-lang.github.io/rust-clippy/rust-1.99.0/index.html#await_holding_invalid_type
  = note: `#[warn(clippy::await_holding_invalid_type)]` on by default

warning: holding a disallowed type across an await point `tracing::span::EnteredSpan`
  --> src/bin/clippy_guard.rs:12:7
   |
12 |   let _g = info_span!("b").entered();
   |       ^^
   |
   = note: 用 Instrument::instrument 代替
   = help: for further information visit https://rust-lang.github.io/rust-clippy/rust-1.99.0/index.html#await_holding_invalid_type

    #[inline]
    fn with_current_subscriber(self) -> WithDispatch<Self> {
        WithDispatch {
            inner: self,
            dispatcher: dispatcher::get_default(|default| default.clone()),
        }
    }

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        let dispatcher = this.dispatcher;
        let future = this.inner;
        let _default = dispatcher::set_default(dispatcher);
        future.poll(cx)
    }

tracing = "=0.1.44"
tracing-core = "=0.1.36"
tokio = { version = "=1.54.1", features = ["macros", "rt-multi-thread"] }

#[path = "../printer.rs"]
mod printer;

use tracing::{Instrument, info, info_span, instrument::WithSubscriber};

#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() {
  // 只对 main 所在的线程生效；spawn 出去的任务跑在工作线程上
  let _guard = tracing::subscriber::set_default(printer::Printer::new());

  println!("[1] spawn(async {{ info!(..) }})");
  tokio::spawn(async { info!("plain") }).await.unwrap();

  println!("[2] spawn(async {{ info!(..) }}.instrument(span))");
  tokio::spawn(async { info!("instrumented") }.instrument(info_span!("req")))
    .await
    .unwrap();

  println!("[3] spawn(async {{ info!(..) }}.with_current_subscriber())");
  tokio::spawn(async { info!("with dispatch") }.with_current_subscriber())
    .await
    .unwrap();

  println!("[4] instrument(span).with_current_subscriber()");
  tokio::spawn(
    async { info!("both") }
      .instrument(info_span!("req2"))
      .with_current_subscriber(),
  )
  .await
  .unwrap();
}

[1] spawn(async { info!(..) })
[2] spawn(async { info!(..) }.instrument(span))
  new_span(req) -> 1
  enter(req)
  exit(req)
  enter(req)
  exit(req)
  try_close(req) refs=0 -> true
[3] spawn(async { info!(..) }.with_current_subscriber())
  event message=with dispatch (in -)
[4] instrument(span).with_current_subscriber()
  new_span(req2) -> 2
  enter(req2)
  event message=both (in req2)
  exit(req2)
  enter(req2)
  exit(req2)
  try_close(req2) refs=0 -> true

  new_span(req) -> 1
[timeout 开始]
  enter(req)
  event message=poll 1 (in req)
  exit(req)
  enter(req)
  event message=poll 2 (in req)
  exit(req)
  enter(req)
  exit(req)
  enter(req)
  event message=Noisy dropped (in req)
  exit(req)
  try_close(req) refs=0 -> true
[timeout 返回 true]