Merge branch 'main' into refactor-resolve-borrow

Author: jlizen

AGENTS.md

@@ -26,3 +26,7 @@ The demo trace is used for:
 - Testing the viewer with real data
 
 Failing to update it will cause the viewer to fail when loading the demo.
+
+## Meta
+
+- Never declare done after pushing or opening a PR until CI is green. Check CI status and fix any failures before moving on.

Cargo.lock

@@ -21,6 +21,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 smallvec = "1"
 dial9-perf-self-profile = { workspace = true, optional = true }
+tracing = "0.1.44"
 
 [features]
 cpu-profiling = ["dep:dial9-perf-self-profile"]

dial9-tokio-telemetry/Cargo.toml

@@ -19,7 +19,7 @@ fn main() -> std::io::Result<()> {
     let mut builder = tokio::runtime::Builder::new_multi_thread();
     builder.worker_threads(4).enable_all();
 
-    let (runtime, _guard) = TracedRuntime::build_and_start(builder, Box::new(writer))?;
+    let (runtime, _guard) = TracedRuntime::build_and_start(builder, writer)?;
 
     runtime.block_on(async {
         // your async code here
@@ -70,7 +70,7 @@ Use `handle.spawn()` instead of `tokio::spawn()`:
 # fn main() -> std::io::Result<()> {
 # let writer = RotatingWriter::new("/tmp/t.bin", 1024, 4096)?;
 # let builder = tokio::runtime::Builder::new_multi_thread();
-let (runtime, guard) = TracedRuntime::build_and_start(builder, Box::new(writer))?;
+let (runtime, guard) = TracedRuntime::build_and_start(builder, writer)?;
 let handle = guard.handle();
 
 runtime.block_on(async {
@@ -100,7 +100,7 @@ On non-Linux platforms these fields are zero.
 
 With the `cpu-profiling` feature, you can enable `perf_event_open`-based CPU sampling. This gives two key pieces of data:
 1. Stack traces when code was running on the CPU — aka flamegraphs
-2. 2. Stack traces when the kernel _descheduled_ your thread. For example, if you use `std::thread::sleep` in your future or are seeing `std::sync::Mutex` contention, this will allow you to see precisely where this is happening in async code.
+2. Stack traces when the kernel _descheduled_ your thread. For example, if you use `std::thread::sleep` in your future or are seeing `std::sync::Mutex` contention, this will allow you to see precisely where this is happening in async code.
 
 Both of these events are tied to the precise instant and thread that they happened on, so you can compare what was different between degraded and normal performance.
 
@@ -117,7 +117,7 @@ let (runtime, guard) = TracedRuntime::builder()
     .with_cpu_profiling(CpuProfilingConfig::default())
     .with_sched_events(SchedEventConfig { include_kernel: true })
     .with_inline_callframe_symbols(true)
-    .build(builder, Box::new(writer))?;
+    .build_and_start(builder, writer)?;
 # Ok(())
 # }
 # #[cfg(not(feature = "cpu-profiling"))]
@@ -126,6 +126,30 @@ let (runtime, guard) = TracedRuntime::builder()
 
 This pulls in [`dial9-perf-self-profile`](/perf-self-profile) for `perf_event_open` access. It records `CpuSample` events with full callchains and `CallframeDef` / `ThreadNameDef` metadata for offline symbolization.
 
+#### Requirements
+
+**Frame pointers**: CPU profile stack traces rely on frame-pointer-based unwinding. Compile your application with frame pointers enabled, otherwise stack traces will be truncated or missing:
+
+```toml
+# .cargo/config.toml
+[build]
+rustflags = ["-C", "force-frame-pointers=yes"]
+```
+
+**`perf_event_paranoid`**: CPU profiling features require `perf_event_paranoid` ≤ 2 for sampling, and ≤ 1 for scheduler event tracking (`with_sched_events`):
+
+```bash
+# check current value
+cat /proc/sys/kernel/perf_event_paranoid
+
+# allow CPU sampling and scheduler event tracking
+sudo sysctl kernel.perf_event_paranoid=1
+```
+
+#### Diagnosing long polls with CPU samples
+
+Because CPU samples are tagged with the worker thread they were collected on, and the trace records which task is being polled on each worker at each instant, the viewer can correlate samples with individual polls. When a poll takes an unusually long time (a "long poll"), the CPU samples collected during that poll show you exactly what code was running — expensive serialization, accidental blocking I/O, lock contention, etc. In the trace viewer, click on a long poll to see its flamegraph, or shift+drag to aggregate CPU samples across a time range.
+
 ## Getting started
 
 `TracedRuntime::build` returns a `(Runtime, TelemetryGuard)`. The guard owns the flush thread and provides a `TelemetryHandle` for enabling/disabling recording at runtime:
@@ -137,7 +161,7 @@ This pulls in [`dial9-perf-self-profile`](/perf-self-profile) for `perf_event_op
 # let builder = tokio::runtime::Builder::new_multi_thread();
 let (runtime, guard) = TracedRuntime::builder()
     .with_task_tracking(true)
-    .build(builder, Box::new(writer))?;
+    .build(builder, writer)?;
 
 // start disabled, enable later
 guard.enable();
@@ -151,7 +175,7 @@ handle.disable();
 
 ### Writers
 
-`RotatingWriter` rotates files and evicts old ones to stay within a total size budget. `SimpleBinaryWriter` writes a single file with no size management, useful for quick experiments.
+`RotatingWriter` rotates files and evicts old ones to stay within a total size budget. For quick experiments, `RotatingWriter::single_file(path)` writes a single file with no rotation.
 
 ### Analyzing traces
 

dial9-tokio-telemetry/README.md

@@ -16,7 +16,7 @@
 #[cfg(target_os = "linux")]
 use dial9_tokio_telemetry::telemetry::CpuProfilingConfig;
 use dial9_tokio_telemetry::telemetry::{
-    NullWriter, SimpleBinaryWriter, TelemetryGuard, TelemetryHandle, TracedRuntime,
+    NullWriter, RotatingWriter, TelemetryGuard, TelemetryHandle, TracedRuntime,
 };
 use hdrhistogram::Histogram;
 use std::sync::Arc;
@@ -172,8 +172,7 @@ fn main() {
 
     let (server_rt, _guard): (tokio::runtime::Runtime, Option<TelemetryGuard>) = match mode {
         "telemetry" => {
-            let writer =
-                Box::new(SimpleBinaryWriter::new("/tmp/overhead_bench_trace.bin").unwrap());
+            let writer = RotatingWriter::single_file("/tmp/overhead_bench_trace.bin").unwrap();
             let mut tb = TracedRuntime::builder().with_task_tracking(true);
             #[cfg(target_os = "linux")]
             {
@@ -186,7 +185,7 @@ fn main() {
         }
         "noop" => {
             let (rt, g) = TracedRuntime::builder()
-                .build_and_start(builder, Box::new(NullWriter))
+                .build_and_start(builder, NullWriter)
                 .unwrap();
             (rt, Some(g))
         }

dial9-tokio-telemetry/benches/overhead_bench.rs

@@ -167,7 +167,7 @@ Simple wrapper combining event type with metrics snapshot.
 - `Send` bound for thread safety
 - Batch write method for efficiency
 - Explicit flush control
-- Implementations: `SimpleBinaryWriter`, `RotatingWriter`, `NullWriter` (for benchmarking)
+- Implementations: `RotatingWriter`, `NullWriter` (for benchmarking)
 
 **Binary Format (v5):**
 Variable-size records optimize for the common case:
@@ -238,7 +238,7 @@ Variable-size records optimize for the common case:
 **Usage Pattern:**
 ```rust
 let writer = RotatingWriter::new("trace.bin", 1_MB, 5_MB)?;
-let (runtime, guard) = TracedRuntime::build(builder, Box::new(writer))?;
+let (runtime, guard) = TracedRuntime::build(builder, writer)?;
 
 // Use runtime normally
 runtime.block_on(async { /* ... */ });
@@ -387,7 +387,7 @@ src/
     ├── events.rs             # EventType, MetricsSnapshot, TelemetryEvent, CPU time helpers
     ├── buffer.rs             # ThreadLocalBuffer
     ├── collector.rs          # CentralCollector (Mutex-based)
-    ├── writer.rs             # TraceWriter trait, SimpleBinaryWriter, RotatingWriter, NullWriter
+    ├── writer.rs             # TraceWriter trait, RotatingWriter, NullWriter
     ├── recorder.rs           # TelemetryRecorder, TracedRuntime, TelemetryGuard
     ├── format.rs             # Binary format v5 (variable-size records)
     └── analysis.rs           # TraceReader, analyze_trace, detect_idle_workers, compute_active_periods

dial9-tokio-telemetry/design/tokio-telemetry-system.md

@@ -0,0 +1,60 @@
+//! Example demonstrating blocking pool tracking.
+//!
+//! Spawns work on tokio's blocking pool via `spawn_blocking`, then writes a
+//! trace. Blocking pool samples should appear with `worker_id = 254`
+//! (BLOCKING_WORKER) instead of 255 (UNKNOWN_WORKER).
+//!
+//! Usage:
+//!   cargo run --release --features cpu-profiling --example blocking_pool_tracking
+
+use dial9_tokio_telemetry::telemetry::{CpuProfilingConfig, RotatingWriter, TracedRuntime};
+use std::time::Duration;
+
+fn burn_cpu(duration: Duration) {
+    let start = std::time::Instant::now();
+    let mut x: u64 = 1;
+    while start.elapsed() < duration {
+        for _ in 0..1000 {
+            x = x.wrapping_mul(6364136223846793005).wrapping_add(1);
+        }
+        std::hint::black_box(x);
+    }
+}
+
+fn main() {
+    let mut builder = tokio::runtime::Builder::new_multi_thread();
+    builder.worker_threads(2).enable_all();
+
+    let writer = RotatingWriter::single_file("blocking_pool_trace.bin").unwrap();
+    let (runtime, _guard) = TracedRuntime::builder()
+        .with_task_tracking(true)
+        .with_cpu_profiling(CpuProfilingConfig {
+            frequency_hz: 999,
+            ..Default::default()
+        })
+        .with_inline_callframe_symbols(true)
+        .build_and_start(builder, writer)
+        .unwrap();
+
+    runtime.block_on(async {
+        // Spawn CPU-intensive work on the blocking pool
+        let handles: Vec<_> = (0..3)
+            .map(|i| {
+                tokio::task::spawn_blocking(move || {
+                    eprintln!("blocking task {i} starting");
+                    burn_cpu(Duration::from_millis(500));
+                    eprintln!("blocking task {i} done");
+                })
+            })
+            .collect();
+
+        for h in handles {
+            h.await.unwrap();
+        }
+
+        // Let flush cycle capture everything
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    });
+
+    eprintln!("Trace written to blocking_pool_trace.bin");
+}

dial9-tokio-telemetry/examples/blocking_pool_tracking.rs

@@ -1,4 +1,4 @@
-use dial9_tokio_telemetry::telemetry::{SimpleBinaryWriter, TracedRuntime};
+use dial9_tokio_telemetry::telemetry::{RotatingWriter, TracedRuntime};
 use std::time::Duration;
 
 async fn blocking_task(id: usize) {
@@ -14,7 +14,7 @@ fn main() {
     let mut builder = tokio::runtime::Builder::new_multi_thread();
     builder.worker_threads(2).enable_all();
 
-    let writer = Box::new(SimpleBinaryWriter::new("blocking_sleep_trace.bin").unwrap());
+    let writer = RotatingWriter::single_file("blocking_sleep_trace.bin").unwrap();
     let (runtime, _guard) = TracedRuntime::builder()
         .with_task_tracking(true)
         .with_cpu_profiling(Default::default())

dial9-tokio-telemetry/examples/blocking_sleep.rs

@@ -10,7 +10,7 @@
 //!   echo 2 | sudo tee /proc/sys/kernel/perf_event_paranoid
 
 use dial9_tokio_telemetry::telemetry::{
-    CpuProfilingConfig, SimpleBinaryWriter, TelemetryEvent, TracedRuntime,
+    CpuProfilingConfig, RotatingWriter, TelemetryEvent, TracedRuntime,
 };
 use std::time::Duration;
 
@@ -38,7 +38,7 @@ fn main() {
     let mut builder = tokio::runtime::Builder::new_multi_thread();
     builder.worker_threads(4).enable_all();
 
-    let writer = Box::new(SimpleBinaryWriter::new(trace_path).unwrap());
+    let writer = RotatingWriter::single_file(trace_path).unwrap();
     let (runtime, guard) = TracedRuntime::builder()
         .with_task_tracking(true)
         .with_cpu_profiling(CpuProfilingConfig::default())

dial9-tokio-telemetry/examples/cpu_profile_workload.rs

@@ -1,4 +1,4 @@
-use dial9_tokio_telemetry::telemetry::{SimpleBinaryWriter, TracedRuntime};
+use dial9_tokio_telemetry::telemetry::{RotatingWriter, TracedRuntime};
 use std::time::Duration;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
@@ -83,7 +83,7 @@ fn main() {
         .and_then(|s| s.parse().ok())
         .unwrap_or(30u64);
 
-    let writer = Box::new(SimpleBinaryWriter::new("long_trace.bin").unwrap());
+    let writer = RotatingWriter::single_file("long_trace.bin").unwrap();
     let (runtime, _guard) = TracedRuntime::builder()
         .build_and_start(builder, writer)
         .unwrap();