fix: eliminate task dump busy loop and move dl_iterate_phdr off hot path

Two fixes for task dump overhead:

1. Add `just_captured` flag to break the capture→wake→poll→capture busy
   loop. The `trace_with` re-poll registers the real waker which fires
   immediately, causing infinite re-scheduling. Now we skip capture on
   the spurious wake, limiting overhead to one extra poll per capture.

2. Split stack walking into two phases:
   - Capture (hot path): collect raw IPs via `_Unwind_GetIP` only — no
     `_Unwind_FindEnclosingFunction`, no `dl_iterate_phdr`, no locks.
   - Emit (rare, Poisson-sampled): trim frames to root/leaf boundaries
     using `_Unwind_FindEnclosingFunction`.

   This moves the global `dl_iterate_phdr` mutex off the per-yield
   capture path onto the infrequent emit path.

Also adds `--no-task-dumps` flag to metrics-service for A/B comparison.

Before: 14K scheduling delays >1ms, worker pegged at 99.7%,
        pthread_mutex_lock 14.3% of CPU samples, POST p99 5.5ms
After:  373 scheduling delays >1ms, workers at 16%,
        mutex contention gone from hotspots, POST p99 2.5ms
Baseline (no dumps): 262 delays >1ms, 12% util, POST p99 1.85ms

Author: rcoh

dial9-tokio-telemetry/src/task_dumped.rs

@@ -100,6 +100,12 @@ pin_project! {
         sample_mean_ns: u64,
         // Per-task PRNG for drawing exponential gaps.
         rng: SplitMix64,
+        // Set after `capture()` re-polls the inner future with the real waker.
+        // The re-poll causes a spurious immediate wake; this flag suppresses
+        // the next capture to break the busy loop (capture → wake → poll →
+        // capture → …). Cleared on the next poll so subsequent real wakes
+        // proceed normally.
+        just_captured: bool,
     }
 }
 
@@ -126,6 +132,7 @@ impl<F> TaskDumped<F> {
             next_sample_ns,
             sample_mean_ns,
             rng,
+            just_captured: false,
         }
     }
 }
@@ -172,9 +179,17 @@ impl<F: Future> Future for TaskDumped<F> {
                 *this.pending_capture_ts = None;
             }
             Poll::Pending => {
-                this.frames.capture(this.inner.as_mut(), cx);
-                let poll_end = crate::telemetry::recorder::poll_start_ts_or_now();
-                *this.pending_capture_ts = NonZeroU64::new(poll_end);
+                // Skip capture if this poll was triggered by the spurious wake
+                // from the previous capture's re-poll. This breaks the busy
+                // loop: capture → wake → poll → capture → …
+                if *this.just_captured {
+                    *this.just_captured = false;
+                } else {
+                    this.frames.capture(this.inner.as_mut(), cx);
+                    *this.just_captured = true;
+                    let poll_end = crate::telemetry::recorder::poll_start_ts_or_now();
+                    *this.pending_capture_ts = NonZeroU64::new(poll_end);
+                }
             }
         }
         result
@@ -207,16 +222,42 @@ impl FrameBuf {
     }
 
     /// Emit one `TaskDumpEvent` per recorded callchain, then clear.
+    /// Trimming via `_Unwind_FindEnclosingFunction` happens here (emit path)
+    /// rather than during capture, keeping the hot path lock-free.
     fn emit(&mut self, shared: &SharedState, task_id: TaskId, capture_ts: u64) {
         shared.if_enabled(|buf| {
-            for i in 0..self.offsets.len() {
-                let start = self.offsets[i];
-                let end = self.offsets.get(i + 1).copied().unwrap_or(self.ips.len());
-                buf.record_encodable_event(&TaskDumpData {
-                    timestamp_ns: capture_ts,
-                    task_id,
-                    callchain: &self.ips[start..end],
-                });
+            // Each chain is stored as 3 entries in offsets:
+            //   [ip_start_index, root_addr_as_usize, leaf_addr_as_usize]
+            for chunk in self.offsets.chunks_exact(3) {
+                let ip_start = chunk[0];
+                let root_addr = chunk[1];
+                let leaf_addr = chunk[2];
+
+                // Determine the end of this chain's IPs: either the start of
+                // the next chain or the end of the ips buffer.
+                let ip_end = self
+                    .offsets
+                    .chunks_exact(3)
+                    .map(|c| c[0])
+                    .find(|&s| s > ip_start)
+                    .unwrap_or(self.ips.len());
+
+                let raw = &self.ips[ip_start..ip_end];
+                let root = if root_addr == 0 {
+                    None
+                } else {
+                    Some(root_addr as *const core::ffi::c_void)
+                };
+                let leaf = leaf_addr as *const core::ffi::c_void;
+                let chain = crate::unwind::trim_frames(raw, root, leaf);
+
+                if !chain.is_empty() {
+                    buf.record_encodable_event(&TaskDumpData {
+                        timestamp_ns: capture_ts,
+                        task_id,
+                        callchain: chain,
+                    });
+                }
             }
         });
         self.clear();
@@ -241,23 +282,18 @@ impl FrameBuf {
             },
             |meta| {
                 offsets.push(ips.len());
-                capture_frames(ips, meta.root_addr, meta.trace_leaf_addr);
+                // Hot path: collect raw IPs only — no _Unwind_FindEnclosingFunction,
+                // no dl_iterate_phdr, no global locks. Trimming to root/leaf
+                // boundaries happens later in emit().
+                crate::unwind::collect_frames_raw(ips);
+                // Stash the root/leaf addresses so we can trim at emit time.
+                offsets.push(meta.root_addr.map_or(0, |p| p as usize));
+                offsets.push(meta.trace_leaf_addr as usize);
             },
         );
     }
 }
 
