fix: Don't register sched events on blocking pool threads (#316)

* fix: Don't register sched events on blocking pool threads

* fix: use Drop-based guard for per-thread telemetry cleanup

Replace manual on_thread_stop cleanup with a ThreadCleanupGuard stored
in a thread-local. The guard's Drop impl handles all per-thread teardown:
clearing CURRENT_HANDLE, removing thread roles, stopping sched profiling,
and unregistering ctimer.

This fixes a perf_event fd leak where the calling thread's CURRENT_HANDLE
(set in attach_runtime) was never cleared, keeping Arc<SharedState> alive
after drop(runtime) + drop(guard). The SchedProfiler inside SharedState
still owned PerfEvents with open fds.

The guard approach is more reliable because:
- The calling thread's cleanup is now handled (was missing before)
- Even if on_thread_stop doesn't fire, the guard runs on thread exit
- Cleanup logic is centralized in one place

* fix: serialize perf fd tests to prevent cargo test parallelism interference

cargo test runs tests within the same binary in parallel (threads in the
same process). Both sched_profiler_fd_leak tests inspect process-wide
perf_event fd counts, so running concurrently causes one test to see the
other's fds. Add a static Mutex to serialize them.

* Revert "fix: use Drop-based guard for per-thread telemetry cleanup"

This reverts commit 75706c8.

* fix: add max_tracked_threads cap to PerfSamplerImpl

Adds a configurable hard cap (default: 256) on the number of threads
that can be tracked simultaneously in per-thread mode. If the cap is
reached, track_current_thread returns an error instead of opening
another fd. This prevents unbounded fd/mmap growth if cleanup fails,
ensuring dial9 never breaks the calling application.

* fix: add missing max_tracked_threads field in ctimer_sampler test

* Cleanup comments

Author: rcoh

dial9-tokio-telemetry/design/cpu-sampling-and-worker-id-design.md

@@ -136,7 +136,7 @@ The `SchedProfiler` captures context switches on each worker thread:
 
 - Created during `TracedRuntimeBuilder::build()` if `with_sched_events()` was called.
 - Stored in `SharedState.sched_profiler` (behind `Mutex<Option<...>>`).
-- `on_thread_start` hook calls `profiler.track_current_thread()` → opens a perf fd for the calling thread.
+- `on_thread_start` registers the thread as `Blocking`; worker threads re-register as `Worker(i)` in `register_tid_if_needed()` on their first poll/park, which also calls `profiler.track_current_thread()` → opens a perf fd for the calling thread.
 - `on_thread_stop` hook calls `profiler.stop_tracking_current_thread()`.
 - Flush thread drains samples, maps tids, writes `CpuSample` events with `source: SchedEvent`.
 

dial9-tokio-telemetry/src/telemetry/cpu_profile.rs

