feature: add resource detail view

Cargo.toml

@@ -4,4 +4,9 @@ members = [
     "console-subscriber",
     "console-api"
 ]
-resolver = "2"
+resolver = "2"
+
+# Patch the dependency on tokio to get Resource details instrumentation
+[patch.crates-io.tokio]
+git = "https://github.com/tokio-rs/tokio"
+rev = "88ebd0361e760e68aca705e40ee987c77b9f7679"

console-api/proto/async_ops.proto

@@ -37,6 +37,17 @@ message AsyncOp {
     // The source of this async operation. Most commonly this should be the name
     // of the method where the instantiation of this op has happened.
     string source = 3;
+    // The ID of the parent async op.
+    //
+    // This field is only set if this async op was created while inside of another
+    // async op.  For example, `tokio::sync`'s `Mutex::lock` internally calls
+    // `Semaphore::acquire`.
+    //
+    // This field can be empty; if it is empty, this async op is not a child of another
+    // async op.
+    common.Id parent_async_op_id = 4;
+    // The resources's ID.
+    common.Id resource_id = 5;
 }
 
 
@@ -45,12 +56,11 @@ message Stats {
     google.protobuf.Timestamp created_at = 1;
     // Timestamp of when the async op was dropped.
     google.protobuf.Timestamp dropped_at = 2;
-    // The resource Id this `AsyncOp` is associated with. Note that both
-    // `resource_id` and `task_id` can be None if this async op has not been polled yet
-    common.Id resource_id = 3;
     // The Id of the task that is awaiting on this op.
     common.Id task_id = 4;
     // Contains the operation poll stats.
     common.PollStats poll_stats = 5;
+    // State attributes of the async op.
+    repeated common.Attribute attributes = 6;
 }
 

console-api/proto/common.proto

@@ -124,3 +124,15 @@ message PollStats {
     // has spent *waiting* to be polled.
     google.protobuf.Duration busy_time = 6;
 }
+
+// State attributes of an entity. These are dependent on the type of the entity.
+//
+// For example, a timer resource will have a duration, while a semaphore resource may
+// have a permit count. Likewise, the async ops of a semaphore may have attributes
+// indicating how many permits they are trying to acquire vs how many are acquired.
+// These values may change over time. Therefore, they live in the runtime stats rather
+// than the static data describing the entity.
+message Attribute {
+    Field field = 1;
+    optional string unit = 2;
+}

console-api/proto/resources.proto