-/// Walk the stack, collecting instruction pointers between `leaf_addr` and
-/// `root_addr`. Calls `_Unwind_Backtrace` directly via [`crate::unwind`],
-/// bypassing the `backtrace` crate's process-wide mutex.
-fn capture_frames(
-    ips: &mut Vec<u64>,
-    root_addr: Option<*const core::ffi::c_void>,
-    leaf_addr: *const core::ffi::c_void,
-) {
-    crate::unwind::collect_frames(ips, root_addr, leaf_addr);
-}
-
 /// Borrowed-callchain view of a task-dump event that implements [`Encodable`]
 /// by interning its ips into the batch's stack pool.
 pub(crate) struct TaskDumpData<'a> {

dial9-tokio-telemetry/src/unwind.rs

@@ -2,10 +2,21 @@
 //!
 //! Bypasses the `backtrace` crate's global mutex by calling `_Unwind_Backtrace`
 //! directly. Adapted from backtrace-0.3.76/src/backtrace/libunwind.rs (Apache-2.0/MIT).
+//!
+//! # Two-phase design
+//!
+//! Capture and trimming are separated to keep the hot path lock-free:
+//!
+//! - [`collect_frames_raw`]: called on every yield. Collects all IPs using only
+//!   `_Unwind_GetIP` — no global locks.
+//! - [`trim_frames`]: called only at emit time (rare, Poisson-sampled). Uses
+//!   `_Unwind_FindEnclosingFunction` to locate the root/leaf boundaries and
+//!   trims the collected IPs. This calls `dl_iterate_phdr` internally but only
+//!   runs on the infrequent emit path.
 
 use core::ffi::c_void;
 
-#[allow(non_camel_case_types)]
+#[allow(non_camel_case_types, dead_code)]
 #[repr(C)]
 enum UnwindReasonCode {
     NoReason = 0,
@@ -23,51 +34,55 @@ unsafe extern "C" {
     fn _Unwind_FindEnclosingFunction(pc: *mut c_void) -> *mut c_void;
 }
 
-struct CallbackData<'a> {
+// ─── Raw collection (hot path, no locks) ────────────────────────────────────
+
+struct RawCallbackData<'a> {
     frame_ips: &'a mut Vec<u64>,
-    above_leaf: bool,
-    root_addr: Option<*const c_void>,
-    leaf_addr: *const c_void,
 }
 
