feat(executor): instrument tasks for tokio-console (#987)

* feat(executor): give tasks a runtime.spawn span

tokio-console collects its data through tracing spans and events that
follow a fixed naming convention; it is not tied to tokio's internals, so
any executor emitting the same spans and events can be observed with it.

Behind a new `console` feature, give every task a `runtime.spawn` span,
entered while the task is polled. That is enough for the console to report
poll counts, busy/idle/scheduled times and the poll time histograms.

The instrumentation lives in the new `console` module, which has an
enabled and a disabled variant. The disabled one is what is compiled
without the feature: `TaskSpan` becomes zero-sized, so the task header
keeps its layout, and every method an empty inlined function.

The span sits in that header, so it is dropped along with the task
allocation, during a panic as well. `drop_future` leaks the future while
unwinding, since dropping it could panic a second time, but leaking the
span would leave the task running in the console forever. The subscriber is
therefore reentered while unwinding, where a panic inside it aborts.

Since the span records where the task was spawned, `Executor::spawn`
becomes `#[track_caller]`. Wrappers around it want the console to blame
their own caller instead of themselves, so also add `spawn_at`, taking the
`SpawnMeta` to attribute the task to.

Adding `tracing` to the workspace also lets `compio-log` take it from there
rather than pinning a version of its own.

* feat(executor): emit runtime::waker events

Emit a `runtime::waker` event from every waker operation of a task, which
is what the console needs to report waker counts and to run its self-wake
and lost-waker lints.

The `op` values the console expects are collected in the `console` module,
next to a note on why `Waker::wake` must not report a drop of its own.

* feat(executor): instrument the future blocked on

The future passed to `block_on` is driven by the runtime instead of being
a task of the executor, so it is invisible to the console although it is
usually the most interesting future of the application.

Add `console::instrument_block_on`, wrapping it into a future that owns a
`block_on` task span. Its waker belongs to the caller of `block_on` rather
than to a task, so wrap it too, in a shim reporting the waker operations
the console expects.

Without the `console` feature the wrapper is the identity function.

* test(executor): assert the console instrumentation

Record the spans and events with a subscriber doing what
`console-subscriber` does, and assert on what it saw: the fields of the
task spans, the poll counts, and that the waker operations of a task
balance out, which is what the console's lost-waker lint looks at.

Run the new test in CI, under miri as well, since it exercises the waker
vtables of both the tasks and the `block_on` shim.

* feat(executor): report blocking closures as blocking tasks

The console treats tasks whose `kind` is `blocking` or `block_on` as not
being driven by a future, and skips the four lints that only make sense for
one: self-wake ratio, lost waker, never-yielded and large future.

`spawn_blocking` produces exactly the kind of task those lints misjudge, and
reports the wrong times on top of that: the task is the future waiting for
the pool, so all of the time in the closure counts as idle and none as busy.

Instrument the closure instead of the future waiting for it. The span is
created on the spawning thread, so the wait for a worker is reported as idle
time, and entered around the closure, so its time is reported as busy. The
future is then left unreported, since it stands for work that is already
accounted for.

* feat(executor): let a task be named

The console gives `task.name` a column of its own, and leaves it empty for
the tasks that do not have it. It is worth setting for the tasks a user did
not spawn themselves, since the location of those points into compio rather
than at the code that asked for the work.

That is the case for every wrapper around `spawn` that is an `async fn`,
since `#[track_caller]` is a no-op on those and a `SpawnMeta` cannot be
forwarded through them, so note that in the limitations as well.

Add `SpawnMeta::named` for the tasks that can make up for it. The crates
that spawn tasks of their own name them in the commits that follow, one
per crate.

* test(executor): assert the console variants present one surface

The `console` module has an enabled and a disabled variant of every type it
exports, and only one of them is ever compiled. The disabled one is what
nearly every build uses, so a difference between the two surfaces reaches
whoever turns the feature on, in code written long after the difference.

Assert that they match, by coercing each item to a function pointer, which
pins its whole signature, and by naming the traits the rest of the crate
relies on.

The guard returned by entering a task's span needs more than a signature:
the enabled one borrows the span, since the console measures the time the
span is entered as the busy time of the task. A disabled guard that owned
itself would let code hold it past the span it is timing and compile, and
only fail once the feature is turned on. Both variants therefore name the
guard through a `EnterGuard<'a>` alias, which an owned guard cannot fill.

* feat(executor): let the task a runtime blocks on be named

`instrument_block_on` captured the caller itself, so a `block_on` task could
only ever be reported by the location of the call. That is enough for the
runtime a user blocks on themselves, but not for the ones started on their
behalf, which all report the same line inside compio.

Take a `SpawnMeta` like the other spawns do, so that a caller can pass one.

* docs(executor): correct what the console does with a task's kind

The console does not treat every kind other than `task` as one it does not
drive itself: it knows `blocking` and `block_on` by name, and lints a task
of any other kind, including one it does not know, as a future of its own.

Record the nightly escape hatch for the attribution of an `async fn` too.

* feat(executor): instrument the future a compatibility layer executes

`compio-compat`'s `execute` drives the executor from a foreign event loop,
the way `block_on` drives it from a loop of its own. It is the same kind of
task, so report it as one, under a name that says what it instruments.

Author: Xerxes-2

Cargo.toml

@@ -72,6 +72,7 @@ slab = "0.4.9"
 socket2 = "0.6.0"
 tempfile = "3.8.1"
 tokio = "1.33.0"
+tracing = { version = "0.1.41", default-features = false, features = ["std"] }
 tracing-subscriber = "0.3.18"
 webpki-roots = "1.0.0"
 widestring = "1.0.2"

compio-executor/Cargo.toml

@@ -20,11 +20,15 @@ compio-send-wrapper = { workspace = true }
 crossbeam-queue = { workspace = true }
 slotmap = { workspace = true }
 
+# Console
+tracing = { workspace = true, optional = true }
+
 [target.'cfg(loom)'.dependencies]
 loom = { version = "0.7", features = ["checkpoint"] }
 
 [dev-dependencies]
 criterion = { workspace = true }
+tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 
 [target.'cfg(unix)'.dev-dependencies]
@@ -33,6 +37,9 @@ nix = { workspace = true, features = ["resource", "signal"] }
 [features]
 enable_log = ["compio-log/enable_log"]
 
+# Instrumentation for `tokio-console`. See the `console` module.
+console = ["dep:tracing"]
+
 [[bench]]
 name = "schedule"
 harness = false

compio-executor/src/console.rs

@@ -0,0 +1,169 @@
+//! [`tokio-console`] instrumentation.
+//!
+//! [`tokio-console`] collects its data through [`tracing`] spans and events
+//! that follow a fixed naming convention. It is *not* tied to tokio's internals
+//! in any way, so any executor emitting the same spans and events can be
+//! observed with it.
+//!
+//! Enable the `console` feature to make this executor emit them:
+//!
+//! * every task gets a `runtime.spawn` span, entered while the task is polled,
+//!   so that the console can compute poll counts, busy/idle/scheduled times and
+//!   the poll time histogram;
+//! * every waker operation emits a `runtime::waker` event, so that the console
+//!   can compute waker counts and detect self-wakes and lost wakers;
+//! * a closure handed to the blocking pool gets such a span too, entered around
+//!   the closure instead of around a poll, so that the time spent in it is
+//!   reported as busy time rather than as idle time.
+//!
+//! When the feature is disabled, all of this compiles down to nothing: the
+//! types in this module become zero-sized and every method an empty inlined
+//! function.
+//!
+//! # Usage
+//!
+//! `console-subscriber` refuses to run unless it can prove that the runtime is
+//! instrumented, which for tokio means the `tokio_unstable` cfg. For other
+//! runtimes it provides the `console_without_tokio_unstable` escape hatch, so a
+//! binary observing compio needs:
+//!
+//! ```toml
+//! # .cargo/config.toml
+//! [build]
+//! rustflags = ["--cfg", "console_without_tokio_unstable"]
+//! ```
+//!
+//! Depending on `console-subscriber` and installing it is then all it takes:
+//!
+//! ```ignore
+//! console_subscriber::init();
+//! compio::runtime::Runtime::new().unwrap().block_on(async {
+//!     // ...
+//! });
+//! ```
+//!
+//! # Limitations
+//!
+//! * The console's data model has one runtime per process, while compio is
+//!   thread-per-core and has one executor per thread. The tasks of all of them
+//!   are listed together; the `thread` field tells them apart.
+//! * The subscriber has to be the global default, which
+//!   `console_subscriber::init` makes it. A span carries the subscriber it was
+//!   created with, but an event goes to whichever one is current on the thread
+//!   emitting it, so a thread-local subscriber misses the waker operations
+//!   other threads perform. Wakers cross threads routinely — that is what
+//!   waking a task from another executor is — and the clone and drop counts of
+//!   one that does no longer balance, leaving the console to report a lost
+//!   waker that is not lost.
+//! * A `block_on` nested inside a task — a runtime built within another one —
+//!   reports the two as separate tasks, but both of their spans are entered on
+//!   the same stack. The console attributes the polls to the inner one for as
+//!   long as that is the case.
+//! * A blocking task has no waker operations, since it is a closure rather than
+//!   a future. The console knows this from its `kind` and does not report a
+//!   lost waker for it.
+//! * A task spawned by an `async fn` is attributed to that function rather than
+//!   to its caller, since [`#[track_caller]`][async-track-caller] is a no-op on
+//!   `async fn`s and [`SpawnMeta`] therefore cannot be forwarded through them.
+//!   The ones compio spawns itself are named to make up for it. A crate willing
+//!   to build on nightly can lift this for its own `async fn`s with the
+//!   `async_fn_track_caller` feature, which attributes them to the `.await` of
+//!   the future they return.
+//! * The resources tab stays empty: timers and in-flight operations are not
+//!   instrumented yet.
+//! * A task's span is closed even when the thread is unwinding, or the console
+//!   would show the task as running forever. The subscriber therefore runs
+//!   during a panic, where a panic of its own aborts instead of unwinding.
+//!
+//! [`tokio-console`]: https://github.com/tokio-rs/console
+//! [`tracing`]: https://docs.rs/tracing
+//! [async-track-caller]: https://github.com/rust-lang/rust/issues/110011
+
+cfg_select! {
+    feature = "console" => {
+        mod enabled;
+        use enabled as imp;
+    }
+    _ => {
+        mod disabled;
+        use disabled as imp;
+    }
+}
+
+pub(crate) use imp::TaskSpan;
+pub use imp::{SpawnMeta, instrument_block_on, instrument_blocking, instrument_execute};
+
+/// An operation on a task's waker, reported as a `runtime::waker` event.
+///
+/// Note that [`Waker::wake`](std::task::Waker::wake) does not call the `drop`
+/// implementation, so the console counts [`Self::Wake`] as both a wake and a
+/// drop. Emitting an additional [`Self::Drop`] for it would make the live waker
+/// count (clones - drops) go negative.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum WakerOp {
+    Clone,
+    Drop,
+    Wake,
+    WakeByRef,
+}
+
+impl WakerOp {
+    /// The `op` value of the event, as expected by the console.
+    ///
+    /// Only the enabled variant reports anything, so only it reads this.
+    #[cfg(feature = "console")]
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Clone => "waker.clone",
+            Self::Drop => "waker.drop",
+            Self::Wake => "waker.wake",
+            Self::WakeByRef => "waker.wake_by_ref",
+        }
+    }
+}
+/// Assertions that the two variants above present the same surface.
+///
+/// Only one of them is ever compiled, and the one compiled by default is the
+/// one nearly every build uses: a difference between the two shows up as a
+/// build failure for whoever turns the feature on, long after the code that
+/// assumed the other shape was written.
+///
+/// Coercing each item to a function pointer pins its whole signature, and
+/// naming [`EnterGuard`] with a lifetime pins the shape of the guard: the
+/// enabled one borrows the span, so a disabled one that owns itself, and would
+/// let code outlive the span it is timing, does not have a lifetime to name.
+#[cfg(test)]
+mod parity {
+    use std::{fmt::Debug, future::Future};
+
+    use super::{imp::EnterGuard, *};
+
+    const _: fn() -> SpawnMeta = SpawnMeta::capture;
+    const _: fn(SpawnMeta, &'static str) -> SpawnMeta = SpawnMeta::named;
+    const _: fn() -> SpawnMeta = SpawnMeta::untracked;
+
+    const _: fn(SpawnMeta) -> TaskSpan = TaskSpan::new::<()>;
+    const _: for<'a> fn(&'a TaskSpan) -> EnterGuard<'a> = TaskSpan::enter;
+    const _: fn(&TaskSpan, WakerOp) = TaskSpan::waker_op;
+
+    /// [`SpawnMeta`] is copied out of a spawn call rather than moved, and
+    /// reaches the dispatcher's threads through its channel.
+    const fn meta<T: Copy + Send + Sync + Unpin + Debug + 'static>() {}
+    const _: () = meta::<SpawnMeta>();
+
+    /// [`TaskSpan`] sits in the task header, which threads share.
+    const fn span<T: Send + Sync + Debug>() {}
+    const _: () = span::<TaskSpan>();
+
+    /// The wrappers return `impl Trait`, so pin them by use instead.
+    #[test]
+    fn the_wrappers_pass_their_argument_through() {
+        assert_eq!(instrument_blocking(SpawnMeta::untracked(), || 1u8)(), 1);
+
+        let fut = instrument_block_on(SpawnMeta::untracked(), std::future::ready(1u8));
+        let _: &dyn Future<Output = u8> = &fut;
+
+        let fut = instrument_execute(SpawnMeta::untracked(), std::future::ready(1u8));
+        let _: &dyn Future<Output = u8> = &fut;
+    }
+}

