Merge branch 'main' into release-0.3

Author: rcoh

Cargo.lock

@@ -29,10 +29,7 @@ fn main() {
     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_cpu_profiling(CpuProfilingConfig::default().frequency_hz(999))
         .build_and_start(builder, writer)
         .unwrap();
 

dial9-tokio-telemetry/examples/blocking_pool_tracking.rs

@@ -21,12 +21,9 @@ pub(crate) fn read_thread_name(tid: u32) -> Option<String> {
 /// Configuration for CPU profiling integration.
 #[derive(Debug, Clone)]
 pub struct CpuProfilingConfig {
-    /// Sampling frequency in Hz. Default: 99 (low overhead).
-    pub frequency_hz: u64,
-    /// Which perf event source to use.
-    pub event_source: EventSource,
-    /// Whether to include kernel stack frames.
-    pub include_kernel: bool,
+    frequency_hz: u64,
+    event_source: EventSource,
+    include_kernel: bool,
 }
 
 impl Default for CpuProfilingConfig {
@@ -39,6 +36,26 @@ impl Default for CpuProfilingConfig {
     }
 }
 
+impl CpuProfilingConfig {
+    /// Sampling frequency in Hz. Default: 99 (low overhead).
+    pub fn frequency_hz(mut self, hz: u64) -> Self {
+        self.frequency_hz = hz;
+        self
+    }
+
+    /// Which perf event source to use.
+    pub fn event_source(mut self, source: EventSource) -> Self {
+        self.event_source = source;
+        self
+    }
+
+    /// Whether to include kernel stack frames.
+    pub fn include_kernel(mut self, yes: bool) -> Self {
+        self.include_kernel = yes;
+        self
+    }
+}
+
 /// Configuration for per-worker sched event capture (context switches).
 ///
 /// Uses `perf_event_open` with `SwContextSwitches` in per-thread mode,
@@ -82,11 +99,12 @@ pub(crate) struct CpuProfiler {
 
 impl CpuProfiler {
     pub(crate) fn start(config: CpuProfilingConfig) -> io::Result<Self> {
-        let sampler = PerfSampler::start(SamplerConfig {
-            event_source: config.event_source,
-            sampling: SamplingMode::FrequencyHz(config.frequency_hz),
-            include_kernel: config.include_kernel,
-        })?;
+        let sampler = PerfSampler::start(
+            SamplerConfig::default()
+                .event_source(config.event_source)
+                .sampling(SamplingMode::FrequencyHz(config.frequency_hz))
+                .include_kernel(config.include_kernel),
+        )?;
         Ok(Self {
             sampler,
             pid: std::process::id(),
@@ -130,11 +148,12 @@ pub(crate) struct SchedProfiler {
 
 impl SchedProfiler {
     pub(crate) fn new(config: SchedEventConfig) -> io::Result<Self> {
-        let sampler = PerfSampler::new_per_thread(SamplerConfig {
-            event_source: EventSource::SwContextSwitches,
-            sampling: SamplingMode::Period(config.sampling_interval.unwrap_or(1)),
-            include_kernel: config.include_kernel,
-        })?;
+        let sampler = PerfSampler::new_per_thread(
+            SamplerConfig::default()
+                .event_source(EventSource::SwContextSwitches)
+                .sampling(SamplingMode::Period(config.sampling_interval.unwrap_or(1)))
+                .include_kernel(config.include_kernel),
+        )?;
         Ok(Self { sampler })
     }
 

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

@@ -82,7 +82,9 @@ fn flush_once(
     let cpu_events_time = Instant::now();
     #[cfg(feature = "cpu-profiling")]
     {
-        event_writer.flush_cpu(shared);
+        if shared.enabled.load(Ordering::Relaxed) {
+            event_writer.flush_cpu(shared);
+        }
     }
     let cpu_flush_duration = cpu_events_time.elapsed();
 
@@ -1095,6 +1097,13 @@ fn run_flush_loop(
             Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
         }
 
+        // When disabled, skip all recording work (queue sampling, metadata
+        // merging, drain coordination, flush). The loop still wakes every
+        // 5ms to check for control commands and the exit signal.
+        if !exit && !shared.enabled.load(Ordering::Relaxed) {
+            continue;
+        }
+
         let now = Instant::now();
         if now.duration_since(last_sample) >= sample_interval {
             last_sample = now;

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

@@ -41,10 +41,7 @@ fn background_symbolization_produces_symbol_table_entries() {
     builder.worker_threads(2).enable_all();
 
     let (runtime, guard) = TracedRuntime::builder()
-        .with_cpu_profiling(CpuProfilingConfig {
-            frequency_hz: 999,
-            ..Default::default()
-        })
+        .with_cpu_profiling(CpuProfilingConfig::default().frequency_hz(999))
         .with_trace_path(&trace_path)
         .with_worker_poll_interval(std::time::Duration::from_millis(50))
         .build_and_start(builder, writer)

dial9-tokio-telemetry/tests/background_symbolization.rs

@@ -33,10 +33,7 @@ fn cpu_sample_timestamps_align_with_wall_clock() {
     builder.worker_threads(num_workers).enable_all();
 
     let (runtime, guard) = TracedRuntime::builder()
-        .with_cpu_profiling(CpuProfilingConfig {
-            frequency_hz: 999,
-            ..Default::default()
-        })
+        .with_cpu_profiling(CpuProfilingConfig::default().frequency_hz(999))
         .build_and_start(builder, writer)
         .unwrap();
 
@@ -274,10 +271,7 @@ fn thread_name_attribution_for_external_and_blocking_threads() {
         .enable_all();
 
     let (runtime, guard) = TracedRuntime::builder()
-        .with_cpu_profiling(CpuProfilingConfig {
-            frequency_hz: 999,
-            ..Default::default()
-        })
+        .with_cpu_profiling(CpuProfilingConfig::default().frequency_hz(999))
         .build_and_start(builder, writer)
         .unwrap();