-extern "C" fn trace_fn(ctx: *mut UnwindContext, arg: *mut c_void) -> UnwindReasonCode {
-    let data = unsafe { &mut *arg.cast::<CallbackData<'_>>() };
-    let ip = unsafe { _Unwind_GetIP(ctx) } as *mut c_void;
-    let symbol_address = unsafe { _Unwind_FindEnclosingFunction(ip) };
-
-    let below_root = data
-        .root_addr
-        .is_none_or(|root| !std::ptr::eq(symbol_address, root));
-
-    if data.above_leaf && below_root {
-        data.frame_ips.push(ip as u64);
-    }
-
-    if std::ptr::eq(symbol_address, data.leaf_addr) {
-        data.above_leaf = true;
-    }
+extern "C" fn raw_trace_fn(ctx: *mut UnwindContext, arg: *mut c_void) -> UnwindReasonCode {
+    let data = unsafe { &mut *arg.cast::<RawCallbackData<'_>>() };
+    let ip = unsafe { _Unwind_GetIP(ctx) } as u64;
+    data.frame_ips.push(ip);
+    UnwindReasonCode::NoReason
+}
 
-    if below_root {
-        UnwindReasonCode::NoReason
-    } else {
-        UnwindReasonCode::Failure
+/// Collect all instruction pointers on the current call stack.
+/// Uses only `_Unwind_GetIP` — no `dl_iterate_phdr`, no global locks.
+pub(crate) fn collect_frames_raw(frame_ips: &mut Vec<u64>) {
+    let mut data = RawCallbackData { frame_ips };
+    unsafe {
+        _Unwind_Backtrace(raw_trace_fn, (&raw mut data).cast());
     }
 }
 
-/// Walk the call stack, collecting instruction pointers between `leaf_addr` and
-/// `root_addr`. This calls `_Unwind_Backtrace` directly without any global lock.
-pub(crate) fn collect_frames(
-    frame_ips: &mut Vec<u64>,
+// ─── Trimming (emit path, uses FindEnclosingFunction) ───────────────────────
+
+/// Trim a raw IP list to only the frames between `leaf_addr` and `root_addr`.
+/// Calls `_Unwind_FindEnclosingFunction` which takes `dl_iterate_phdr`'s lock,
+/// but this only runs on the infrequent emit path.
+pub(crate) fn trim_frames(
+    frame_ips: &[u64],
     root_addr: Option<*const c_void>,
     leaf_addr: *const c_void,
-) {
-    let mut data = CallbackData {
-        frame_ips,
-        above_leaf: false,
-        root_addr,
-        leaf_addr,
-    };
-    unsafe {
-        _Unwind_Backtrace(trace_fn, (&raw mut data).cast());
+) -> &[u64] {
+    // Find the first frame inside the leaf function (start of interesting range).
+    let leaf_start = frame_ips.iter().position(|&ip| {
+        let sym = unsafe { _Unwind_FindEnclosingFunction(ip as *mut c_void) };
+        std::ptr::eq(sym, leaf_addr)
+    });
+
+    // Find the first frame inside the root function (end of interesting range).
+    let root_end = root_addr.and_then(|root| {
+        frame_ips.iter().position(|&ip| {
+            let sym = unsafe { _Unwind_FindEnclosingFunction(ip as *mut c_void) };
+            std::ptr::eq(sym, root)
+        })
+    });
+
+    match (leaf_start, root_end) {
+        (Some(start), Some(end)) if start + 1 < end => &frame_ips[start + 1..end],
+        (Some(start), None) if start + 1 < frame_ips.len() => &frame_ips[start + 1..],
+        _ => frame_ips,
     }
 }

dial9-viewer/ui/demo-trace.bin

@@ -48,7 +48,7 @@ struct Args {
 
     #[arg(
         long,
-        default_value = "100000000",
+        default_value = "100000000", // 100 MB
         help = "Max trace file size in bytes"
     )]
     trace_max_file_size: u64,
@@ -77,6 +77,9 @@ struct Args {
 
     #[arg(long, help = "AWS region for S3 uploads (defaults to SDK default)")]
     s3_region: Option<String>,
+
+    #[arg(long, help = "Disable task dump capture")]
+    no_task_dumps: bool,
 }
 
 #[derive(Clone)]
@@ -147,8 +150,8 @@ fn main() -> std::io::Result<()> {
     if args.demo {
         args.run_duration = 4;
         args.worker_threads = 2;
-        args.trace_max_file_size = 20_000_000;
-        args.trace_max_total_size = 20_000_000;
+        args.trace_max_file_size = 100_000_000;
+        args.trace_max_total_size = 100_000_000;
     }
 
     let writer = RotatingWriter::builder()
@@ -169,12 +172,16 @@ fn main() -> std::io::Result<()> {
     builder.worker_threads(args.worker_threads).enable_all();
     let traced_builder = TracedRuntime::builder()
         .with_trace_path(&args.trace_path)
-        .with_task_tracking(true)
-        .with_task_dumps(
+        .with_task_tracking(true);
+    let traced_builder = if args.no_task_dumps {
+        traced_builder
+    } else {
+        traced_builder.with_task_dumps(
             TaskDumpConfig::builder()
                 .idle_threshold(Duration::from_millis(5))
                 .build(),
-        );
+        )
+    };
     #[cfg(target_os = "linux")]
     let traced_builder = traced_builder
         .with_cpu_profiling(CpuProfilingConfig::default())