compio-executor/src/console/disabled.rs

@@ -0,0 +1,107 @@
+//! No-op stand-ins used when the `console` feature is disabled.
+
+use std::marker::PhantomData;
+
+/// Metadata of a spawned task, reported to the console.
+///
+/// Zero-sized and inert unless the `console` feature is enabled.
+#[derive(Debug, Clone, Copy)]
+pub struct SpawnMeta;
+
+impl SpawnMeta {
+    /// Capture the location of the caller.
+    ///
+    /// Discards it. The spawns that call this stay `#[track_caller]` either
+    /// way: a feature of this crate cannot reach the wrappers in the crates
+    /// that depend on it, so gating them on one of their own would attribute
+    /// every task to compio itself in a build that enables only this one. The
+    /// implicit argument is dead here, and mostly optimised away.
+    #[inline(always)]
+    pub fn capture() -> Self {
+        Self
+    }
+
+    /// Name the task, which the console displays in a column of its own.
+    #[inline(always)]
+    pub fn named(self, _name: &'static str) -> Self {
+        self
+    }
+
+    /// Do not report the task to the console at all.
+    #[inline(always)]
+    pub fn untracked() -> Self {
+        Self
+    }
+}
+
+/// The guard returned by [`TaskSpan::enter`].
+///
+/// The enabled variant measures the busy time of a task as the time its guard
+/// is alive, so dropping it right away is a bug that this makes visible in both
+/// configurations. It borrows the span for the same reason: the enabled guard
+/// does, and code that compiles without the feature has to compile with it.
+#[must_use = "the task span is exited as soon as this is dropped"]
+pub(crate) struct Entered<'a>(PhantomData<&'a TaskSpan>);
+
+/// The guard [`TaskSpan::enter`] returns, named the same in both variants so
+/// that the parity assertions can reach it.
+pub(crate) type EnterGuard<'a> = Entered<'a>;
+
+/// The `runtime.spawn` span of a task.
+#[derive(Debug)]
+pub(crate) struct TaskSpan;
+
+impl TaskSpan {
+    #[inline(always)]
+    #[expect(
+        clippy::extra_unused_type_parameters,
+        reason = "mirrors the enabled variant, which records the future's size"
+    )]
+    pub(crate) fn new<F>(_meta: SpawnMeta) -> Self {
+        Self
+    }
+
+    #[inline(always)]
+    pub(crate) fn enter(&self) -> EnterGuard<'_> {
+        Entered(PhantomData)
+    }
+
+    #[inline(always)]
+    pub(crate) fn waker_op(&self, _op: super::WakerOp) {}
+}
+
+/// Instrument a closure about to be handed to the blocking pool, so that it
+/// shows up as a blocking task in the console.
+///
+/// This is a no-op unless the `console` feature is enabled.
+///
+/// Plumbing for `compio-runtime`, not covered by this crate's semver.
+#[doc(hidden)]
+#[inline(always)]
+pub fn instrument_blocking<T, F: FnOnce() -> T>(_meta: SpawnMeta, f: F) -> impl FnOnce() -> T {
+    f
+}
+
+/// Instrument a future blocked on by the runtime, so that it shows up as a
+/// task in the console.
+///
+/// This is a no-op unless the `console` feature is enabled.
+///
+/// Plumbing for `compio-runtime`, not covered by this crate's semver.
+#[doc(hidden)]
+#[inline(always)]
+pub fn instrument_block_on<F: Future>(_meta: SpawnMeta, fut: F) -> impl Future<Output = F::Output> {
+    fut
+}
+
+/// Instrument a future executed by a compatibility layer, so that it shows up
+/// as a task in the console.
+///
+/// This is a no-op unless the `console` feature is enabled.
+///
+/// Plumbing for `compio-compat`, not covered by this crate's semver.
+#[doc(hidden)]
+#[inline(always)]
+pub fn instrument_execute<F: Future>(_meta: SpawnMeta, fut: F) -> impl Future<Output = F::Output> {
+    fut
+}