Implement minimal (metrics-only) taskdumps.

examples/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2018"
 # If you copy one of the examples into a new project, you should be using
 # [dependencies] instead, and delete the **path**.
 [dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full", "tracing"] }
+tokio = { version = "1.0.0", path = "../tokio", features = ["full", "taskdump", "tracing"] }
 tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
 tokio-stream = { version = "0.1", path = "../tokio-stream" }
 
@@ -90,3 +90,7 @@ path = "named-pipe-ready.rs"
 [[example]]
 name = "named-pipe-multi-client"
 path = "named-pipe-multi-client.rs"
+
+[[example]]
+name = "taskdump"
+path = "taskdump.rs"

examples/taskdump.rs

@@ -0,0 +1,59 @@
+//! A service that emits task dumps to stdout upon receipt of a SIGUSR1 signal.
+
+#![warn(rust_2018_idioms)]
+
+use tokio::{
+    time::{sleep, Duration},
+    runtime::Handle,
+    signal::unix::{signal, SignalKind}
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let _ = tokio::spawn(async {
+        // Listen for SIGUSR1
+        let mut stream = signal(SignalKind::user_defined1()).unwrap();
+        loop {
+            // Wait for receipt of SIGUSR1
+            stream.recv().await;
+            // Write a JSON taskdump to stdout.
+            println!("{}", Handle::current());
+        }
+    });
+
+    let sleep = sleep(Duration::from_secs(20));
+
+    tokio::select! {
+        _ = tokio::spawn(busy::work()) => {}
+        _ = tokio::spawn(busy::work()) => {}
+        _ = sleep => {/* exit the application */}
+    };
+
+    Ok(())
+}
+
+mod busy {
+    use futures::future::{BoxFuture, FutureExt};
+
+    #[inline(never)]
+    pub async fn work() {
+        loop {
+            for i in 0..u8::MAX {
+                recurse(i).await;
+            }
+        }
+    }
+
+    #[inline(never)]
+    pub fn recurse(depth: u8) -> BoxFuture<'static, ()> {
+        async move {
+            tokio::task::yield_now().await;
+            if depth == 0 {
+                return;
+            } else {
+                recurse(depth - 1).await;
+            }
+        }
+        .boxed()
+    }
+}

tokio/Cargo.toml

@@ -85,6 +85,10 @@ signal = [
   "windows-sys/Win32_System_Console",
 ]
 sync = []
+taskdump = [
+  "serde",
+  "serde_json",
+]
 test-util = ["rt", "sync", "time"]
 time = []
 
@@ -115,6 +119,12 @@ socket2 = { version = "0.4.4", optional = true, features = [ "all" ] }
 # Requires `--cfg tokio_unstable` to enable.
 [target.'cfg(tokio_unstable)'.dependencies]
 tracing = { version = "0.1.25", default-features = false, features = ["std"], optional = true } # Not in full
+serde = { version = "1.0", optional = true }
+serde_json = { version = "1.0", optional = true }
+
+[target.'cfg(tokio_unstable)'.dev-dependencies]
+jsonschema = { version = "0.16.1" }
+schemars = { version = "0.8.11" }
 
 [target.'cfg(unix)'.dependencies]
 libc = { version = "0.2.42", optional = true }

tokio/src/lib.rs

@@ -461,6 +461,12 @@ compile_error!("Tokio's build script has incorrectly detected wasm.");
 ))]
 compile_error!("Only features sync,macros,io-util,rt,time are supported on wasm.");
 
+#[cfg(all(
+    not(tokio_unstable),
+    feature = "taskdump"
+))]
+compile_error!("Feature taskdump requires tokio_unstable.");
+
 // Includes re-exports used by macros.
 //
 // This module is not intended to be part of the public API. In general, any

tokio/src/macros/cfg.rs

@@ -361,6 +361,25 @@ macro_rules! cfg_not_rt_multi_thread {
     }
 }
 