@@ -59,7 +59,7 @@ impl CpuProfilingConfig {
 /// Configuration for per-worker sched event capture (context switches).
 ///
 /// Uses `perf_event_open` with `SwContextSwitches` in per-thread mode,
-/// so each worker thread gets its own perf fd via `on_thread_start`.
+/// so each worker thread gets its own perf fd on first poll/park.
 #[derive(Debug, Clone, Default)]
 pub struct SchedEventConfig {
     sampling_interval: Option<u64>,

dial9-tokio-telemetry/src/telemetry/recorder/mod.rs

@@ -241,12 +241,10 @@ fn register_hooks(
                     .lock()
                     .unwrap()
                     .insert(tid, crate::telemetry::events::ThreadRole::Blocking);
-                if let Ok(mut prof) = s_start.sched_profiler.lock()
-                    && let Some(ref mut p) = *prof
-                {
-                    // Register the current thread for sched event sampling.
-                    let _ = p.track_current_thread();
-                }
+                // Sched event sampling is deferred to register_tid_if_needed(),
+                // which runs only for worker threads on their first poll/park.
+                // This avoids opening perf fds for blocking pool threads.
+
                 // Registers the current thread for the CPU-profiling fallback (ctimer).
                 // No-op when perf is the active backend (perf uses inherit).
                 let _ = dial9_perf_self_profile::register_current_thread();

dial9-tokio-telemetry/src/telemetry/recorder/runtime_context.rs

@@ -110,6 +110,7 @@ fn register_worker_if_needed(ctx: &RuntimeContext, local_index: usize, global_id
 }
 
 /// Register the current thread's OS tid for CPU profiling (once per thread).
+/// Also starts sched event sampling for this worker thread.
 #[cfg(feature = "cpu-profiling")]
 fn register_tid_if_needed(global_id: u64, shared: &SharedState) {
     TID_REGISTERED.with(|cell| {
@@ -119,6 +120,15 @@ fn register_tid_if_needed(global_id: u64, shared: &SharedState) {
                 os_tid,
                 crate::telemetry::events::ThreadRole::Worker(global_id as usize),
             );
+            // Start sched event sampling for this worker thread. Deferred from
+            // on_thread_start so that only worker threads (not blocking pool
+            // threads) open perf fds.
+            if let Ok(mut prof) = shared.sched_profiler.lock()
+                && let Some(ref mut p) = *prof
+                && let Err(e) = p.track_current_thread()
+            {
+                tracing::warn!("failed to start sched profiling for worker thread: {e}");
+            }
             cell.set(true);
         }
     });

dial9-tokio-telemetry/tests/sched_profiler_fd_leak.rs

@@ -0,0 +1,144 @@
+//! Integration test: sched profiler should only track worker threads, not blocking pool threads.
+//!
+//! Before the fix, every thread that started (including blocking pool threads) opened a
+//! perf_event_open fd + 2 MB mmap ring buffer for sched event sampling. With Tokio's default
+//! blocking pool limit of 512 threads, this could exhaust file descriptors.
+//!
+//! Additionally, `stop_tracking_current_thread` never cleaned up fds because `open_perf_event`
+//! stored `tid: 0` (the "current thread" sentinel) while the cleanup searched for `gettid()`.
+
+#![cfg(all(feature = "cpu-profiling", target_os = "linux"))]
+
+mod common;
+
+use dial9_tokio_telemetry::telemetry::TracedRuntime;
+use dial9_tokio_telemetry::telemetry::cpu_profile::SchedEventConfig;
+use std::sync::Mutex;
+use std::time::Duration;
+
+/// Serialize tests that inspect process-wide perf_event fd counts.
+/// `cargo test` runs tests in the same binary in parallel (threads),
+/// so concurrent tests would see each other's fds.
+static PERF_FD_TEST_LOCK: Mutex<()> = Mutex::new(());
+
+/// Count open perf_event fds specifically.
+fn count_perf_fds() -> usize {
+    std::fs::read_dir("/proc/self/fd")
+        .expect("failed to read /proc/self/fd")
+        .filter_map(|e| e.ok())
+        .filter(|e| {
+            std::fs::read_link(e.path())
+                .map(|p| p.to_string_lossy().contains("perf_event"))
+                .unwrap_or(false)
+        })
+        .count()
+}
+
+/// Spawning many blocking threads should NOT cause fd count to grow proportionally.
+///
+/// With the bug, each `spawn_blocking` call opens a perf fd that is never reclaimed.
+/// After the fix, only worker threads (a fixed, small number) get perf fds.
+#[test]
+fn sched_profiler_fds_bounded_with_many_blocking_threads() {
+    let _lock = PERF_FD_TEST_LOCK.lock().unwrap();
+    let (writer, _events) = common::CapturingWriter::new();
+
+    let num_workers = 2;
+    let num_blocking_tasks = 50;
+
+    let mut builder = tokio::runtime::Builder::new_multi_thread();
+    builder.worker_threads(num_workers).enable_all();
+
+    let (runtime, guard) = TracedRuntime::builder()
+        .with_sched_events(SchedEventConfig::default())
+        .build_and_start(builder, writer)
+        .unwrap();
+
+    // Let workers start and resolve their identity.
+    runtime.block_on(async {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    });
+
+    let perf_fds_before = count_perf_fds();
+
+    // Spawn many blocking tasks. Each one creates a new blocking pool thread.
+    // Use std::thread::sleep to ensure they actually block and force new threads.
+    runtime.block_on(async {
+        let mut handles = Vec::new();
+        for _ in 0..num_blocking_tasks {
+            handles.push(tokio::task::spawn_blocking(|| {
+                std::thread::sleep(Duration::from_millis(50));
+            }));
+        }
+        for h in handles {
+            h.await.unwrap();
+        }
+        // Wait for threads to exit and on_thread_stop to fire.
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    });
+
+    let perf_fds_after = count_perf_fds();
+
+    drop(runtime);
+    drop(guard);
+
+    // Only worker threads should have perf fds. Before the fix, we'd see
+    // ~50 new perf fds (one per blocking thread). After the fix, the count
+    // should stay at exactly num_workers.
+    assert_eq!(
+        perf_fds_before, perf_fds_after,
+        "perf fd count changed from {perf_fds_before} to {perf_fds_after} after \
+         spawning {num_blocking_tasks} blocking tasks. \
+         Sched profiler is likely opening fds for blocking pool threads."
+    );
+}
+
+/// Verify that sched profiler fds are properly cleaned up when the runtime shuts down.
+///
+/// This catches the tid=0 bug where `stop_tracking_current_thread` can never find
+/// the event to remove because `open_perf_event` stored tid=0 instead of the real tid.
+#[test]
+fn sched_profiler_fds_cleaned_up_on_shutdown() {
+    let _lock = PERF_FD_TEST_LOCK.lock().unwrap();
+    assert_eq!(count_perf_fds(), 0, "no perf fds should exist before test");
+
+    {
+        let (writer, _events) = common::CapturingWriter::new();
+
+        let num_workers = 4;
+        let mut builder = tokio::runtime::Builder::new_multi_thread();
+        builder.worker_threads(num_workers).enable_all();
+
+        let (runtime, guard) = TracedRuntime::builder()
+            .with_sched_events(SchedEventConfig::default())
+            .build_and_start(builder, writer)
+            .unwrap();
+
+        // Do some work so workers resolve their identity.
+        runtime.block_on(async {
+            for _ in 0..10 {
+                tokio::spawn(async { tokio::task::yield_now().await })
+                    .await
+                    .unwrap();
+            }
+            tokio::time::sleep(Duration::from_millis(200)).await;
+        });
+
+        // Workers should have perf fds while running.
+        let perf_fds_during = count_perf_fds();
+        assert!(
+            perf_fds_during > 0,
+            "expected perf fds while runtime is running, got 0"
+        );
+
+        drop(runtime);
+        drop(guard);
+    }
+
+    let perf_fds_after = count_perf_fds();
+    assert_eq!(
+        perf_fds_after, 0,
+        "leaked {perf_fds_after} perf_event fds after runtime shutdown. \
+         stop_tracking_current_thread likely failed to find events due to tid=0 bug."
+    );
+}

perf-self-profile/src/sampler.rs

@@ -41,6 +41,10 @@ pub struct SamplerConfig {
     pub(crate) event_source: EventSource,
     pub(crate) sampling: SamplingMode,
     pub(crate) include_kernel: bool,
+    /// Maximum number of threads that can be tracked simultaneously in
+    /// per-thread mode. Prevents unbounded fd/mmap growth if cleanup fails.
+    /// Default: 256.
+    pub(crate) max_tracked_threads: usize,
 }
 
 impl Default for SamplerConfig {
@@ -49,6 +53,7 @@ impl Default for SamplerConfig {
             sampling: SamplingMode::FrequencyHz(999),
             event_source: EventSource::SwCpuClock,
             include_kernel: false,
+            max_tracked_threads: 256,
         }
     }
 }
@@ -72,6 +77,14 @@ impl SamplerConfig {
         self.include_kernel = yes;
         self
     }
+
+    /// Maximum number of threads tracked simultaneously in per-thread mode.
+    /// If the cap is reached, `track_current_thread` returns an error instead
+    /// of opening another fd. Default: 256.
+    pub fn max_tracked_threads(mut self, max: usize) -> Self {
+        self.max_tracked_threads = max;
+        self
+    }
 }
 
 /// A single sample captured from perf events.

perf-self-profile/src/sys/linux/ctimer_sampler.rs

@@ -216,6 +216,7 @@ mod tests {
             sampling: SamplingMode::Period(1),
             event_source: EventSource::SwCpuClock,
             include_kernel: false,
+            max_tracked_threads: 256,
         };
         let err = CtimerSampler::start(&config).unwrap_err();
         assert_eq!(err.kind(), io::ErrorKind::Unsupported);

perf-self-profile/src/sys/linux/perf_sampler.rs

@@ -92,6 +92,7 @@ pub(super) struct PerfSamplerImpl {
     parse_config: ParseConfig<Little>,
     attr: perf_event_attr,
     include_kernel: bool,
+    max_tracked_threads: usize,
 }
 
 impl PerfSamplerImpl {
@@ -122,6 +123,7 @@ impl PerfSamplerImpl {
             parse_config: ParseConfig::from(attr),
             attr,
             include_kernel: config.include_kernel,
+            max_tracked_threads: config.max_tracked_threads,
         })
     }
 
@@ -150,6 +152,7 @@ impl PerfSamplerImpl {
             parse_config: ParseConfig::from(attr),
             attr,
             include_kernel: config.include_kernel,
+            max_tracked_threads: config.max_tracked_threads,
         })
     }
 
@@ -279,7 +282,18 @@ impl super::sampler::SamplerBackend for PerfSamplerImpl {
     // Per-thread mode: call from the thread you want to monitor.
     // This opens an event fd scoped to the calling tid with cpu=-1.
     fn track_current_thread(&mut self) -> io::Result<()> {
-        let ev = open_perf_event(&mut self.attr, 0, -1)?;
+        if self.events.len() >= self.max_tracked_threads {
+            return Err(io::Error::other(format!(
+                "perf sampler: max tracked threads ({}) reached, \
+                 refusing to open another fd",
+                self.max_tracked_threads
+            )));
+        }
+        let mut ev = open_perf_event(&mut self.attr, 0, -1)?;
+        // open_perf_event stores tid=0 (the "current thread" sentinel passed to
+        // perf_event_open). Resolve to the real tid so stop_tracking_current_thread
+        // can find this event later.
+        ev.tid = gettid() as i32;
         self.events.push(ev);
         Ok(())
     }