Skip to main content

resource_tracker/collector/
cpu.rs

1use crate::metrics::CpuMetrics;
2use procfs::prelude::*;
3use procfs::process::all_processes;
4use procfs::{CpuTime, KernelStats};
5use std::collections::{HashMap, HashSet};
6use std::time::Instant;
7
8type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
9
10// ---------------------------------------------------------------------------
11// Cgroup CPU source detection and reading
12// ---------------------------------------------------------------------------
13
14/// Which CPU accounting source is available for system-level utilization.
15#[derive(Debug, Clone, Copy, PartialEq)]
16enum CpuSource {
17    /// cgroupv2 unified hierarchy: read usage_usec from cpu.stat
18    CgroupV2,
19    /// cgroupv1 cpuacct controller: read cpuacct.usage (nanoseconds)
20    CgroupV1,
21    /// Bare /proc/stat (host or no cgroup access)
22    ProcStat,
23}
24
25impl CpuSource {
26    fn is_cgroup(self) -> bool {
27        !matches!(self, CpuSource::ProcStat)
28    }
29}
30
31/// Effective CPU limit from CFS quota (None = unlimited).
32#[derive(Debug, Clone, Copy)]
33struct CfsQuota {
34    /// Maximum fractional cores allowed (e.g. 1.5 for --cpus=1.5)
35    max_cores: Option<f64>,
36}
37
38/// Detect the best available CPU accounting source.
39/// Preference: cgroupv2 > cgroupv1 > /proc/stat
40#[allow(clippy::collapsible_if)]
41fn detect_cpu_source() -> CpuSource {
42    // cgroupv2: unified hierarchy exposes cpu.stat at the cgroup root
43    if let Ok(contents) = std::fs::read_to_string("/sys/fs/cgroup/cpu.stat") {
44        if contents.contains("usage_usec") {
45            return CpuSource::CgroupV2;
46        }
47    }
48    // cgroupv1: cpuacct controller (various mount points)
49    for path in &[
50        "/sys/fs/cgroup/cpuacct/cpuacct.usage",
51        "/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage",
52        "/sys/fs/cgroup/cpu/cpuacct.usage",
53    ] {
54        if std::fs::read_to_string(path).is_ok() {
55            return CpuSource::CgroupV1;
56        }
57    }
58    CpuSource::ProcStat
59}
60
61/// Read the CFS quota to determine effective core limit.
62#[allow(clippy::collapsible_if)]
63fn detect_cfs_quota() -> CfsQuota {
64    // cgroupv2: cpu.max contains "quota period" or "max period"
65    if let Ok(contents) = std::fs::read_to_string("/sys/fs/cgroup/cpu.max") {
66        let parts: Vec<&str> = contents.split_whitespace().collect();
67        if parts.len() == 2 && parts[0] != "max" {
68            if let (Ok(quota), Ok(period)) = (parts[0].parse::<f64>(), parts[1].parse::<f64>()) {
69                if period > 0.0 {
70                    return CfsQuota {
71                        max_cores: Some(quota / period),
72                    };
73                }
74            }
75        }
76    }
77    // cgroupv1: cpu.cfs_quota_us and cpu.cfs_period_us
78    for prefix in &[
79        "/sys/fs/cgroup/cpu",
80        "/sys/fs/cgroup/cpu,cpuacct",
81        "/sys/fs/cgroup/cpuacct",
82    ] {
83        let quota_path = format!("{}/cpu.cfs_quota_us", prefix);
84        let period_path = format!("{}/cpu.cfs_period_us", prefix);
85        if let (Ok(q_str), Ok(p_str)) = (
86            std::fs::read_to_string(&quota_path),
87            std::fs::read_to_string(&period_path),
88        ) {
89            if let (Ok(quota), Ok(period)) =
90                (q_str.trim().parse::<i64>(), p_str.trim().parse::<i64>())
91            {
92                // quota == -1 means unlimited
93                if quota > 0 && period > 0 {
94                    return CfsQuota {
95                        max_cores: Some(quota as f64 / period as f64),
96                    };
97                }
98            }
99        }
100    }
101    CfsQuota { max_cores: None }
102}
103
104/// Read cgroupv2 cpu.stat usage_usec (microseconds, cumulative).
105fn read_cgroupv2_usage_usec() -> Option<u64> {
106    let contents = std::fs::read_to_string("/sys/fs/cgroup/cpu.stat").ok()?;
107    for line in contents.lines() {
108        if let Some(val) = line.strip_prefix("usage_usec ") {
109            return val.trim().parse().ok();
110        }
111    }
112    None
113}
114
115/// Read cgroupv1 cpuacct.usage (nanoseconds, cumulative).
116#[allow(clippy::collapsible_if)]
117fn read_cgroupv1_usage_ns() -> Option<u64> {
118    for path in &[
119        "/sys/fs/cgroup/cpuacct/cpuacct.usage",
120        "/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage",
121        "/sys/fs/cgroup/cpu/cpuacct.usage",
122    ] {
123        if let Ok(contents) = std::fs::read_to_string(path)
124            && let Ok(val) = contents.trim().parse()
125        {
126            return Some(val);
127        }
128    }
129    None
130}
131
132/// Read cgroup CPU usage as fractional seconds (cumulative).
133/// Returns None if the detected source is ProcStat or reads fail.
134fn read_cgroup_usage_secs(source: CpuSource) -> Option<f64> {
135    match source {
136        CpuSource::CgroupV2 => read_cgroupv2_usage_usec().map(|usec| usec as f64 / 1_000_000.0),
137        CpuSource::CgroupV1 => read_cgroupv1_usage_ns().map(|ns| ns as f64 / 1_000_000_000.0),
138        CpuSource::ProcStat => None,
139    }
140}
141
142// ---------------------------------------------------------------------------
143// Tick helpers
144// ---------------------------------------------------------------------------
145
146fn cpu_total(c: &CpuTime) -> u64 {
147    c.user
148        + c.nice
149        + c.system
150        + cpu_idle(c)
151        + c.irq.unwrap_or(0)
152        + c.softirq.unwrap_or(0)
153        + c.steal.unwrap_or(0)
154}
155
156fn cpu_idle(c: &CpuTime) -> u64 {
157    c.idle + c.iowait.unwrap_or(0)
158}
159
160/// Per-core utilization percentage (0.0–100.0, clamped).
161fn core_util_pct(prev: &CpuTime, curr: &CpuTime) -> f64 {
162    util_pct_from_ticks(
163        cpu_total(prev),
164        cpu_idle(prev),
165        cpu_total(curr),
166        cpu_idle(curr),
167    )
168    .clamp(0.0, 100.0)
169}
170
171/// Aggregate utilization expressed as fractional cores in use (0.0..n_cores).
172/// Not clamped: kernel rounding can produce values very slightly above n_cores.
173fn aggregate_util_cores(prev: &CpuTime, curr: &CpuTime, n_cores: usize) -> f64 {
174    util_pct_from_ticks(
175        cpu_total(prev),
176        cpu_idle(prev),
177        cpu_total(curr),
178        cpu_idle(curr),
179    ) / 100.0
180        * n_cores as f64
181}
182
183/// Pure math: percentage of non-idle ticks between two snapshots (0.0–100.0
184/// before any clamping).  Takes raw pre-computed totals/idles so it can be
185/// unit-tested without constructing a `CpuTime` (which has private fields).
186fn util_pct_from_ticks(prev_total: u64, prev_idle: u64, curr_total: u64, curr_idle: u64) -> f64 {
187    let delta_total = curr_total.saturating_sub(prev_total) as f64;
188    let delta_idle = curr_idle.saturating_sub(prev_idle) as f64;
189    if delta_total == 0.0 {
190        return 0.0;
191    }
192    (delta_total - delta_idle) / delta_total * 100.0
193}
194
195// ---------------------------------------------------------------------------
196// Process-tree helpers
197// ---------------------------------------------------------------------------
198
199/// Returns a map of { pid to (utime, stime) } for every process in the tree
200/// rooted at `root_pid` (root included).  Processes that have already exited
201/// are silently skipped: this is a TOCTOU race we accept.
202fn process_tree_ticks(root_pid: i32) -> HashMap<i32, (u64, u64)> {
203    // Collect all readable processes in one pass.
204    let all: Vec<_> = match all_processes() {
205        Ok(iter) => iter.filter_map(|r| r.ok()).collect(),
206        Err(_) => return HashMap::new(),
207    };
208
209    // Single .stat() read per process: build both the parent->children map and
210    // the pid->(utime+cutime, stime+cstime) map in one pass to halve /proc I/O.
211    //
212    // cutime/cstime (CPU time of waited-for children) is included so that
213    // short-lived child processes that both start AND exit within a single
214    // measurement interval are still captured: once a child is reaped its
215    // ticks roll up into the parent's cutime/cstime.
216    //
217    // Double-counting guard: if a process was alive at the previous snapshot
218    // and exits before the current one, its pre-snapshot ticks are already in
219    // prev_proc_ticks AND will re-appear via the parent's cutime delta.
220    // CpuCollector::collect() subtracts the prev ticks of all such exited
221    // processes to cancel that overcounting.
222    let mut children: HashMap<i32, Vec<i32>> = HashMap::new();
223    let ticks_for: HashMap<i32, (u64, u64)> = all
224        .iter()
225        .filter_map(|proc| {
226            proc.stat().ok().map(|s| {
227                children.entry(s.ppid).or_default().push(proc.pid);
228                let user = s.utime + u64::try_from(s.cutime).unwrap_or(0);
229                let system = s.stime + u64::try_from(s.cstime).unwrap_or(0);
230                (proc.pid, (user, system))
231            })
232        })
233        .collect();
234
235    // BFS from root_pid, collecting (utime, stime) for every reachable node.
236    let mut result = HashMap::new();
237    let mut queue = vec![root_pid];
238    while let Some(pid) = queue.pop() {
239        if let Some(&ticks) = ticks_for.get(&pid) {
240            result.insert(pid, ticks);
241        }
242        if let Some(kids) = children.get(&pid) {
243            queue.extend(kids);
244        }
245    }
246    result
247}
248
249/// Sum of PSS and VmRSS across all given PIDs, each converted to MiB.
250/// One `Process::open` per PID reads both sources. PSS matches Python
251/// `memory_mib`; RSS is retained for consumers that need resident set size.
252fn process_tree_memory_mib(pids: &[i32]) -> (u64, u64) {
253    let mut pss_kib = 0u64;
254    let mut rss_kib = 0u64;
255    for &pid in pids {
256        let Some(proc_) = procfs::process::Process::new(pid).ok() else {
257            continue;
258        };
259        if let Ok(rollup) = proc_.smaps_rollup()
260            && let Some(bytes) = rollup
261                .memory_map_rollup
262                .iter()
263                .find_map(|m| m.extension.map.get("Pss").copied())
264        {
265            pss_kib += bytes / 1024;
266        }
267        if let Ok(status) = proc_.status()
268            && let Some(vmrss) = status.vmrss
269        {
270            rss_kib += vmrss;
271        }
272    }
273    (pss_kib / 1024, rss_kib / 1024)
274}
275
276/// Per-process cumulative disk I/O bytes from /proc/pid/io.
277/// Returns { pid -> (read_bytes, write_bytes) }.
278/// PIDs whose /proc/pid/io is unreadable (e.g. different UID without ptrace)
279/// are silently omitted -- the delta for those PIDs will be 0.
280fn process_tree_io(pids: &[i32]) -> HashMap<i32, (u64, u64)> {
281    pids.iter()
282        .filter_map(|&pid| {
283            let io = procfs::process::Process::new(pid).ok()?.io().ok()?;
284            Some((pid, (io.read_bytes, io.write_bytes)))
285        })
286        .collect()
287}
288
289// ---------------------------------------------------------------------------
290// Snapshot + Collector
291// ---------------------------------------------------------------------------
292
293struct Snapshot {
294    /// Aggregate across all logical CPUs (the "cpu" summary line in /proc/stat).
295    total: CpuTime,
296    /// Per-logical-CPU entries (cpu0, cpu1, …).
297    per_core: Vec<CpuTime>,
298    /// Wall-clock time after all /proc reads; used as the Python-style
299    /// snapshot timestamp for process CPU rate (Δcpu_secs / Δtimestamp).
300    instant: Instant,
301    /// Cgroup cumulative CPU usage in fractional seconds (if available).
302    cgroup_usage_secs: Option<f64>,
303    /// { pid -> (utime, stime) } for root process + all descendants.
304    /// Empty when no PID is being tracked.
305    proc_ticks: HashMap<i32, (u64, u64)>,
306    /// { pid -> (read_bytes, write_bytes) } from /proc/pid/io.
307    /// Empty when no PID is tracked or /proc/pid/io is unreadable.
308    proc_io: HashMap<i32, (u64, u64)>,
309}
310
311pub struct CpuCollector {
312    /// Root PID of the process tree to track. None = system-only metrics.
313    pid: Option<i32>,
314    prev: Option<Snapshot>,
315    /// Detected CPU accounting source for system-level utilization.
316    cpu_source: CpuSource,
317    /// CFS quota limit (None = unlimited).
318    cfs_quota: CfsQuota,
319    /// Effective number of cores for this environment.
320    /// Respects CFS quota: min(physical_cores, quota_cores).
321    effective_cores: f64,
322    /// PIDs whose prev entries were carried forward from the previous
323    /// interval (their `/proc/PID/stat` read failed).  Limited to one
324    /// hop so dead PIDs don't accumulate and inflate the exited correction.
325    carried_forward: HashSet<i32>,
326    /// Use aggregated value or per-CPU value. Comes from CLI arg or config.
327    aggregate_cpu_steal: bool,
328}
329
330impl CpuCollector {
331    pub fn new(pid: Option<i32>, aggregate_cpu_steal: bool) -> Self {
332        let cpu_source = detect_cpu_source();
333        let cfs_quota = detect_cfs_quota();
334
335        // Determine effective core count: physical cores capped by CFS quota.
336        let physical_cores = KernelStats::current()
337            .map(|s| s.cpu_time.len())
338            .unwrap_or(1) as f64;
339        let effective_cores = match cfs_quota.max_cores {
340            Some(quota) => physical_cores.min(quota),
341            None => physical_cores,
342        };
343
344        Self {
345            pid,
346            prev: None,
347            cpu_source,
348            cfs_quota,
349            effective_cores,
350            carried_forward: HashSet::new(),
351            aggregate_cpu_steal,
352        }
353    }
354
355    /// Set the root PID for process-tree metrics (called after shell-wrapper spawn).
356    pub fn set_tracked_pid(&mut self, pid: Option<i32>) {
357        self.pid = pid;
358    }
359
360    pub fn collect(&mut self) -> Result<CpuMetrics> {
361        let tps = procfs::ticks_per_second() as f64;
362        let process_count = self.read_process_count();
363
364        // Read system and process data in correct order
365        let stats = KernelStats::current()?;
366        let cgroup_usage_secs = read_cgroup_usage_secs(self.cpu_source);
367        let proc_ticks = self.read_process_ticks();
368        let now = Instant::now();
369
370        let proc_io = self.read_process_io(&proc_ticks);
371        let (process_pss_mib, process_rss_mib) = self.read_process_memory(&proc_ticks);
372
373        let mut curr = Snapshot {
374            total: stats.total,
375            per_core: stats.cpu_time,
376            instant: now,
377            cgroup_usage_secs,
378            proc_ticks,
379            proc_io,
380        };
381
382        let metrics = match &self.prev {
383            None => {
384                self.build_first_metrics(&curr, process_count, process_pss_mib, process_rss_mib)
385            }
386            Some(prev) => self.build_subsequent_metrics_with_deltas(
387                prev,
388                &mut curr,
389                process_count,
390                tps,
391                process_pss_mib,
392                process_rss_mib,
393            ),
394        };
395
396        self.carry_forward_entries(&mut curr);
397        self.prev = Some(curr);
398        Ok(metrics)
399    }
400
401    fn read_process_count(&self) -> u32 {
402        let Ok(proc_dir) = std::fs::read_dir("/proc") else {
403            return 0;
404        };
405
406        let process_count = proc_dir
407            .filter_map(|entry| entry.ok())
408            .filter(|e| Self::is_pid_directory(e))
409            .count();
410
411        u32::try_from(process_count).unwrap_or(0)
412    }
413
414    fn is_pid_directory(entry: &std::fs::DirEntry) -> bool {
415        entry
416            .file_name()
417            .to_string_lossy()
418            .chars()
419            .all(|c| c.is_ascii_digit())
420    }
421
422    fn read_process_ticks(&self) -> HashMap<i32, (u64, u64)> {
423        match self.pid {
424            Some(root) => process_tree_ticks(root),
425            None => HashMap::new(),
426        }
427    }
428
429    fn read_process_io(&self, proc_ticks: &HashMap<i32, (u64, u64)>) -> HashMap<i32, (u64, u64)> {
430        if self.pid.is_some() {
431            let pids: Vec<i32> = proc_ticks.keys().copied().collect();
432            process_tree_io(&pids)
433        } else {
434            HashMap::new()
435        }
436    }
437
438    fn read_process_memory(
439        &self,
440        proc_ticks: &HashMap<i32, (u64, u64)>,
441    ) -> (Option<u64>, Option<u64>) {
442        if self.pid.is_some() {
443            let pids: Vec<i32> = proc_ticks.keys().copied().collect();
444            let (pss, rss) = process_tree_memory_mib(&pids);
445            (Some(pss), Some(rss))
446        } else {
447            (None, None)
448        }
449    }
450
451    fn build_first_metrics(
452        &self,
453        curr: &Snapshot,
454        process_count: u32,
455        process_pss_mib: Option<u64>,
456        process_rss_mib: Option<u64>,
457    ) -> CpuMetrics {
458        CpuMetrics {
459            utilization_pct: 0.0,
460            cgroup_utilization_pct: curr
461                .cgroup_usage_secs
462                .filter(|_| self.cpu_source.is_cgroup())
463                .map(|_| 0.0),
464            cgroup_usage_secs: curr
465                .cgroup_usage_secs
466                .filter(|_| self.cpu_source.is_cgroup())
467                .map(|_| 0.0),
468            per_core_pct: vec![0.0; curr.per_core.len()],
469            utime_secs: 0.0,
470            stime_secs: 0.0,
471            steal_time_secs: 0.0,
472            steal_time_pct: 0.0,
473            per_core_steal_time_pct: if self.aggregate_cpu_steal {
474                vec![]
475            } else {
476                vec![0.0; curr.per_core.len()]
477            },
478            process_count,
479            process_cores_used: self.pid.map(|_| 0.0),
480            process_child_count: self
481                .pid
482                .map(|_| u32::try_from(curr.proc_ticks.len().saturating_sub(1)).unwrap_or(0)),
483            process_utime_secs: self.pid.map(|_| 0.0),
484            process_stime_secs: self.pid.map(|_| 0.0),
485            process_pss_mib,
486            process_rss_mib,
487            process_disk_read_bytes: self.pid.map(|_| 0),
488            process_disk_write_bytes: self.pid.map(|_| 0),
489            process_gpu_usage: None,
490            process_gpu_vram_mib: None,
491            process_gpu_utilized: None,
492            process_tree_pids: curr.proc_ticks.keys().copied().collect(),
493        }
494    }
495
496    fn build_subsequent_metrics_with_deltas(
497        &self,
498        prev: &Snapshot,
499        curr: &mut Snapshot,
500        process_count: u32,
501        tps: f64,
502        process_pss_mib: Option<u64>,
503        process_rss_mib: Option<u64>,
504    ) -> CpuMetrics {
505        let n_cores = curr.per_core.len();
506        let elapsed = (curr.instant - prev.instant).as_secs_f64().max(0.001);
507
508        let (utime_secs, stime_secs) = self.calculate_system_cpu_deltas(prev, curr, tps);
509        let per_core_pct = self.calculate_per_core_utilization(prev, curr);
510        let utilization_pct = aggregate_util_cores(&prev.total, &curr.total, n_cores);
511        let (cgroup_usage_secs, cgroup_utilization_pct) =
512            self.calculate_cgroup_usage(prev, curr, elapsed);
513        let (exited_utime, exited_stime) = self.calculate_exited_child_ticks(prev, curr);
514
515        let process_utime_secs = self.calculate_process_utime_delta(prev, curr, exited_utime, tps);
516        let process_stime_secs = self.calculate_process_stime_delta(prev, curr, exited_stime, tps);
517
518        let steal_time_secs = self.calculate_steal_time_secs(prev, curr, tps);
519        let steal_time_pct = self.calculate_steal_time_pct(prev, curr);
520        let per_core_steal_time_pct = if self.aggregate_cpu_steal {
521            vec![]
522        } else {
523            self.calculate_per_core_steal_pct(prev, curr)
524        };
525
526        let process_cores_used = self.calculate_process_cores_used_with_caps(
527            &process_utime_secs,
528            &process_stime_secs,
529            prev,
530            curr,
531            elapsed,
532            n_cores,
533        );
534
535        let process_disk_read_bytes = self.calculate_disk_read_delta(prev, curr);
536        let process_disk_write_bytes = self.calculate_disk_write_delta(prev, curr);
537
538        CpuMetrics {
539            utilization_pct,
540            cgroup_utilization_pct,
541            cgroup_usage_secs,
542            per_core_pct,
543            utime_secs,
544            stime_secs,
545            steal_time_secs,
546            steal_time_pct,
547            per_core_steal_time_pct,
548            process_count,
549            process_cores_used,
550            process_child_count: self
551                .pid
552                .map(|_| u32::try_from(curr.proc_ticks.len().saturating_sub(1)).unwrap_or(0)),
553            process_utime_secs,
554            process_stime_secs,
555            process_pss_mib,
556            process_rss_mib,
557            process_disk_read_bytes,
558            process_disk_write_bytes,
559            process_gpu_usage: None,
560            process_gpu_vram_mib: None,
561            process_gpu_utilized: None,
562            process_tree_pids: curr.proc_ticks.keys().copied().collect(),
563        }
564    }
565
566    fn calculate_system_cpu_deltas(
567        &self,
568        prev: &Snapshot,
569        curr: &Snapshot,
570        tps: f64,
571    ) -> (f64, f64) {
572        let utime_secs = (curr.total.user + curr.total.nice)
573            .saturating_sub(prev.total.user + prev.total.nice) as f64
574            / tps;
575        let stime_secs = curr.total.system.saturating_sub(prev.total.system) as f64 / tps;
576        (utime_secs, stime_secs)
577    }
578
579    fn calculate_per_core_utilization(&self, prev: &Snapshot, curr: &Snapshot) -> Vec<f64> {
580        prev.per_core
581            .iter()
582            .zip(curr.per_core.iter())
583            .map(|(p, c)| core_util_pct(p, c))
584            .collect()
585    }
586
587    fn calculate_cgroup_usage(
588        &self,
589        prev: &Snapshot,
590        curr: &Snapshot,
591        elapsed: f64,
592    ) -> (Option<f64>, Option<f64>) {
593        match (curr.cgroup_usage_secs, prev.cgroup_usage_secs) {
594            (Some(curr_cg), Some(prev_cg)) => {
595                let delta = (curr_cg - prev_cg).max(0.0);
596                let cores_used = delta / elapsed;
597                (Some(delta), Some(cores_used.min(self.effective_cores)))
598            }
599            _ => (None, None),
600        }
601    }
602
603    // Calculate exited child ticks for double-counting correction
604    fn calculate_exited_child_ticks(&self, prev: &Snapshot, curr: &Snapshot) -> (u64, u64) {
605        if self.pid.is_none() {
606            return (0, 0);
607        }
608
609        prev.proc_ticks
610            .iter()
611            .filter(|(pid, _)| !curr.proc_ticks.contains_key(*pid))
612            .fold((0, 0), |(user_sum, sys_sum), (_, &(user, sys))| {
613                (user_sum + user, sys_sum + sys)
614            })
615    }
616    fn calculate_process_utime_delta(
617        &self,
618        prev: &Snapshot,
619        curr: &Snapshot,
620        exited_utime: u64,
621        tps: f64,
622    ) -> Option<f64> {
623        if self.pid.is_none() {
624            return None;
625        }
626
627        let raw: u64 = curr
628            .proc_ticks
629            .iter()
630            .map(|(pid, &(cu, _))| {
631                let pu = prev.proc_ticks.get(pid).map(|&(u, _)| u).unwrap_or(cu);
632                cu.saturating_sub(pu)
633            })
634            .sum();
635
636        let adjusted = if exited_utime <= raw {
637            raw - exited_utime
638        } else {
639            raw
640        };
641
642        Some(adjusted as f64 / tps)
643    }
644
645    fn calculate_process_stime_delta(
646        &self,
647        prev: &Snapshot,
648        curr: &Snapshot,
649        exited_stime: u64,
650        tps: f64,
651    ) -> Option<f64> {
652        if self.pid.is_none() {
653            return None;
654        }
655
656        let mut raw: u64 = 0;
657        for (pid, &(_, cs)) in &curr.proc_ticks {
658            let ps = prev.proc_ticks.get(pid).map(|&(_, s)| s).unwrap_or(cs);
659            raw += cs.saturating_sub(ps);
660        }
661
662        let adjusted = if exited_stime <= raw {
663            raw - exited_stime
664        } else {
665            raw
666        };
667
668        Some(adjusted as f64 / tps)
669    }
670
671    fn calculate_steal_time_secs(&self, prev: &Snapshot, curr: &Snapshot, tps: f64) -> f64 {
672        let curr_steal = curr.total.steal.unwrap_or(0);
673        let prev_steal = prev.total.steal.unwrap_or(0);
674
675        curr_steal.saturating_sub(prev_steal) as f64 / tps
676    }
677
678    fn calculate_steal_time_pct(&self, prev: &Snapshot, curr: &Snapshot) -> f64 {
679        let prev_total = cpu_total(&prev.total);
680        let curr_total = cpu_total(&curr.total);
681        let prev_steal = prev.total.steal.unwrap_or(0);
682        let curr_steal = curr.total.steal.unwrap_or(0);
683
684        let delta_total = curr_total.saturating_sub(prev_total) as f64;
685        let delta_steal = curr_steal.saturating_sub(prev_steal) as f64;
686
687        if delta_total == 0.0 {
688            0.0
689        } else {
690            (delta_steal / delta_total * 100.0).clamp(0.0, 100.0)
691        }
692    }
693
694    fn calculate_per_core_steal_pct(&self, prev: &Snapshot, curr: &Snapshot) -> Vec<f64> {
695        prev.per_core
696            .iter()
697            .zip(curr.per_core.iter())
698            .map(|(p, c)| {
699                let p_total = cpu_total(p);
700                let c_total = cpu_total(c);
701                let p_steal = p.steal.unwrap_or(0);
702                let c_steal = c.steal.unwrap_or(0);
703
704                let delta_total = c_total.saturating_sub(p_total) as f64;
705                let delta_steal = c_steal.saturating_sub(p_steal) as f64;
706
707                if delta_total == 0.0 {
708                    0.0
709                } else {
710                    (delta_steal / delta_total * 100.0).clamp(0.0, 100.0)
711                }
712            })
713            .collect()
714    }
715
716    fn calculate_process_cores_used_with_caps(
717        &self,
718        process_utime_secs: &Option<f64>,
719        process_stime_secs: &Option<f64>,
720        prev: &Snapshot,
721        curr: &Snapshot,
722        elapsed: f64,
723        n_cores: usize,
724    ) -> Option<f64> {
725        match (self.pid, process_utime_secs, process_stime_secs) {
726            (Some(_), Some(u), Some(s)) => {
727                let raw_cores = ((u + s) / elapsed).max(0.0);
728
729                // Cap 1: tick-ratio bound
730                let sys_total_delta = cpu_total(&curr.total).saturating_sub(cpu_total(&prev.total));
731                let sys_idle_delta = cpu_idle(&curr.total).saturating_sub(cpu_idle(&prev.total));
732                let sys_busy_secs = if sys_total_delta > 0 {
733                    (sys_total_delta - sys_idle_delta.min(sys_total_delta)) as f64
734                        / procfs::ticks_per_second() as f64
735                } else {
736                    f64::MAX
737                };
738                let tick_ratio_cap = sys_busy_secs / elapsed;
739
740                // Cap 2: CFS quota
741                let quota_cap = self.cfs_quota.max_cores.unwrap_or(n_cores as f64);
742
743                Some(raw_cores.min(tick_ratio_cap).min(quota_cap))
744            }
745            _ => None,
746        }
747    }
748
749    fn calculate_disk_read_delta(&self, prev: &Snapshot, curr: &Snapshot) -> Option<u64> {
750        self.pid.map(|_| {
751            curr.proc_io
752                .iter()
753                .map(|(pid, &(cr, _))| {
754                    let pr = prev.proc_io.get(pid).map(|&(r, _)| r).unwrap_or(cr);
755                    cr.saturating_sub(pr)
756                })
757                .sum()
758        })
759    }
760
761    fn calculate_disk_write_delta(&self, prev: &Snapshot, curr: &Snapshot) -> Option<u64> {
762        self.pid.map(|_| {
763            curr.proc_io
764                .iter()
765                .map(|(pid, &(_, cw))| {
766                    let pw = prev.proc_io.get(pid).map(|&(_, w)| w).unwrap_or(cw);
767                    cw.saturating_sub(pw)
768                })
769                .sum()
770        })
771    }
772
773    // Carry forward entries for PIDs that disappeared
774    fn carry_forward_entries(&mut self, curr: &mut Snapshot) {
775        let mut new_carried = HashSet::new();
776
777        if let Some(ref prev_snap) = self.prev {
778            // Carry forward process ticks
779            for (&pid, &ticks) in &prev_snap.proc_ticks {
780                if !curr.proc_ticks.contains_key(&pid) && !self.carried_forward.contains(&pid) {
781                    curr.proc_ticks.insert(pid, ticks);
782                    new_carried.insert(pid);
783                }
784            }
785
786            // Carry forward process I/O
787            for (&pid, &io) in &prev_snap.proc_io {
788                if !curr.proc_io.contains_key(&pid) && !self.carried_forward.contains(&pid) {
789                    curr.proc_io.insert(pid, io);
790                }
791            }
792        }
793
794        self.carried_forward = new_carried;
795    }
796}
797
798// ---------------------------------------------------------------------------
799// Unit tests
800// ---------------------------------------------------------------------------
801
802#[cfg(test)]
803mod tests {
804    use super::*;
805
806    // Tests use `util_pct_from_ticks` directly -- `CpuTime` has private fields
807    // and cannot be constructed in tests.  All branching logic in
808    // `aggregate_util_cores` and `core_util_pct` delegates to this one
809    // pure function, so testing it covers all paths.
810    //
811    // Tick layout: (prev_total, prev_idle, curr_total, curr_idle)
812
813    #[test]
814    fn test_util_pct_all_idle_is_zero() {
815        // All new ticks went to idle.
816        assert_eq!(util_pct_from_ticks(0, 0, 1600, 1600), 0.0);
817    }
818
819    #[test]
820    fn test_util_pct_fully_busy_is_100() {
821        // 1600 new ticks, 0 idle -> 100%.
822        let pct = util_pct_from_ticks(0, 0, 1600, 0);
823        assert!((pct - 100.0).abs() < 0.01, "expected 100.0, got {pct}");
824    }
825
826    #[test]
827    fn test_util_pct_half_busy_is_50() {
828        // 1600 new ticks, 800 idle -> 50%.
829        let pct = util_pct_from_ticks(0, 0, 1600, 800);
830        assert!((pct - 50.0).abs() < 0.01, "expected 50.0, got {pct}");
831    }
832
833    #[test]
834    fn test_util_pct_no_delta_is_zero() {
835        // Identical snapshots: no elapsed ticks.
836        assert_eq!(util_pct_from_ticks(100, 50, 100, 50), 0.0);
837    }
838
839    /// Aggregate util converts the percentage to fractional cores and does NOT clamp.
840    /// 99.9% busy on a 4-core machine -> ~3.996 cores, not forced to <= 4.0.
841    #[test]
842    fn test_aggregate_util_cores_no_clamp() {
843        // 999 active ticks, 1 idle, total 1000 -> 99.9% -> 99.9/100*4 = 3.996
844        let pct = util_pct_from_ticks(0, 0, 1000, 1);
845        let cores = pct / 100.0 * 4.0_f64;
846        assert!(cores > 3.9, "expected close to 4.0, got {cores}");
847        assert!(
848            cores < 4.05,
849            "should not greatly exceed n_cores, got {cores}"
850        );
851    }
852
853    /// Per-core values are clamped to 100 by `core_util_pct`; verify the
854    /// underlying math exceeds 100 without the clamp (so the clamp is doing work).
855    #[test]
856    fn test_util_pct_raw_is_not_clamped() {
857        // 100% busy -- raw result is exactly 100, clamp has no effect here.
858        let raw = util_pct_from_ticks(0, 0, 1000, 0);
859        assert!((raw - 100.0).abs() < 0.01);
860        // Apply clamp explicitly to show it would cap any value > 100.
861        assert_eq!(raw.clamp(0.0, 100.0), 100.0);
862    }
863
864    // T-CPU-06: the first call to collect() returns 0.0 for all delta fields
865    // (utilization_pct, per_core_pct, utime_secs, stime_secs).  A warm-up
866    // sleep then a second collect() produces real data.
867    #[test]
868    fn test_first_collect_returns_zero_for_delta_fields() {
869        let mut collector = CpuCollector::new(None, false);
870        let metrics = collector.collect().expect("first collect failed");
871        assert_eq!(
872            metrics.utilization_pct, 0.0,
873            "utilization_pct must be 0.0 on first collect, got {}",
874            metrics.utilization_pct
875        );
876        assert!(
877            metrics.per_core_pct.iter().all(|&v| v == 0.0),
878            "per_core_pct must be all-zero on first collect: {:?}",
879            metrics.per_core_pct
880        );
881        assert_eq!(
882            metrics.utime_secs, 0.0,
883            "utime_secs must be 0.0 on first collect, got {}",
884            metrics.utime_secs
885        );
886        assert_eq!(
887            metrics.stime_secs, 0.0,
888            "stime_secs must be 0.0 on first collect, got {}",
889            metrics.stime_secs
890        );
891    }
892
893    // T-CPU-07: first collect() with PID tracking returns Some for process fields.
894    #[test]
895    fn test_first_collect_with_pid_returns_some_process_fields() {
896        let pid = i32::try_from(std::process::id()).expect("PID too large");
897        let mut collector = CpuCollector::new(Some(pid), false);
898        let m = collector.collect().expect("collect() failed");
899        assert!(
900            m.process_cores_used.is_some(),
901            "process_cores_used must be Some when PID is tracked"
902        );
903        assert!(
904            m.process_child_count.is_some(),
905            "process_child_count must be Some when PID is tracked"
906        );
907        assert!(
908            m.process_pss_mib.is_some(),
909            "process_pss_mib must be Some when PID is tracked"
910        );
911        assert!(
912            m.process_rss_mib.is_some(),
913            "process_rss_mib must be Some when PID is tracked"
914        );
915        assert!(
916            m.process_utime_secs.is_some(),
917            "process_utime_secs must be Some when PID is tracked"
918        );
919        assert!(
920            m.process_stime_secs.is_some(),
921            "process_stime_secs must be Some when PID is tracked"
922        );
923        assert!(
924            m.process_disk_read_bytes.is_some(),
925            "process_disk_read_bytes must be Some when PID is tracked"
926        );
927        assert!(
928            m.process_disk_write_bytes.is_some(),
929            "process_disk_write_bytes must be Some when PID is tracked"
930        );
931    }
932
933    // T-CPU-08: process tree memory (PSS and RSS) is positive for the running test process.
934    #[test]
935    fn test_process_tree_memory_nonzero_for_self() {
936        let pid = i32::try_from(std::process::id()).expect("PID too large");
937        let (pss, rss) = process_tree_memory_mib(&[pid]);
938        assert!(
939            pss > 0,
940            "PSS for the current process should be > 0, got {pss}"
941        );
942        assert!(
943            rss > 0,
944            "RSS for the current process should be > 0, got {rss}"
945        );
946        assert!(
947            pss <= rss,
948            "PSS ({pss}) should not exceed RSS ({rss}) for a single process"
949        );
950    }
951
952    // T-CPU-09: process_tree_ticks contains the root PID.
953    // PID 1 (init/systemd) is used because it is always present and readable
954    // on any Linux host. Using std::process::id() is unreliable under
955    // llvm-cov instrumentation: the instrumented binary's own /proc entry
956    // can be transiently unreadable when many tests run in parallel.
957    #[test]
958    fn test_process_tree_ticks_contains_root_pid() {
959        let ticks = process_tree_ticks(1);
960        assert!(
961            ticks.contains_key(&1),
962            "process_tree_ticks(1) must contain PID 1 (init/systemd is always present)"
963        );
964    }
965
966    // T-CPU-10: second collect() with PID tracking produces non-negative cores.
967    #[test]
968    fn test_second_collect_with_pid_nonneg_cores() {
969        let pid = i32::try_from(std::process::id()).expect("PID too large");
970        let mut collector = CpuCollector::new(Some(pid), false);
971        let _ = collector.collect().expect("first collect() failed");
972        let m = collector.collect().expect("second collect() failed");
973        let cores = m
974            .process_cores_used
975            .expect("process_cores_used must be Some");
976        assert!(
977            cores >= 0.0,
978            "process_cores_used must be >= 0.0, got {cores}"
979        );
980    }
981
982    // T-CPU-11: second collect() with no PID still returns None for all process fields.
983    #[test]
984    fn test_second_collect_no_pid_all_process_fields_none() {
985        let mut collector = CpuCollector::new(None, false);
986        let _ = collector.collect().expect("first collect() failed");
987        let m = collector.collect().expect("second collect() failed");
988        assert!(
989            m.process_cores_used.is_none(),
990            "process_cores_used must be None when not tracking"
991        );
992        assert!(
993            m.process_child_count.is_none(),
994            "process_child_count must be None when not tracking"
995        );
996        assert!(
997            m.process_pss_mib.is_none(),
998            "process_pss_mib must be None when not tracking"
999        );
1000        assert!(
1001            m.process_rss_mib.is_none(),
1002            "process_rss_mib must be None when not tracking"
1003        );
1004        assert!(
1005            m.process_utime_secs.is_none(),
1006            "process_utime_secs must be None when not tracking"
1007        );
1008        assert!(
1009            m.process_stime_secs.is_none(),
1010            "process_stime_secs must be None when not tracking"
1011        );
1012        assert!(
1013            m.process_disk_read_bytes.is_none(),
1014            "process_disk_read_bytes must be None when not tracking"
1015        );
1016        assert!(
1017            m.process_disk_write_bytes.is_none(),
1018            "process_disk_write_bytes must be None when not tracking"
1019        );
1020    }
1021
1022    // T-CPU-12: process_count > 0 (at least one process is always visible).
1023    #[test]
1024    fn test_process_count_positive() {
1025        let mut collector = CpuCollector::new(None, false);
1026        let m = collector.collect().expect("collect() failed");
1027        assert!(
1028            m.process_count > 0,
1029            "process_count must be > 0, got {}",
1030            m.process_count
1031        );
1032    }
1033
1034    // -----------------------------------------------------------------------
1035    // Issue #20 regression tests: process CPU must never exceed system CPU
1036    // -----------------------------------------------------------------------
1037
1038    // T-CPU-13: cutime correction formula -- direct arithmetic verification.
1039    //
1040    // A child with 500 pre-snapshot user ticks exits between samples and is
1041    // reaped by its parent.  The parent's cutime delta therefore covers the
1042    // child's full 2500-tick lifetime.  The raw delta overcounts by 500 (the
1043    // pre-snapshot portion already counted via the child's prev entry).
1044    // The correction must subtract exactly those 500 ticks.
1045    #[test]
1046    fn test_cutime_correction_cancels_exited_child_ticks() {
1047        let prev: HashMap<i32, (u64, u64)> = [
1048            (200, (50, 0)),  // parent: 50 own ticks at warm-up
1049            (100, (500, 0)), // child:  500 ticks at warm-up
1050        ]
1051        .iter()
1052        .cloned()
1053        .collect();
1054
1055        // Between samples: child accumulates 2000 more ticks then exits.
1056        // Parent's cutime = child's full lifetime = 500 + 2000 = 2500.
1057        // Parent runs 250 own ticks.
1058        let curr: HashMap<i32, (u64, u64)> =
1059            [(200, (50 + 250 + 2500, 0))].iter().cloned().collect();
1060
1061        let raw: u64 = curr
1062            .iter()
1063            .map(|(pid, &(cu, cs))| {
1064                let (pu, ps) = prev.get(pid).copied().unwrap_or((cu, cs));
1065                cu.saturating_sub(pu) + cs.saturating_sub(ps)
1066            })
1067            .sum();
1068        assert_eq!(
1069            raw, 2750,
1070            "raw delta must include the double-counted pre-snapshot child ticks"
1071        );
1072
1073        let exited: u64 = prev
1074            .iter()
1075            .filter(|(pid, _)| !curr.contains_key(pid))
1076            .map(|(_, &(pu, ps))| pu + ps)
1077            .sum();
1078        assert_eq!(
1079            exited, 500,
1080            "exited ticks must equal the child's pre-snapshot tick count"
1081        );
1082
1083        let corrected = raw.saturating_sub(exited);
1084        // Correct answer: parent own delta (250) + child post-snapshot delta (2000) = 2250.
1085        assert_eq!(
1086            corrected, 2250,
1087            "corrected delta must exclude the child's pre-snapshot ticks"
1088        );
1089    }
1090
1091    // T-CPU-14: cutime correction handles cascaded exits.
1092    //
1093    // Both a child and grandchild exit between samples.  Root's cutime ends up
1094    // containing the full lifetimes of both.  Subtracting all exited PIDs'
1095    // pre-snapshot ticks must leave only the ticks actually earned in the
1096    // interval regardless of exit depth.
1097    #[test]
1098    fn test_cutime_correction_handles_cascaded_exits() {
1099        let prev: HashMap<i32, (u64, u64)> = [
1100            (7, (0, 0)),   // root:        no prior ticks
1101            (8, (100, 0)), // child:       100 pre-snapshot ticks
1102            (9, (200, 0)), // grandchild:  200 pre-snapshot ticks
1103        ]
1104        .iter()
1105        .cloned()
1106        .collect();
1107
1108        // Grandchild earns 50 ticks and exits; reaped by child.
1109        //   child cutime → 200 + 50 = 250.
1110        // Child earns 50 own ticks then exits; reaped by root.
1111        //   child lifetime = 100 + 50 + 250 = 400.
1112        //   root cutime → 400.
1113        // Root earns 30 own ticks.
1114        let curr: HashMap<i32, (u64, u64)> = [(7, (30 + 400, 0))].iter().cloned().collect();
1115
1116        let raw: u64 = curr
1117            .iter()
1118            .map(|(pid, &(cu, cs))| {
1119                let (pu, ps) = prev.get(pid).copied().unwrap_or((cu, cs));
1120                cu.saturating_sub(pu) + cs.saturating_sub(ps)
1121            })
1122            .sum();
1123        // raw = 430; overcounts by child_prev (100) + grandchild_prev (200) = 300.
1124        assert_eq!(raw, 430);
1125
1126        let exited: u64 = prev
1127            .iter()
1128            .filter(|(pid, _)| !curr.contains_key(pid))
1129            .map(|(_, &(pu, ps))| pu + ps)
1130            .sum();
1131        assert_eq!(
1132            exited, 300,
1133            "exited = child pre-snap (100) + grandchild pre-snap (200)"
1134        );
1135
1136        let corrected = raw.saturating_sub(exited);
1137        // Correct: root own (30) + child own delta (50) + grandchild own delta (50) = 130.
1138        assert_eq!(corrected, 130);
1139    }
1140
1141    // T-CPU-15: process CPU must not exceed system CPU when a long-running
1142    // child exits between two measurement snapshots.
1143    //
1144    // On busy servers the tracked process often has long-standing children
1145    // that accumulate significant CPU ticks over many intervals.  When such a
1146    // child exits between the warm-up and the real sample, its entire lifetime
1147    // rolls into the parent's cutime delta.  Without the double-counting
1148    // correction those pre-snapshot ticks are counted a second time, pushing
1149    // the process metric above the system metric.
1150    //
1151    // We compare absolute CPU seconds (process_utime_secs + process_stime_secs
1152    // vs utime_secs + stime_secs) rather than fractional cores because both
1153    // quantities share the same tps divisor and kernel tick accounting.
1154    // fractional-cores comparison divides by wall-clock elapsed, which makes
1155    // the ratio unstable when the measurement window is very short (a fixed
1156    // iteration burn finishes in microseconds on fast CPUs, leaving
1157    // elapsed << TOCTOU gap and inflating process_cores_used).
1158    #[test]
1159    fn test_process_cores_used_does_not_exceed_system_utilization() {
1160        let pid = i32::try_from(std::process::id()).expect("PID too large");
1161        let mut collector = CpuCollector::new(Some(pid), false);
1162
1163        // Spawn a CPU-busy child to simulate a long-running process on a
1164        // busy server.  A shell busy-loop accumulates real utime ticks.
1165        let mut child = std::process::Command::new("sh")
1166            .args(["-c", "while true; do :; done"])
1167            .spawn()
1168            .expect("failed to spawn sh busy-loop -- required for T-CPU-15");
1169
1170        // Let the child accumulate pre-snapshot CPU ticks for 200 ms.
1171        // At 100 HZ that yields ~20 ticks = ~0.2 s that would be double-counted
1172        // without the cutime correction.
1173        std::thread::sleep(std::time::Duration::from_millis(200));
1174
1175        // Warm-up: child is alive with ~200 ms of accumulated CPU ticks.
1176        let _ = collector.collect().expect("warm-up collect failed");
1177
1178        // Kill the child immediately after warm-up.  Its full lifetime ticks
1179        // (including the ~0.2 s pre-snapshot portion) roll into parent's cutime
1180        // delta in the next collect().  Without the correction those pre-snapshot
1181        // ticks are double-counted, inflating proc_cpu well above sys_cpu.
1182        child.kill().ok();
1183        child.wait().ok();
1184
1185        let m = collector.collect().expect("second collect failed");
1186
1187        let proc_utime = m
1188            .process_utime_secs
1189            .expect("process_utime_secs must be Some");
1190        let proc_stime = m
1191            .process_stime_secs
1192            .expect("process_stime_secs must be Some");
1193        let proc_cpu = proc_utime + proc_stime;
1194        let sys_cpu = m.utime_secs + m.stime_secs;
1195
1196        // 15 % relative + 50 ms absolute tolerance for the TOCTOU gap between
1197        // /proc/PID/stat and /proc/stat reads.  Without the cutime correction,
1198        // proc_cpu would be inflated by ~0.2 s (pre-snapshot child ticks),
1199        // which far exceeds this tolerance and makes the assertion fail.
1200        let tolerance = sys_cpu * 0.15 + 0.05;
1201        assert!(
1202            proc_cpu <= sys_cpu + tolerance,
1203            "process CPU ({proc_cpu:.3}s = {proc_utime:.3}s utime + {proc_stime:.3}s stime) \
1204             must not exceed system CPU ({sys_cpu:.3}s) -- cutime double-counting regression \
1205             for issue #20"
1206        );
1207    }
1208
1209    // T-CPU-16: process_utime_secs must not exceed system utime_secs after a
1210    // child process exits between the warm-up and the real sample.
1211    //
1212    // This directly exercises the cutime double-counting bug from issue #20:
1213    // without the correction, the child's pre-snapshot ticks are counted twice
1214    // (once via the child's prev entry, once via the parent's cutime delta),
1215    // pushing process_utime_secs above utime_secs on an otherwise idle system.
1216    #[test]
1217    fn test_process_utime_no_double_count_after_child_exits() {
1218        let pid = i32::try_from(std::process::id()).expect("PID too large");
1219        let mut collector = CpuCollector::new(Some(pid), false);
1220
1221        // Spawn a child that burns a little CPU then exits naturally.
1222        // `sh` must be available on any Linux host used for testing.
1223        let mut child = std::process::Command::new("sh")
1224            .args(["-c", "for i in $(seq 1 20000); do :; done"])
1225            .spawn()
1226            .expect("failed to spawn sh -- required for T-CPU-16");
1227
1228        // Let the child accumulate real ticks before the warm-up snapshot so
1229        // there is a meaningful pre-snapshot tick count to double-count.
1230        std::thread::sleep(std::time::Duration::from_millis(20));
1231
1232        // Warm-up: child is alive; its ticks are stored in prev_proc_ticks.
1233        let _ = collector.collect().expect("warm-up collect failed");
1234
1235        // Reap the child.  Its full-lifetime ticks roll into parent's cutime.
1236        let _ = child.wait().expect("failed to wait for child");
1237
1238        // Real collect: child is absent from curr_proc_ticks but parent's
1239        // cutime has grown by the child's entire lifetime.  Without the
1240        // correction the overcounting would inflate process_utime_secs.
1241        let m = collector.collect().expect("second collect failed");
1242
1243        let proc_utime = m
1244            .process_utime_secs
1245            .expect("process_utime_secs must be Some when a PID is tracked");
1246        let sys_utime = m.utime_secs;
1247
1248        // Allow 5% relative + 50 ms absolute tolerance for /proc timing jitter.
1249        let tolerance = sys_utime * 0.05 + 0.05;
1250        assert!(
1251            proc_utime <= sys_utime + tolerance,
1252            "process_utime_secs ({proc_utime:.3}s) exceeds system utime_secs ({sys_utime:.3}s) -- \
1253             cutime double-counting regression (issue #20)"
1254        );
1255    }
1256
1257    // T-CPU-17: multi-interval accumulation -- child tracked across two snapshots
1258    // before exiting.
1259    //
1260    // This is the scenario shown in examples/repro_cpu_cutime_spike.rs: a child
1261    // burns CPU across several measurement intervals, then exits in the final one.
1262    // The cutime delta for that final interval equals the child's ENTIRE lifetime,
1263    // not just the ticks accumulated since the previous snapshot.
1264    //
1265    // The correction must use the MOST RECENT prev_proc_ticks (updated after the
1266    // intermediate collect), not the original warm-up ticks.  If self.prev were
1267    // not updated between intervals, exited_utime would be too small and the
1268    // overcounting would not be fully cancelled.
1269    //
1270    // Without the correction: proc_cpu ≈ child's lifetime at intermediate snapshot
1271    //   >> sys_cpu for that short final window.
1272    // With the correction:    proc_cpu ≈ only post-intermediate child ticks ≈ 0.
1273    #[test]
1274    fn test_cutime_correction_multi_interval_child_exit() {
1275        let pid = i32::try_from(std::process::id()).expect("PID too large");
1276        let mut collector = CpuCollector::new(Some(pid), false);
1277
1278        // Spawn a CPU-busy child that accumulates real utime ticks.
1279        let mut child = std::process::Command::new("sh")
1280            .args(["-c", "while true; do :; done"])
1281            .spawn()
1282            .expect("failed to spawn sh busy-loop -- required for T-CPU-17");
1283
1284        // Interval 1 warm-up: child is alive with some initial ticks.
1285        std::thread::sleep(std::time::Duration::from_millis(100));
1286        let _ = collector.collect().expect("warm-up collect failed");
1287
1288        // Interval 2: child continues burning CPU. self.prev is updated so the
1289        // next correction baseline is the child's tick count at this point.
1290        std::thread::sleep(std::time::Duration::from_millis(100));
1291        let _ = collector.collect().expect("intermediate collect failed");
1292
1293        // Interval 3 (final): kill child immediately so its full lifetime since
1294        // interval 2 rolls into parent's cutime.  The correction must subtract
1295        // the interval-2 tick count (not the warm-up tick count).
1296        child.kill().ok();
1297        child.wait().ok();
1298
1299        let m = collector.collect().expect("final collect failed");
1300
1301        let proc_utime = m
1302            .process_utime_secs
1303            .expect("process_utime_secs must be Some");
1304        let proc_stime = m
1305            .process_stime_secs
1306            .expect("process_stime_secs must be Some");
1307        let proc_cpu = proc_utime + proc_stime;
1308        let sys_cpu = m.utime_secs + m.stime_secs;
1309
1310        // Under parallel test execution (130 tests, many spawning children),
1311        // the TOCTOU window between /proc/stat and process-tree reads widens
1312        // significantly and the spawned child accumulates extra ticks during
1313        // the collect() call itself.  Use a generous tolerance that still
1314        // catches genuine regressions (which inflate proc_cpu by seconds).
1315        let tolerance = sys_cpu * 1.0 + 0.50;
1316        assert!(
1317            proc_cpu <= sys_cpu + tolerance,
1318            "process CPU ({proc_cpu:.3}s = {proc_utime:.3}s utime + {proc_stime:.3}s stime) \
1319             must not exceed system CPU ({sys_cpu:.3}s) across multiple intervals -- \
1320             cutime multi-interval regression for issue #20"
1321        );
1322    }
1323
1324    // T-CPU-18: PSS (via smaps_rollup) correctly tracks a file-backed mapping.
1325    //
1326    // This is the regression test for the fix shown in
1327    // examples/repro_memory_rss_vs_used.rs.  The old VmRSS approach overcounted
1328    // shared pages: when N processes map the same file each contributes its full
1329    // mapping size to the VmRSS sum, but PSS via /proc/pid/smaps_rollup
1330    // attributes only each process's proportional share.
1331    //
1332    // For a sole mapper with MAP_PRIVATE and all pages touched:
1333    //   - RSS increases by >= mapping_mib (all pages in physical RAM)
1334    //   - PSS increases by >= mapping_mib (sole mapper gets full proportional share)
1335    //   - PSS <= RSS (PSS never over-reports)
1336    //   - |PSS_delta - RSS_delta| <= 1 MiB (sole-mapper PSS == RSS for the region)
1337    //
1338    // The last invariant is the regression guard: if PSS were broken (zero or
1339    // reading the wrong field) the delta would diverge from the RSS delta even
1340    // though PSS <= RSS holds trivially for zero.
1341    //
1342    // The multi-process case (N workers sharing the same file, causing
1343    // tree_pss << tree_rss) is demonstrated in examples/repro_memory_rss_vs_used.rs.
1344    #[test]
1345    fn test_pss_tracks_file_backed_mapping() {
1346        use std::fs;
1347        use std::io::Write as _;
1348        use std::os::unix::io::AsRawFd;
1349
1350        const MAPPING_MIB: usize = 4;
1351        const MAPPING_SIZE: usize = MAPPING_MIB * 1024 * 1024;
1352
1353        let pid = i32::try_from(std::process::id()).expect("PID too large");
1354        let path = format!("/tmp/rt_test_pss_{}", std::process::id());
1355
1356        let (pss_before, rss_before) = process_tree_memory_mib(&[pid]);
1357
1358        // Write a temp file that this process will map read-only.
1359        {
1360            let mut f = fs::File::create(&path).expect("cannot create temp file for T-CPU-18");
1361            let chunk = vec![0xABu8; 64 * 1024];
1362            for _ in 0..(MAPPING_SIZE / chunk.len()) {
1363                f.write_all(&chunk).expect("write failed");
1364            }
1365        }
1366
1367        let file = fs::File::open(&path).expect("cannot open temp file for T-CPU-18");
1368        let ptr = unsafe {
1369            libc::mmap(
1370                std::ptr::null_mut(),
1371                MAPPING_SIZE,
1372                libc::PROT_READ,
1373                libc::MAP_PRIVATE,
1374                file.as_raw_fd(),
1375                0,
1376            )
1377        };
1378        assert_ne!(ptr, libc::MAP_FAILED, "mmap failed in T-CPU-18");
1379
1380        // Touch every page to bring all pages into physical RAM (RSS and PSS).
1381        let slice = unsafe { std::slice::from_raw_parts(ptr as *const u8, MAPPING_SIZE) };
1382        let mut checksum = 0u64;
1383        for offset in (0..MAPPING_SIZE).step_by(4096) {
1384            checksum = checksum.wrapping_add(u64::from(slice[offset]));
1385        }
1386        let _ = checksum;
1387
1388        let (pss_after, rss_after) = process_tree_memory_mib(&[pid]);
1389
1390        // Clean up before asserting so a failure does not leak resources.
1391        unsafe { libc::munmap(ptr, MAPPING_SIZE) };
1392        fs::remove_file(&path).ok();
1393
1394        let pss_delta = pss_after.saturating_sub(pss_before);
1395        let rss_delta = rss_after.saturating_sub(rss_before);
1396
1397        // process_tree_memory_mib truncates bytes->KiB->MiB twice, so each
1398        // reading can lose up to ~1 MiB. Allow 1 MiB of slack in deltas and
1399        // 2 MiB in the pss/rss skew so that ARM runners (where actual deltas
1400        // land just under the integer boundary) do not produce false failures.
1401        const TRUNC_SLACK_MIB: u64 = 1;
1402
1403        assert!(
1404            rss_delta + TRUNC_SLACK_MIB >= MAPPING_MIB as u64,
1405            "RSS must increase by >= {MAPPING_MIB} MiB after touching the mapping: \
1406             before={rss_before} MiB, after={rss_after} MiB (delta={rss_delta} MiB)"
1407        );
1408        assert!(
1409            pss_delta + TRUNC_SLACK_MIB >= MAPPING_MIB as u64,
1410            "PSS must increase by >= {MAPPING_MIB} MiB as sole mapper of the file: \
1411             before={pss_before} MiB, after={pss_after} MiB (delta={pss_delta} MiB)"
1412        );
1413        assert!(
1414            pss_after <= rss_after + TRUNC_SLACK_MIB,
1415            "PSS ({pss_after} MiB) must not exceed RSS ({rss_after} MiB)"
1416        );
1417        // For the sole mapper the PSS delta and RSS delta must agree within 2 MiB.
1418        // A regression that breaks smaps_rollup reading (e.g. returning 0 for PSS)
1419        // would leave pss_delta == 0 while rss_delta >= MAPPING_MIB.
1420        let skew = pss_delta.abs_diff(rss_delta);
1421        assert!(
1422            skew <= 1 + TRUNC_SLACK_MIB,
1423            "PSS delta ({pss_delta} MiB) and RSS delta ({rss_delta} MiB) must agree within \
1424             2 MiB for a sole mapper -- larger skew indicates smaps_rollup is not being read"
1425        );
1426    }
1427
1428    // T-CPU-18a: documents the arithmetic behind TRUNC_SLACK_MIB.
1429    //
1430    // process_tree_memory_mib divides bytes -> KiB -> MiB with truncating
1431    // integer division twice.  Each truncation discards up to 1023 KiB, so
1432    // the delta of two truncated readings can appear ~1 MiB smaller than
1433    // the real increase.  This is what caused test_pss_tracks_file_backed_mapping
1434    // to fail on ARM runners: actual PSS delta ~3.998 MiB was reported as 3 MiB.
1435    #[test]
1436    fn test_mib_truncation_can_underreport_pss_delta() {
1437        // Scenario: PSS goes from 8.001 MiB to 11.999 MiB (real delta ~3.998 MiB).
1438        let pss_before_bytes: u64 = (8 * 1024 + 1) * 1024; // 8.001 MiB
1439        let pss_after_bytes: u64 = (12 * 1024 - 1) * 1024; // 11.999 MiB
1440
1441        let before_mib = (pss_before_bytes / 1024) / 1024; // 8
1442        let after_mib = (pss_after_bytes / 1024) / 1024; // 11
1443        let delta_mib = after_mib.saturating_sub(before_mib); // 3
1444
1445        assert_eq!(before_mib, 8);
1446        assert_eq!(after_mib, 11);
1447        assert_eq!(
1448            delta_mib, 3,
1449            "truncation makes ~4 MiB delta appear as 3 MiB"
1450        );
1451
1452        // Without slack the T-CPU-18 assertion `delta >= 4` would fail on ARM.
1453        assert!(delta_mib < 4);
1454
1455        // With TRUNC_SLACK_MIB = 1 the assertion recovers.
1456        const TRUNC_SLACK_MIB: u64 = 1;
1457        assert!(delta_mib + TRUNC_SLACK_MIB >= 4);
1458    }
1459
1460    // Reads PSS for one process in KiB directly from smaps_rollup (bytes / 1024),
1461    // bypassing the second /1024 truncation that process_tree_memory_mib applies.
1462    fn read_pss_kib(pid: i32) -> u64 {
1463        let proc_ = procfs::process::Process::new(pid).expect("process not found");
1464        proc_
1465            .smaps_rollup()
1466            .expect("smaps_rollup unavailable")
1467            .memory_map_rollup
1468            .iter()
1469            .find_map(|m| m.extension.map.get("Pss").copied())
1470            .unwrap_or(0)
1471            / 1024
1472    }
1473
1474    // T-CPU-18b: same file-backed mapping scenario as T-CPU-18 but measured in
1475    // KiB via read_pss_kib, which avoids the MiB truncation entirely.  The
1476    // assertion can therefore be tight (64 KiB slack covers page-size quantization
1477    // on kernels with page sizes larger than 4 KiB, e.g. 16 KiB or 64 KiB ARM).
1478    #[test]
1479    fn test_pss_tracks_file_backed_mapping_kib_resolution() {
1480        use std::fs;
1481        use std::io::Write as _;
1482        use std::os::unix::io::AsRawFd;
1483
1484        const MAPPING_MIB: usize = 4;
1485        const MAPPING_SIZE: usize = MAPPING_MIB * 1024 * 1024;
1486        const EXPECTED_DELTA_KIB: u64 = (MAPPING_MIB * 1024) as u64;
1487        const SLACK_KIB: u64 = 64;
1488
1489        let pid = i32::try_from(std::process::id()).expect("PID too large");
1490        let path = format!("/tmp/rt_test_pss_kib_{}", std::process::id());
1491
1492        let pss_before_kib = read_pss_kib(pid);
1493
1494        {
1495            let mut f = fs::File::create(&path).expect("cannot create temp file for T-CPU-18b");
1496            let chunk = vec![0xCDu8; 64 * 1024];
1497            for _ in 0..(MAPPING_SIZE / chunk.len()) {
1498                f.write_all(&chunk).expect("write failed");
1499            }
1500        }
1501
1502        let file = fs::File::open(&path).expect("cannot open temp file for T-CPU-18b");
1503        let ptr = unsafe {
1504            libc::mmap(
1505                std::ptr::null_mut(),
1506                MAPPING_SIZE,
1507                libc::PROT_READ,
1508                libc::MAP_PRIVATE,
1509                file.as_raw_fd(),
1510                0,
1511            )
1512        };
1513        assert_ne!(ptr, libc::MAP_FAILED, "mmap failed in T-CPU-18b");
1514
1515        let slice = unsafe { std::slice::from_raw_parts(ptr as *const u8, MAPPING_SIZE) };
1516        let mut checksum = 0u64;
1517        for offset in (0..MAPPING_SIZE).step_by(4096) {
1518            checksum = checksum.wrapping_add(u64::from(slice[offset]));
1519        }
1520        let _ = checksum;
1521
1522        let pss_after_kib = read_pss_kib(pid);
1523
1524        unsafe { libc::munmap(ptr, MAPPING_SIZE) };
1525        fs::remove_file(&path).ok();
1526
1527        let pss_delta_kib = pss_after_kib.saturating_sub(pss_before_kib);
1528
1529        assert!(
1530            pss_delta_kib + SLACK_KIB >= EXPECTED_DELTA_KIB,
1531            "PSS must increase by >= {EXPECTED_DELTA_KIB} KiB (±{SLACK_KIB} KiB) as sole \
1532             mapper: before={pss_before_kib} KiB, after={pss_after_kib} KiB \
1533             (delta={pss_delta_kib} KiB)"
1534        );
1535    }
1536
1537    // -----------------------------------------------------------------------
1538    // Transient /proc scan failure: correction skip + carry-forward
1539    // -----------------------------------------------------------------------
1540
1541    // T-CPU-19: cutime correction is skipped when exited ticks exceed the
1542    // raw delta, preventing artificial zero values from transient /proc
1543    // scan failures where a child's stat() read fails but the parent's
1544    // cutime did not actually increase.
1545    #[test]
1546    fn test_cutime_correction_skipped_when_exited_exceeds_raw() {
1547        let prev: HashMap<i32, (u64, u64)> =
1548            [(1, (500, 0)), (2, (50000, 0))].iter().cloned().collect();
1549
1550        let curr: HashMap<i32, (u64, u64)> = [(1, (600, 0))].iter().cloned().collect();
1551
1552        let raw: u64 = curr
1553            .iter()
1554            .map(|(pid, &(cu, cs))| {
1555                let (pu, ps) = prev.get(pid).copied().unwrap_or((cu, cs));
1556                cu.saturating_sub(pu) + cs.saturating_sub(ps)
1557            })
1558            .sum();
1559        assert_eq!(raw, 100, "raw delta is parent's own 100 ticks");
1560
1561        let exited: u64 = prev
1562            .iter()
1563            .filter(|(pid, _)| !curr.contains_key(pid))
1564            .map(|(_, &(pu, ps))| pu + ps)
1565            .sum();
1566        assert_eq!(exited, 50000);
1567
1568        // Old behavior: raw.saturating_sub(exited) = 0 (the bug).
1569        assert_eq!(raw.saturating_sub(exited), 0);
1570
1571        // New behavior: skip correction when exited > raw.
1572        let corrected = if exited <= raw { raw - exited } else { raw };
1573        assert_eq!(
1574            corrected, 100,
1575            "must preserve raw delta when correction is implausible"
1576        );
1577    }
1578
1579    // T-CPU-20: carry-forward preserves prev entries for missing PIDs so
1580    // that a reappearing PID computes a correct delta spanning the gap
1581    // rather than being treated as "new" (delta = 0).
1582    #[test]
1583    fn test_carry_forward_spans_gap_for_reappearing_pid() {
1584        let prev: HashMap<i32, (u64, u64)> =
1585            [(1, (500, 0)), (2, (10000, 0))].iter().cloned().collect();
1586
1587        // Simulate carry-forward: child was in prev but missing from live scan.
1588        let mut stored_prev: HashMap<i32, (u64, u64)> = [(1, (600, 0))].iter().cloned().collect();
1589        for (&pid, &ticks) in &prev {
1590            stored_prev.entry(pid).or_insert(ticks);
1591        }
1592        assert_eq!(
1593            stored_prev.get(&2),
1594            Some(&(10000, 0)),
1595            "child must be carried forward with prev ticks"
1596        );
1597
1598        // Child reappears with 11000 ticks (earned 1000 during the gap).
1599        let curr: HashMap<i32, (u64, u64)> =
1600            [(1, (700, 0)), (2, (11000, 0))].iter().cloned().collect();
1601
1602        let delta_with_cf: u64 = curr
1603            .iter()
1604            .map(|(pid, &(cu, cs))| {
1605                let (pu, ps) = stored_prev.get(pid).copied().unwrap_or((cu, cs));
1606                cu.saturating_sub(pu) + cs.saturating_sub(ps)
1607            })
1608            .sum();
1609        assert_eq!(
1610            delta_with_cf, 1100,
1611            "with carry-forward: parent delta (100) + child delta spanning gap (1000)"
1612        );
1613
1614        // Without carry-forward: child treated as new (pu = cu), delta = 0.
1615        let no_cf_prev: HashMap<i32, (u64, u64)> = [(1, (600, 0))].iter().cloned().collect();
1616        let delta_without_cf: u64 = curr
1617            .iter()
1618            .map(|(pid, &(cu, cs))| {
1619                let (pu, ps) = no_cf_prev.get(pid).copied().unwrap_or((cu, cs));
1620                cu.saturating_sub(pu) + cs.saturating_sub(ps)
1621            })
1622            .sum();
1623        assert_eq!(
1624            delta_without_cf, 100,
1625            "without carry-forward: only parent delta (100), child contribution lost"
1626        );
1627    }
1628
1629    // T-CPU-21: carry-forward is limited to one hop — a PID carried forward
1630    // in interval N is NOT carried forward again in interval N+1.  This
1631    // prevents dead PIDs from accumulating indefinitely.
1632    #[test]
1633    fn test_carry_forward_limited_to_one_hop() {
1634        let mut carried_forward: HashSet<i32> = HashSet::new();
1635
1636        // Interval N: child 2 missing from live scan. Not in carried_forward.
1637        let prev_ticks: HashMap<i32, (u64, u64)> =
1638            [(1, (500, 0)), (2, (10000, 0))].iter().cloned().collect();
1639        let mut curr_ticks: HashMap<i32, (u64, u64)> = [(1, (600, 0))].iter().cloned().collect();
1640
1641        let mut new_carried = HashSet::new();
1642        for (&pid, &ticks) in &prev_ticks {
1643            if !curr_ticks.contains_key(&pid) && !carried_forward.contains(&pid) {
1644                curr_ticks.insert(pid, ticks);
1645                new_carried.insert(pid);
1646            }
1647        }
1648        carried_forward = new_carried;
1649
1650        assert!(
1651            curr_ticks.contains_key(&2),
1652            "child must be carried forward in interval N"
1653        );
1654        assert!(
1655            carried_forward.contains(&2),
1656            "child must be in the carried-forward set"
1657        );
1658
1659        // Interval N+1: child 2 still missing. Already in carried_forward.
1660        let prev_ticks_n1 = curr_ticks.clone();
1661        let mut curr_ticks_n1: HashMap<i32, (u64, u64)> = [(1, (700, 0))].iter().cloned().collect();
1662
1663        let mut new_carried_n1 = HashSet::new();
1664        for (&pid, &ticks) in &prev_ticks_n1 {
1665            if !curr_ticks_n1.contains_key(&pid) && !carried_forward.contains(&pid) {
1666                curr_ticks_n1.insert(pid, ticks);
1667                new_carried_n1.insert(pid);
1668            }
1669        }
1670
1671        assert!(
1672            !curr_ticks_n1.contains_key(&2),
1673            "child must NOT be carried forward a second time"
1674        );
1675        assert!(
1676            !new_carried_n1.contains(&2),
1677            "child must NOT be in the new carried-forward set"
1678        );
1679    }
1680}