+macro_rules! cfg_taskdump {
+    ($($item:item)*) => {
+        $(
+            #[cfg(all(tokio_unstable, feature = "taskdump"))]
+            #[cfg_attr(docsrs, doc(cfg(tokio_unstable), feature = "taskdump"))]
+            $item
+        )*
+    }
+}
+
+macro_rules! cfg_not_taskdump {
+    ($($item:item)*) => {
+        $(
+            #[cfg(not(all(tokio_unstable, feature = "taskdump")))]
+            $item
+        )*
+    }
+}
+
 macro_rules! cfg_test_util {
     ($($item:item)*) => {
         $(

tokio/src/runtime/handle.rs

@@ -321,6 +321,22 @@ cfg_metrics! {
     }
 }
 
+cfg_taskdump! {
+    /// # Task Dumps
+    /// 
+    /// Display-formatting a `Handle` produces a JSON dump of the runtime's internal state.
+    /// 
+    /// This feature is in very early development, and its API and output is subject to change.
+    impl fmt::Display for Handle {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            use serde_json::to_string_pretty;
+            let serializable = super::taskdump::Runtime::from(&self);
+            let json = to_string_pretty(&serializable).map_err(|_| fmt::Error)?;
+            f.write_str(&json)
+        }
+    }
+}
+
 /// Error returned by `try_current` when no Runtime has been started
 #[derive(Debug)]
 pub struct TryCurrentError {

tokio/src/runtime/mod.rs

@@ -261,3 +261,7 @@ cfg_rt! {
     /// After thread starts / before thread stops
     type Callback = std::sync::Arc<dyn Fn() + Send + Sync>;
 }
+
+cfg_taskdump! {
+    mod taskdump;
+}

tokio/src/runtime/taskdump.rs

@@ -0,0 +1,121 @@
+//! Serializable structures for task dumps. 
+
+use super::scheduler::{self, current_thread};
+use crate::loom::sync::Arc;
+use serde::Serialize;
+
+#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
+use scheduler::multi_thread;
+
+#[cfg_attr(test, derive(schemars::JsonSchema))]
+#[derive(Serialize)]
+#[serde(tag = "flavor")]
+/// A taskdump of a Tokio runtime.
+pub(super) enum Runtime {
+    /// A taskdump of a current-thread runtime.
+    CurrentThread(CurrentThread),
+    /// A taskdump of a multi-thread runtime.
+    MultiThread(MultiThread),
+}
+
+impl Runtime {
+    pub(super) fn from(handle: &super::Handle) -> Self {
+        let metrics = handle.metrics();
+        match &handle.inner {
+            scheduler::Handle::CurrentThread(h) => {
+                Runtime::CurrentThread(CurrentThread::from(h.clone(), metrics))
+            }
+            #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
+            scheduler::Handle::MultiThread(h) => {
+                Runtime::MultiThread(MultiThread::from(h.clone(), metrics))
+            }
+        }
+    }
+}
+
+/// A taskdump of a current-thread runtime.
+#[cfg_attr(test, derive(schemars::JsonSchema))]
+#[derive(Serialize)]
+pub(super) struct CurrentThread {
+    /// Runtime metrics.
+    metrics: CurrentThreadMetrics,
+}
+
+impl CurrentThread {
+    fn from(_handle: Arc<current_thread::Handle>, metrics: super::RuntimeMetrics) -> Self {
+        Self {
+            metrics: CurrentThreadMetrics {
+                remote_schedule_count: metrics.remote_schedule_count(),
+                #[cfg(feature = "net")]
+                io_driver: Some(IODriverMetrics {
+                    fd_registered_count: metrics.io_driver_fd_registered_count(),
+                    fd_deregistered_count: metrics.io_driver_fd_deregistered_count(),
+                }),
+                #[cfg(not(feature = "net"))]
+                io_driver: None,
+            },
+        }
+    }
+}
+
+#[cfg_attr(test, derive(schemars::JsonSchema))]
+#[derive(Serialize)]
+pub(super) struct MultiThread {
+    /// Runtime metrics.
+    metrics: MultiThreadMetrics,
+}
+
+impl MultiThread {
+    pub(super) fn from(_handle: Arc<multi_thread::Handle>, metrics: super::RuntimeMetrics) -> Self {
+        Self {
+            metrics: MultiThreadMetrics {
+                num_workers: metrics.num_workers(),
+                num_blocking_threads: metrics.num_blocking_threads(),
+                num_idle_blocking_threads: metrics.num_idle_blocking_threads(),
+                remote_schedule_count: metrics.remote_schedule_count(),
+                #[cfg(feature = "net")]
+                io_driver: Some(IODriverMetrics {
+                    fd_registered_count: metrics.io_driver_fd_registered_count(),
+                    fd_deregistered_count: metrics.io_driver_fd_deregistered_count(),
+                }),
+                #[cfg(not(feature = "net"))]
+                io_driver: None,
+            },
+        }
+    }
+}
+
+/// Metrics for a current-thread runtime.
+#[cfg_attr(test, derive(schemars::JsonSchema))]
+#[derive(Serialize)]
+struct CurrentThreadMetrics {
+    /// The total number of tasks scheduled from outside of the runtime.
+    remote_schedule_count: u64,
+    /// IO driver metrics.
+    io_driver: Option<IODriverMetrics>,
+}
+
+/// Metrics for a multi-thread runtime.
+#[cfg_attr(test, derive(schemars::JsonSchema))]
+#[derive(serde::Serialize)]
+struct MultiThreadMetrics {
+    /// The number of worker threads used by the runtime.
+    num_workers: usize,
+    /// The number of additional threads spawned by the runtime.
+    num_blocking_threads: usize,
+    /// The number of idle threads, which have spawned for spawn_blocking calls.
+    num_idle_blocking_threads: usize,
+    /// The total number of tasks scheduled from outside of the runtime.
+    remote_schedule_count: u64,
+    /// IO driver metrics.
+    io_driver: Option<IODriverMetrics>,
+}
+
+#[cfg_attr(test, derive(schemars::JsonSchema))]
+#[derive(Serialize)]
+struct IODriverMetrics {
+    /// The number of file descriptors that have been registered with the runtime’s I/O driver.
+    fd_registered_count: u64,
+    /// The number of file descriptors that have been deregistered by the runtime’s I/O driver
+    fd_deregistered_count: u64,
+}

tokio/src/runtime/tests/mod.rs

@@ -70,4 +70,8 @@ cfg_not_loom! {
 
     #[cfg(miri)]
     mod task;
+
+    cfg_taskdump! {
+        mod taskdump;
+    }
 }

tokio/src/runtime/tests/queue.rs

@@ -141,7 +141,7 @@ fn stress1() {
             }
 
             cfg_metrics! {
-                assert_metrics!(metrics, steal_count == n as _);
+                assert_metrics!(metrics, steal_count == n as u64);
             }
 
             n