@@ -41,6 +41,9 @@ message Resource {
     Kind kind = 4;
     // The location in code where the resource was created.
     common.Location location = 5;
+    // The ID of the parent resource.
+    common.Id parent_resource_id = 6;
+    bool is_internal = 7;
 
     message Kind {
         oneof kind {
@@ -59,17 +62,8 @@ message Stats {
     google.protobuf.Timestamp created_at = 1;
     // Timestamp of when the resource was dropped.
     google.protobuf.Timestamp dropped_at = 2;
-    // State attributes of the resource. These are dependent on the type of the resource.
-    // For example, a timer resource will have a duration while a semaphore resource may
-    // have permits as an attribute. These values may change over time as the state of
-    // the resource changes. Therefore, they live in the runtime stats rather than the
-    // static data describing the resource.
-    repeated Attribute attributes = 3;
-
-    message Attribute {
-        common.Field field = 1;
-        optional string unit = 2;
-    }
+    // State attributes of the resource.
+    repeated common.Attribute attributes = 3;
 }
 
 message PollOp {

console-subscriber/Cargo.toml

@@ -11,7 +11,7 @@ parking_lot = ["parking_lot_crate", "tracing-subscriber/parking_lot"]
 
 [dependencies]
 
-tokio = { version = "^1.13", features = ["sync", "time", "macros", "tracing"] }
+tokio = { version = "^1.14", features = ["sync", "time", "macros", "tracing"] }
 tokio-stream = "0.1"
 thread_local = "1.1.3"
 console-api = { path = "../console-api", features = ["transport"] }

console-subscriber/examples/barrier.rs

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Barrier;
+use tokio::task;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    console_subscriber::init();
+    task::Builder::default()
+        .name("main-task")
+        .spawn(async move {
+            let mut handles = Vec::with_capacity(30);
+            let barrier = Arc::new(Barrier::new(30));
+            for i in 0..30 {
+                let c = barrier.clone();
+                let task_name = format!("task-{}", i);
+                handles.push(task::Builder::default().name(&task_name).spawn(async move {
+                    tokio::time::sleep(Duration::from_secs(i)).await;
+                    let wait_result = c.wait().await;
+                    wait_result
+                }));
+            }
+
+            // Will not resolve until all "after wait" messages have been printed
+            let mut num_leaders = 0;
+            for handle in handles {
+                let wait_result = handle.await.unwrap();
+                if wait_result.is_leader() {
+                    num_leaders += 1;
+                }
+            }
+
+            tokio::time::sleep(Duration::from_secs(10)).await;
+            // Exactly one barrier will resolve as the "leader"
+            assert_eq!(num_leaders, 1);
+        })
+        .await?;
+
+    Ok(())
+}

console-subscriber/examples/mutex.rs

@@ -0,0 +1,31 @@
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::{sync::Mutex, task};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    console_subscriber::init();
+    task::Builder::default()
+        .name("main-task")
+        .spawn(async move {
+            let count = Arc::new(Mutex::new(0));
+            for i in 0..5 {
+                let my_count = Arc::clone(&count);
+                let task_name = format!("increment-{}", i);
+                tokio::task::Builder::default()
+                    .name(&task_name)
+                    .spawn(async move {
+                        for _ in 0..10 {
+                            let mut lock = my_count.lock().await;
+                            *lock += 1;
+                            tokio::time::sleep(Duration::from_secs(1)).await;
+                        }
+                    });
+            }
+
+            while *count.lock().await < 50 {}
+        })
+        .await?;
+
+    Ok(())
+}

console-subscriber/examples/rwlock.rs

@@ -0,0 +1,37 @@
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::{sync::RwLock, task};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    console_subscriber::init();
+    task::Builder::default()
+        .name("main-task")
+        .spawn(async move {
+            let count = Arc::new(RwLock::new(0));
+            for i in 0..5 {
+                let my_count = Arc::clone(&count);
+                let task_name = format!("increment-{}", i);
+                tokio::task::Builder::default()
+                    .name(&task_name)
+                    .spawn(async move {
+                        for _ in 0..10 {
+                            let mut lock = my_count.write().await;
+                            *lock += 1;
+                            tokio::time::sleep(Duration::from_secs(1)).await;
+                        }
+                    });
+            }
+
+            loop {
+                let c = count.read().await;
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                if *c >= 50 {
+                    break;
+                }
+            }
+        })
+        .await?;
+
+    Ok(())
+}

console-subscriber/examples/semaphore.rs

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::task;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    console_subscriber::init();
+    task::Builder::default()
+        .name("main-task")
+        .spawn(async move {
+            let sem = Arc::new(tokio::sync::Semaphore::new(0));
+            let mut tasks = Vec::default();
+            for i in 0..5 {
+                let acquire_sem = Arc::clone(&sem);
+                let add_sem = Arc::clone(&sem);
+                let acquire_task_name = format!("acquire-{}", i);
+                let add_task_name = format!("add-{}", i);
+                tasks.push(
+                    tokio::task::Builder::default()
+                        .name(&acquire_task_name)
+                        .spawn(async move {
+                            let _permit = acquire_sem.acquire_many(i).await.unwrap();
+                            tokio::time::sleep(Duration::from_secs(i as u64 * 2)).await;
+                        }),
+                );
+                tasks.push(tokio::task::Builder::default().name(&add_task_name).spawn(
+                    async move {
+                        tokio::time::sleep(Duration::from_secs(i as u64 * 5)).await;
+                        add_sem.add_permits(i as usize);
+                    },
+                ));
+            }
+
+            let all_tasks = futures::future::try_join_all(tasks);
+            all_tasks.await.unwrap();
+        })
+        .await?;
+
+    Ok(())
+}