keyhog_profile/
lib.rs

1//! Low-overhead causal profiling for KeyHog runs.
2//!
3//! The disabled hot path performs one relaxed atomic load and does not read the
4//! clock. An enabled run records fixed scanner stages with allocation-free atomic
5//! counters. Run identity, state transitions, and process resources are sampled
6//! only at macro boundaries.
7
8use serde::{Deserialize, Serialize};
9use std::cell::Cell;
10use std::fmt;
11use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
12use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
13#[cfg(not(target_os = "linux"))]
14use sysinfo::{ProcessesToUpdate, System};
15
16/// Stable wire schema for persisted profiling records.
17pub const PROFILE_SCHEMA: &str = "keyhog-profile-v1";
18
19/// Fixed measurement stages shared by scanner, source, verifier, and reporter paths.
20#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
21#[serde(rename_all = "kebab-case")]
22#[repr(usize)]
23pub enum Stage {
24    SourceAcquire = 0,
25    SourceWalk,
26    SourceRead,
27    Preprocess,
28    Phase1Triggers,
29    BackendDispatch,
30    HotPatterns,
31    ConfirmedPatterns,
32    Phase2Prefilter,
33    Phase2KeywordAc,
34    Phase2SharedAc,
35    Phase2AnchoredVerify,
36    Phase2WholeChunk,
37    GenericDetection,
38    Entropy,
39    MachineLearning,
40    Decode,
41    Suppression,
42    LiveVerification,
43    Reporting,
44    SourceQueueWait,
45    IncrementalLookup,
46    BackendSelect,
47    ResultMerge,
48    ScannerQueueWait,
49}
50
51impl Stage {
52    /// Every stage in stable wire order.
53    pub const ALL: [Self; 25] = [
54        Self::SourceAcquire,
55        Self::SourceWalk,
56        Self::SourceRead,
57        Self::Preprocess,
58        Self::Phase1Triggers,
59        Self::BackendDispatch,
60        Self::HotPatterns,
61        Self::ConfirmedPatterns,
62        Self::Phase2Prefilter,
63        Self::Phase2KeywordAc,
64        Self::Phase2SharedAc,
65        Self::Phase2AnchoredVerify,
66        Self::Phase2WholeChunk,
67        Self::GenericDetection,
68        Self::Entropy,
69        Self::MachineLearning,
70        Self::Decode,
71        Self::Suppression,
72        Self::LiveVerification,
73        Self::Reporting,
74        Self::SourceQueueWait,
75        Self::IncrementalLookup,
76        Self::BackendSelect,
77        Self::ResultMerge,
78        Self::ScannerQueueWait,
79    ];
80
81    #[inline]
82    const fn index(self) -> usize {
83        self as usize
84    }
85
86    /// Stable text label used by human reports.
87    pub const fn as_str(self) -> &'static str {
88        match self {
89            Self::SourceAcquire => "source-acquire",
90            Self::SourceWalk => "source-walk",
91            Self::SourceRead => "source-read",
92            Self::Preprocess => "preprocess",
93            Self::Phase1Triggers => "phase1-triggers",
94            Self::BackendDispatch => "backend-dispatch",
95            Self::HotPatterns => "hot-patterns",
96            Self::ConfirmedPatterns => "confirmed-patterns",
97            Self::Phase2Prefilter => "phase2-prefilter",
98            Self::Phase2KeywordAc => "phase2-keyword-ac",
99            Self::Phase2SharedAc => "phase2-shared-ac",
100            Self::Phase2AnchoredVerify => "phase2-anchored-verify",
101            Self::Phase2WholeChunk => "phase2-whole-chunk",
102            Self::GenericDetection => "generic-detection",
103            Self::Entropy => "entropy",
104            Self::MachineLearning => "machine-learning",
105            Self::Decode => "decode",
106            Self::Suppression => "suppression",
107            Self::LiveVerification => "live-verification",
108            Self::Reporting => "reporting",
109            Self::SourceQueueWait => "source-queue-wait",
110            Self::IncrementalLookup => "incremental-lookup",
111            Self::BackendSelect => "backend-select",
112            Self::ResultMerge => "result-merge",
113            Self::ScannerQueueWait => "scanner-queue-wait",
114        }
115    }
116}
117
118const STAGE_COUNT: usize = Stage::ALL.len();
119const fn zero_counters() -> [AtomicU64; STAGE_COUNT] {
120    [const { AtomicU64::new(0) }; STAGE_COUNT]
121}
122static ENABLED: AtomicBool = AtomicBool::new(false);
123static SESSION_ACTIVE: AtomicBool = AtomicBool::new(false);
124static ELAPSED_NS: [AtomicU64; STAGE_COUNT] = zero_counters();
125static CALLS: [AtomicU64; STAGE_COUNT] = zero_counters();
126static ATTRIBUTED_NS: [AtomicU64; STAGE_COUNT] = zero_counters();
127static SESSION_ELAPSED_NS: [AtomicU64; STAGE_COUNT] = zero_counters();
128static SESSION_CALLS: [AtomicU64; STAGE_COUNT] = zero_counters();
129static SESSION_ATTRIBUTED_NS: [AtomicU64; STAGE_COUNT] = zero_counters();
130static INPUT_BYTES: AtomicU64 = AtomicU64::new(0);
131static INPUT_UNITS: AtomicU64 = AtomicU64::new(0);
132static SESSION_INPUT_BYTES: AtomicU64 = AtomicU64::new(0);
133static SESSION_INPUT_UNITS: AtomicU64 = AtomicU64::new(0);
134static RUN_SEQUENCE: AtomicU64 = AtomicU64::new(0);
135
136/// Optional attribution for work performed inside a derived input.
137#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
138#[repr(u8)]
139pub enum Attribution {
140    #[default]
141    Root = 0,
142    Decoded = 1,
143}
144
145thread_local! {
146    static ATTRIBUTION: Cell<Attribution> = const { Cell::new(Attribution::Root) };
147}
148
149/// Replace this thread's attribution and return its previous value.
150pub fn set_attribution(attribution: Attribution) -> Attribution {
151    ATTRIBUTION.with(|slot| slot.replace(attribution))
152}
153
154/// Return whether fixed-stage profiling is active.
155#[inline]
156pub fn enabled() -> bool {
157    ENABLED.load(Ordering::Relaxed)
158}
159
160/// Enable or disable fixed-stage profiling.
161///
162/// Prefer [`Session::start`] for operator runs because it also captures identity,
163/// resources, and state transitions. This switch remains available to libraries
164/// and microbenchmarks that only need stage counters.
165pub fn set_enabled(enabled: bool) {
166    ENABLED.store(enabled, Ordering::Relaxed);
167}
168
169/// Allocation-free stage guard. It contains no start timestamp while disabled.
170#[must_use]
171pub struct Span {
172    stage: Stage,
173    started: Option<Instant>,
174    session_recording: bool,
175}
176
177impl Span {
178    /// Whether this span reads and will record a clock measurement.
179    pub fn is_recording(&self) -> bool {
180        self.started.is_some()
181    }
182}
183
184/// Start one fixed-stage measurement.
185#[inline]
186pub fn span(stage: Stage) -> Span {
187    let recording = enabled();
188    Span {
189        stage,
190        started: recording.then(Instant::now),
191        session_recording: recording && SESSION_ACTIVE.load(Ordering::Relaxed),
192    }
193}
194
195impl Drop for Span {
196    #[inline]
197    fn drop(&mut self) {
198        let Some(started) = self.started else {
199            return;
200        };
201        let elapsed = u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX);
202        let index = self.stage.index();
203        ELAPSED_NS[index].fetch_add(elapsed, Ordering::Relaxed);
204        CALLS[index].fetch_add(1, Ordering::Relaxed);
205        if self.session_recording {
206            SESSION_ELAPSED_NS[index].fetch_add(elapsed, Ordering::Relaxed);
207            SESSION_CALLS[index].fetch_add(1, Ordering::Relaxed);
208        }
209        if ATTRIBUTION.with(|slot| slot.get()) == Attribution::Decoded {
210            ATTRIBUTED_NS[index].fetch_add(elapsed, Ordering::Relaxed);
211            if self.session_recording {
212                SESSION_ATTRIBUTED_NS[index].fetch_add(elapsed, Ordering::Relaxed);
213            }
214        }
215    }
216}
217
218/// Add source bytes processed by the current profile.
219#[inline]
220pub fn add_input_bytes(bytes: u64) {
221    if enabled() {
222        INPUT_BYTES.fetch_add(bytes, Ordering::Relaxed);
223        if SESSION_ACTIVE.load(Ordering::Relaxed) {
224            SESSION_INPUT_BYTES.fetch_add(bytes, Ordering::Relaxed);
225        }
226    }
227}
228
229/// Add source units such as files, objects, responses, or chunks.
230#[inline]
231pub fn add_input_units(units: u64) {
232    if enabled() {
233        INPUT_UNITS.fetch_add(units, Ordering::Relaxed);
234        if SESSION_ACTIVE.load(Ordering::Relaxed) {
235            SESSION_INPUT_UNITS.fetch_add(units, Ordering::Relaxed);
236        }
237    }
238}
239
240/// One aggregate fixed-stage measurement.
241#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
242pub struct StageMeasurement {
243    pub stage: Stage,
244    pub elapsed_ns: u64,
245    pub calls: u64,
246    pub attributed_ns: u64,
247}
248
249/// Atomically read and clear all fixed-stage counters.
250pub fn take_stage_measurements() -> Vec<StageMeasurement> {
251    take_measurements_from(&ELAPSED_NS, &CALLS, &ATTRIBUTED_NS)
252}
253
254fn take_measurements_from(
255    elapsed: &[AtomicU64; STAGE_COUNT],
256    calls: &[AtomicU64; STAGE_COUNT],
257    attributed: &[AtomicU64; STAGE_COUNT],
258) -> Vec<StageMeasurement> {
259    Stage::ALL
260        .into_iter()
261        .filter_map(|stage| {
262            let index = stage.index();
263            let elapsed_ns = elapsed[index].swap(0, Ordering::Relaxed);
264            let calls = calls[index].swap(0, Ordering::Relaxed);
265            let attributed_ns = attributed[index].swap(0, Ordering::Relaxed);
266            (elapsed_ns != 0 || calls != 0 || attributed_ns != 0).then_some(StageMeasurement {
267                stage,
268                elapsed_ns,
269                calls,
270                attributed_ns,
271            })
272        })
273        .collect()
274}
275
276fn take_session_stage_measurements() -> Vec<StageMeasurement> {
277    take_measurements_from(&SESSION_ELAPSED_NS, &SESSION_CALLS, &SESSION_ATTRIBUTED_NS)
278}
279
280/// Atomically read and clear aggregate input bytes and units.
281pub fn take_input_totals() -> (u64, u64) {
282    (
283        INPUT_BYTES.swap(0, Ordering::Relaxed),
284        INPUT_UNITS.swap(0, Ordering::Relaxed),
285    )
286}
287
288fn take_session_input_totals() -> (u64, u64) {
289    (
290        SESSION_INPUT_BYTES.swap(0, Ordering::Relaxed),
291        SESSION_INPUT_UNITS.swap(0, Ordering::Relaxed),
292    )
293}
294
295/// Discard fixed-stage counters and input totals.
296pub fn reset() {
297    let _ = take_stage_measurements();
298    INPUT_BYTES.store(0, Ordering::Relaxed);
299    INPUT_UNITS.store(0, Ordering::Relaxed);
300}
301
302fn reset_session() {
303    let _ = take_session_stage_measurements();
304    SESSION_INPUT_BYTES.store(0, Ordering::Relaxed);
305    SESSION_INPUT_UNITS.store(0, Ordering::Relaxed);
306}
307
308/// Cache state that materially changes run cost.
309#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
310#[serde(rename_all = "kebab-case")]
311pub enum CacheState {
312    #[default]
313    Unknown,
314    Disabled,
315    Cold,
316    Warm,
317}
318
319/// Daemon state that materially changes startup and resident work.
320#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
321#[serde(rename_all = "kebab-case")]
322pub enum DaemonState {
323    #[default]
324    Off,
325    Client,
326    Worker,
327    Mass,
328}
329
330/// Coarse causal state of a profiling run.
331#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
332#[serde(rename_all = "kebab-case")]
333pub enum RunState {
334    Created,
335    Acquiring,
336    Scanning,
337    Resolving,
338    Verifying,
339    Reporting,
340    Completed,
341    Failed,
342}
343
344/// Identity and execution choices required to compare two run records honestly.
345#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
346pub struct RunIdentity {
347    pub run_id: String,
348    pub binary_version: String,
349    pub detector_digest: String,
350    pub config_digest: String,
351    pub source_kind: String,
352    pub workload_class: String,
353    pub backend_requested: String,
354    pub backend_selected: Option<String>,
355    pub cache_state: CacheState,
356    pub daemon_state: DaemonState,
357    pub scanner_threads: usize,
358    pub reader_threads: Option<usize>,
359    pub logical_cpus: usize,
360}
361
362impl RunIdentity {
363    /// Construct a run identity with a process-unique identifier and explicit state.
364    pub fn new(
365        binary_version: impl Into<String>,
366        detector_digest: impl Into<String>,
367        config_digest: impl Into<String>,
368        source_kind: impl Into<String>,
369        workload_class: impl Into<String>,
370        backend_requested: impl Into<String>,
371    ) -> Self {
372        let sequence = RUN_SEQUENCE.fetch_add(1, Ordering::Relaxed);
373        let unix_ns = SystemTime::now()
374            .duration_since(UNIX_EPOCH)
375            .unwrap_or_default()
376            .as_nanos();
377        Self {
378            run_id: format!("{}-{unix_ns}-{sequence}", std::process::id()),
379            binary_version: binary_version.into(),
380            detector_digest: detector_digest.into(),
381            config_digest: config_digest.into(),
382            source_kind: source_kind.into(),
383            workload_class: workload_class.into(),
384            backend_requested: backend_requested.into(),
385            backend_selected: None,
386            cache_state: CacheState::Unknown,
387            daemon_state: DaemonState::Off,
388            scanner_threads: 0,
389            reader_threads: None,
390            logical_cpus: std::thread::available_parallelism().map_or(1, usize::from),
391        }
392    }
393}
394
395/// One run-state transition relative to session start.
396#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
397pub struct StateTransition {
398    pub state: RunState,
399    pub elapsed_ns: u64,
400}
401
402/// Process resource observation associated with a run-state boundary.
403#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
404pub struct ResourceSample {
405    pub state: RunState,
406    pub elapsed_ns: u64,
407    pub snapshot: ResourceSnapshot,
408}
409
410/// Process resource observation at a macro boundary.
411#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize, Deserialize)]
412pub struct ResourceSnapshot {
413    pub cpu_time_ms: Option<u64>,
414    pub resident_bytes: Option<u64>,
415    pub virtual_bytes: Option<u64>,
416    pub thread_count: Option<u64>,
417}
418
419/// One completed macro state with its wall time and boundary resource deltas.
420#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
421pub struct StateMeasurement {
422    pub state: RunState,
423    pub elapsed_ns: u64,
424    pub cpu_time_ms: Option<u64>,
425    pub aggregate_cpu_milli_percent: Option<u64>,
426    pub resident_start_bytes: Option<u64>,
427    pub resident_end_bytes: Option<u64>,
428    pub threads_start: Option<u64>,
429    pub threads_end: Option<u64>,
430}
431
432#[cfg(target_os = "linux")]
433fn process_resources() -> ResourceSnapshot {
434    fn status_value(status: &str, field: &str, scale: u64) -> Option<u64> {
435        status.lines().find_map(|line| {
436            let value = line.strip_prefix(field)?.split_whitespace().next()?;
437            value.parse::<u64>().ok()?.checked_mul(scale)
438        })
439    }
440
441    fn cpu_time_ms() -> Option<u64> {
442        let stat = std::fs::read_to_string("/proc/self/stat").ok()?;
443        let command_end = stat.rfind(')')?;
444        let mut fields = stat.get(command_end + 2..)?.split_whitespace();
445        let user_ticks = fields.nth(11)?.parse::<u64>().ok()?;
446        let system_ticks = fields.next()?.parse::<u64>().ok()?;
447        // SAFETY: sysconf reads a process constant and receives no pointer.
448        let ticks_per_second = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
449        if ticks_per_second <= 0 {
450            return None;
451        }
452        let milliseconds =
453            (u128::from(user_ticks) + u128::from(system_ticks)) * 1_000 / ticks_per_second as u128;
454        u64::try_from(milliseconds).ok()
455    }
456
457    let status = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
458    let cpu_time_ms = cpu_time_ms();
459    ResourceSnapshot {
460        cpu_time_ms,
461        resident_bytes: status_value(&status, "VmRSS:", 1024),
462        virtual_bytes: status_value(&status, "VmSize:", 1024),
463        thread_count: status_value(&status, "Threads:", 1),
464    }
465}
466
467#[cfg(not(target_os = "linux"))]
468fn process_resources() -> ResourceSnapshot {
469    let Ok(pid) = sysinfo::get_current_pid() else {
470        return ResourceSnapshot::default();
471    };
472    let mut system = System::new();
473    let pids = [pid];
474    system.refresh_processes(ProcessesToUpdate::Some(&pids), true);
475    let Some(process) = system.process(pid) else {
476        return ResourceSnapshot::default();
477    };
478    ResourceSnapshot {
479        cpu_time_ms: Some(process.accumulated_cpu_time()),
480        resident_bytes: Some(process.memory()),
481        virtual_bytes: Some(process.virtual_memory()),
482        thread_count: process.tasks().map(|tasks| tasks.len() as u64),
483    }
484}
485
486/// Resource change across a completed profile session.
487#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
488pub struct ResourceUsage {
489    pub start: ResourceSnapshot,
490    pub finish: ResourceSnapshot,
491    pub max_observed_resident_bytes: Option<u64>,
492    pub max_observed_threads: Option<u64>,
493    pub aggregate_cpu_percent: Option<f64>,
494}
495
496fn max_option(left: Option<u64>, right: Option<u64>) -> Option<u64> {
497    match (left, right) {
498        (Some(left), Some(right)) => Some(left.max(right)),
499        (left, right) => left.or(right),
500    }
501}
502
503fn cpu_percent(
504    start_cpu_ms: Option<u64>,
505    finish_cpu_ms: Option<u64>,
506    elapsed_ns: u64,
507) -> Option<f64> {
508    start_cpu_ms
509        .zip(finish_cpu_ms)
510        .filter(|(start, finish)| finish >= start)
511        .and_then(|(start, finish)| {
512            let wall_ms = elapsed_ns as f64 / 1_000_000.0;
513            (wall_ms > 0.0).then_some((finish - start) as f64 * 100.0 / wall_ms)
514        })
515}
516
517fn cpu_milli_percent(
518    start_cpu_ms: Option<u64>,
519    finish_cpu_ms: Option<u64>,
520    elapsed_ns: u64,
521) -> Option<u64> {
522    if elapsed_ns == 0 {
523        return None;
524    }
525    let (start, finish) = start_cpu_ms
526        .zip(finish_cpu_ms)
527        .filter(|(start, finish)| finish >= start)?;
528    let numerator = u128::from(finish - start) * 100_000_000_000_u128;
529    Some(u64::try_from(numerator / u128::from(elapsed_ns)).unwrap_or(u64::MAX))
530}
531
532fn resource_usage(
533    start: ResourceSnapshot,
534    finish: ResourceSnapshot,
535    wall: Duration,
536    samples: &[ResourceSample],
537) -> ResourceUsage {
538    let aggregate_cpu_percent = cpu_percent(
539        start.cpu_time_ms,
540        finish.cpu_time_ms,
541        u64::try_from(wall.as_nanos()).unwrap_or(u64::MAX),
542    );
543    let max_observed_resident_bytes = samples
544        .iter()
545        .filter_map(|sample| sample.snapshot.resident_bytes)
546        .fold(
547            max_option(start.resident_bytes, finish.resident_bytes),
548            |maximum, value| max_option(maximum, Some(value)),
549        );
550    let max_observed_threads = samples
551        .iter()
552        .filter_map(|sample| sample.snapshot.thread_count)
553        .fold(
554            max_option(start.thread_count, finish.thread_count),
555            |maximum, value| max_option(maximum, Some(value)),
556        );
557    ResourceUsage {
558        max_observed_resident_bytes,
559        max_observed_threads,
560        start,
561        finish,
562        aggregate_cpu_percent,
563    }
564}
565
566fn state_measurements(
567    transitions: &[StateTransition],
568    samples: &[ResourceSample],
569) -> Vec<StateMeasurement> {
570    transitions
571        .windows(2)
572        .zip(samples.windows(2))
573        .filter_map(|(transition, sample)| {
574            let elapsed_ns = transition[1]
575                .elapsed_ns
576                .checked_sub(transition[0].elapsed_ns)?;
577            let start = sample[0].snapshot;
578            let finish = sample[1].snapshot;
579            let cpu_time_ms = start
580                .cpu_time_ms
581                .zip(finish.cpu_time_ms)
582                .filter(|(start, finish)| finish >= start)
583                .map(|(start, finish)| finish - start);
584            Some(StateMeasurement {
585                state: transition[0].state,
586                elapsed_ns,
587                cpu_time_ms,
588                aggregate_cpu_milli_percent: cpu_milli_percent(
589                    start.cpu_time_ms,
590                    finish.cpu_time_ms,
591                    elapsed_ns,
592                ),
593                resident_start_bytes: start.resident_bytes,
594                resident_end_bytes: finish.resident_bytes,
595                threads_start: start.thread_count,
596                threads_end: finish.thread_count,
597            })
598        })
599        .collect()
600}
601
602/// Complete replayable profile record.
603#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
604pub struct RunProfile {
605    pub schema: String,
606    pub identity: RunIdentity,
607    pub status: RunState,
608    pub wall_time_ns: u64,
609    pub input_bytes: u64,
610    pub input_units: u64,
611    pub stages: Vec<StageMeasurement>,
612    pub transitions: Vec<StateTransition>,
613    #[serde(default)]
614    pub states: Vec<StateMeasurement>,
615    pub resource_samples: Vec<ResourceSample>,
616    pub resources: ResourceUsage,
617}
618
619impl RunProfile {
620    /// Serialize the stable record as pretty JSON.
621    pub fn to_json_pretty(&self) -> serde_json::Result<String> {
622        serde_json::to_string_pretty(self)
623    }
624
625    /// Render a compact operator report without secrets or source content.
626    pub fn render_text(&self) -> String {
627        let reader_threads = self
628            .identity
629            .reader_threads
630            .map_or_else(|| "auto".to_owned(), |threads| threads.to_string());
631        let throughput_mib_s = if self.wall_time_ns == 0 {
632            0.0
633        } else {
634            self.input_bytes as f64 * 1_000_000_000.0 / self.wall_time_ns as f64 / (1024.0 * 1024.0)
635        };
636        let mut output = format!(
637            "KeyHog profile {}\n\
638             state={} source={} workload={} backend_requested={} backend_selected={} cache={} daemon={} wall_ms={:.3}\n\
639             version={} detector_digest={} config_digest={}\n\
640             input_bytes={} input_units={} throughput_mib_s={throughput_mib_s:.3} scanner_threads={} reader_threads={} logical_cpus={}\n",
641            self.identity.run_id,
642            state_name(self.status),
643            self.identity.source_kind,
644            self.identity.workload_class,
645            self.identity.backend_requested,
646            self.identity
647                .backend_selected
648                .as_deref()
649                .unwrap_or("unselected"),
650            cache_name(self.identity.cache_state),
651            daemon_name(self.identity.daemon_state),
652            self.wall_time_ns as f64 / 1_000_000.0,
653            self.identity.binary_version,
654            self.identity.detector_digest,
655            self.identity.config_digest,
656            self.input_bytes,
657            self.input_units,
658            self.identity.scanner_threads,
659            reader_threads,
660            self.identity.logical_cpus,
661        );
662        for state in &self.states {
663            output.push_str(&format!(
664                "macro {:<12} wall_ms={:.3}",
665                state_name(state.state),
666                state.elapsed_ns as f64 / 1_000_000.0,
667            ));
668            if let Some(cpu) = state.aggregate_cpu_milli_percent {
669                output.push_str(&format!(" cpu={:.1}%", cpu as f64 / 1_000.0));
670            } else {
671                output.push_str(" cpu=unavailable");
672            }
673            if let (Some(start), Some(finish)) =
674                (state.resident_start_bytes, state.resident_end_bytes)
675            {
676                output.push_str(&format!(" rss_bytes={start}->{finish}"));
677            }
678            if let (Some(start), Some(finish)) = (state.threads_start, state.threads_end) {
679                output.push_str(&format!(" threads={start}->{finish}"));
680            }
681            output.push('\n');
682        }
683        for stage in &self.stages {
684            let per_call_us = if stage.calls == 0 {
685                0.0
686            } else {
687                stage.elapsed_ns as f64 / stage.calls as f64 / 1_000.0
688            };
689            output.push_str(&format!(
690                "  {:<24} {:>10.3} ms calls={} per_call_us={per_call_us:.3} attributed_ms={:.3}\n",
691                stage.stage.as_str(),
692                stage.elapsed_ns as f64 / 1_000_000.0,
693                stage.calls,
694                stage.attributed_ns as f64 / 1_000_000.0,
695            ));
696        }
697        if let Some(state) = self.states.iter().max_by_key(|state| state.elapsed_ns) {
698            output.push_str(&format!(
699                "bottleneck macro={} wall_ms={:.3}",
700                state_name(state.state),
701                state.elapsed_ns as f64 / 1_000_000.0,
702            ));
703        }
704        if let Some(stage) = self
705            .stages
706            .iter()
707            .filter(|stage| stage.stage != Stage::BackendDispatch)
708            .max_by_key(|stage| stage.elapsed_ns)
709        {
710            output.push_str(&format!(
711                " summed_stage={} summed_ms={:.3}",
712                stage.stage.as_str(),
713                stage.elapsed_ns as f64 / 1_000_000.0,
714            ));
715        }
716        if !self.states.is_empty() || !self.stages.is_empty() {
717            output.push('\n');
718        }
719        if let Some(cpu) = self.resources.aggregate_cpu_percent {
720            output.push_str(&format!("resources aggregate_cpu={cpu:.1}%"));
721        } else {
722            output.push_str("resources aggregate_cpu=unavailable");
723        }
724        if let Some(rss) = self.resources.max_observed_resident_bytes {
725            output.push_str(&format!(" max_observed_rss_bytes={rss}"));
726        }
727        if let Some(threads) = self.resources.max_observed_threads {
728            output.push_str(&format!(" max_observed_threads={threads}"));
729        }
730        output.push('\n');
731        output
732    }
733}
734
735fn state_name(state: RunState) -> &'static str {
736    match state {
737        RunState::Created => "created",
738        RunState::Acquiring => "acquiring",
739        RunState::Scanning => "scanning",
740        RunState::Resolving => "resolving",
741        RunState::Verifying => "verifying",
742        RunState::Reporting => "reporting",
743        RunState::Completed => "completed",
744        RunState::Failed => "failed",
745    }
746}
747
748fn cache_name(state: CacheState) -> &'static str {
749    match state {
750        CacheState::Unknown => "unknown",
751        CacheState::Disabled => "disabled",
752        CacheState::Cold => "cold",
753        CacheState::Warm => "warm",
754    }
755}
756
757fn daemon_name(state: DaemonState) -> &'static str {
758    match state {
759        DaemonState::Off => "off",
760        DaemonState::Client => "client",
761        DaemonState::Worker => "worker",
762        DaemonState::Mass => "mass",
763    }
764}
765
766/// Error returned when a process-global profile session is already active.
767#[derive(Clone, Copy, Debug, Eq, PartialEq)]
768pub struct SessionActive;
769
770impl fmt::Display for SessionActive {
771    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
772        formatter.write_str("a KeyHog profile session is already active in this process")
773    }
774}
775
776impl std::error::Error for SessionActive {}
777
778/// One causal profiling session.
779///
780/// Sessions are process-global because deep scanner spans use allocation-free
781/// static counters. Concurrent daemon requests must profile separately rather
782/// than merge unrelated state into one record.
783pub struct Session {
784    identity: Option<RunIdentity>,
785    started: Instant,
786    resources_at_start: ResourceSnapshot,
787    transitions: Vec<StateTransition>,
788    resource_samples: Vec<ResourceSample>,
789    finished: bool,
790}
791
792impl Session {
793    /// Start a fresh session and reset all fixed-stage counters.
794    pub fn start(identity: RunIdentity) -> Result<Self, SessionActive> {
795        SESSION_ACTIVE
796            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
797            .map_err(|_| SessionActive)?;
798        reset();
799        reset_session();
800        set_enabled(true);
801        let started = Instant::now();
802        let resources_at_start = process_resources();
803        Ok(Self {
804            identity: Some(identity),
805            started,
806            resources_at_start,
807            transitions: vec![StateTransition {
808                state: RunState::Created,
809                elapsed_ns: 0,
810            }],
811            resource_samples: vec![ResourceSample {
812                state: RunState::Created,
813                elapsed_ns: 0,
814                snapshot: resources_at_start,
815            }],
816            finished: false,
817        })
818    }
819
820    /// Mutate run identity before the session is finalized.
821    pub fn identity_mut(&mut self) -> &mut RunIdentity {
822        self.identity
823            .as_mut()
824            .expect("unfinished profile owns identity")
825    }
826
827    /// Record an explicit macro state transition.
828    pub fn transition(&mut self, state: RunState) {
829        let elapsed_ns = u64::try_from(self.started.elapsed().as_nanos()).unwrap_or(u64::MAX);
830        self.transitions.push(StateTransition { state, elapsed_ns });
831        self.resource_samples.push(ResourceSample {
832            state,
833            elapsed_ns,
834            snapshot: process_resources(),
835        });
836    }
837
838    /// Finish the session and return its complete structured record.
839    pub fn finish(mut self, status: RunState) -> RunProfile {
840        self.transition(status);
841        set_enabled(false);
842        let wall = self.started.elapsed();
843        let finish_resources = self
844            .resource_samples
845            .last()
846            .map_or(self.resources_at_start, |sample| sample.snapshot);
847        let (input_bytes, input_units) = take_session_input_totals();
848        let stages = take_session_stage_measurements();
849        reset();
850        let resource_samples = std::mem::take(&mut self.resource_samples);
851        let states = state_measurements(&self.transitions, &resource_samples);
852        let resources = resource_usage(
853            self.resources_at_start,
854            finish_resources,
855            wall,
856            &resource_samples,
857        );
858        let profile = RunProfile {
859            schema: PROFILE_SCHEMA.to_string(),
860            identity: self
861                .identity
862                .take()
863                .expect("unfinished profile owns identity"),
864            status,
865            wall_time_ns: u64::try_from(wall.as_nanos()).unwrap_or(u64::MAX),
866            input_bytes,
867            input_units,
868            stages,
869            transitions: std::mem::take(&mut self.transitions),
870            states,
871            resource_samples,
872            resources,
873        };
874        self.finished = true;
875        SESSION_ACTIVE.store(false, Ordering::Release);
876        profile
877    }
878}
879
880impl Drop for Session {
881    fn drop(&mut self) {
882        if !self.finished {
883            set_enabled(false);
884            reset();
885            reset_session();
886            SESSION_ACTIVE.store(false, Ordering::Release);
887        }
888    }
889}
keyhog_profile/lib.rs

keyhog_profile/
lib.rs
