ktstr/ctprof_compare/

aggregate.rs

//! Per-group aggregate value carrier and the merge primitives
//! that compose it.
//!
//! Two layers:
//!
//! 1. [`Aggregated`] — closed enum of the per-metric aggregate
//!    shapes a thread group produces: [`Aggregated::Sum`] /
//!    [`Aggregated::Max`] for the numeric reduction rules,
//!    [`Aggregated::OrdinalRange`] for `[min, max]` interval rules
//!    over signed kernel ordinals, [`Aggregated::Mode`] for the
//!    categorical mode-with-tally aggregation, and
//!    [`Aggregated::Affinity`] for the cpuset-cardinality summary.
//!    Each variant carries the shape its rule's accessor produced
//!    after the cross-thread reduction. `numeric()` projects to
//!    `Option<f64>` for the delta-math / sort path; categorical
//!    rules return `None`. The [`std::fmt::Display`] impl is
//!    consumed by [`super::scale::format_value_cell`] when the
//!    rule's ladder is [`super::ScaleLadder::None`].
//!
//! 2. [`merge_aggregated_into`] — the per-variant merge primitive
//!    that composes summaries when N candidate groups collapse
//!    onto one baseline-keyed row (the N:1 fudge merge): Sum
//!    counters add, Max peaks take max-of-maxes (NOT a sum —
//!    summing peaks invents peaks taller than any thread
//!    observed), OrdinalRange unions the bounds, Mode unions the
//!    per-value tally maps, Affinity widens the cardinality bounds
//!    and collapses the exact CPU list to `None` on any mismatch.
//!    Variant mismatches are silently dropped — every group's
//!    metrics map shares the same [`super::CTPROF_METRICS`] keys,
//!    so the fall-through arm is defense-in-depth.
//!
//! The [`AffinitySummary`] CPU-set cardinality carrier and the
//! [`format_cpu_range`] runs-collapsing renderer travel with
//! [`Aggregated::Affinity`] because the renderer is consumed only
//! by the Affinity arm of the [`std::fmt::Display`] impl.

use std::collections::BTreeMap;
use std::fmt;

/// Aggregated metric value for a single [`super::ThreadGroup`].
///
/// Carries both a numeric projection (used for delta math and
/// sort order) and a display form. Not every rule produces a
/// numeric — the categorical rules
/// ([`super::AggRule::Mode`] / [`super::AggRule::ModeChar`] /
/// [`super::AggRule::ModeBool`]) aggregate to a string, which has no
/// scalar — so the numeric is optional and rows without one
/// fall to the bottom of the default sort.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum Aggregated {
    /// Group-wide sum produced by the
    /// [`super::AggRule::SumCount`] / [`super::AggRule::SumNs`] /
    /// [`super::AggRule::SumTicks`] / [`super::AggRule::SumBytes`] rules. The
    /// dispatch unwraps the typed newtype's inner `u64` after
    /// the [`crate::metric_types::Summable::sum_across`]
    /// reduction; storage stays u64 to preserve full precision
    /// across the entire schedstats / byte / tick range with
    /// no lossy cast at aggregation time. The render-time
    /// auto-scale ladder is derived from the typed accessor
    /// newtype via [`super::AggRule::ladder`] — there is no
    /// separate `unit` tag on the metric def.
    Sum(u64),
    /// Group-wide maximum produced by the
    /// [`super::AggRule::MaxPeak`] / [`super::AggRule::MaxGaugeNs`] /
    /// [`super::AggRule::MaxGaugeCount`] rules. Distinct variant from
    /// `Sum` so a downstream consumer that wants to surface
    /// "the worst single thread" rather than "the
    /// summed-across-threads value" can match without name-
    /// matching against the metric registry. Storage is u64 to
    /// preserve full ns precision across the entire schedstats
    /// range (no `as f64` lossy cast at aggregation time).
    Max(u64),
    /// Group-wide `[min, max]` interval produced by the
    /// [`super::AggRule::RangeI32`] / [`super::AggRule::RangeU32`] rules.
    /// Both bounds widen to `i64` at the dispatch boundary
    /// (`i64::from(OrdinalI32.0)` / `i64::from(OrdinalU32.0)`)
    /// — `OrdinalI32` carries a signed kernel-side range
    /// (`nice` includes negative values) and `OrdinalU32` fits
    /// into `i64` losslessly, so a single signed scalar
    /// represents both ordinal widths without losing the sign
    /// from `OrdinalI32` or wrapping the magnitude from
    /// `OrdinalU32`. Delta math takes the midpoint
    /// (`(min + max) / 2`) so a one-sided shift surfaces in
    /// the rendered delta column.
    OrdinalRange {
        min: i64,
        max: i64,
    },
    /// Categorical aggregate carrying per-value counts across
    /// the bucket. `tallies` maps each observed value to its
    /// occurrence count; `total` is the bucket size (count of
    /// every contributor, including any empty-string fallbacks).
    /// The mode (most-frequent value) is derived on demand via
    /// [`Aggregated::mode_value`] / [`Aggregated::mode_count`]
    /// so cross-bucket merges (N:1 fudge) compose correctly:
    /// unioning two buckets' tallies gives the true cross-bucket
    /// frequency for each value, not just the per-bucket
    /// max-count. Empty buckets surface as `tallies: empty,
    /// total: bucket_size`; the renderer handles the empty case
    /// by emitting `~` in place of the mode value.
    Mode {
        tallies: BTreeMap<String, usize>,
        total: usize,
    },
    Affinity(AffinitySummary),
    /// The metric's capture-gated family was NOT measured for any thread in the
    /// group — the taskstats genetlink query failed for every thread (the
    /// query-level `ThreadState::taskstats_measured` flag is false: CONFIG_TASKSTATS
    /// off / no CAP_NET_ADMIN / query raced exit), or — for the jemalloc byte
    /// counters — no thread's per-thread probe READ succeeded (non-jemalloc
    /// process, probe could not attach, or the read failed) — distinct from a
    /// measured zero. (A sub-family disabled while the query still succeeds — e.g.
    /// CONFIG_TASK_XACCT off, or the `kernel.task_delayacct` sysctl off, with the
    /// other family on — now IS Absent: the measured predicate ANDs the per-thread
    /// query-Ok with the per-sub-family active flag baked from host probes. See the
    /// `taskstats_measured` field doc.) Produced by [fn@super::aggregate]'s
    /// caller when the family has a
    /// measured predicate, the bucket is non-empty, and no contributing thread
    /// captured it. `numeric()` returns `None`, so a derived metric consuming it
    /// (e.g. `total_offcpu_delay_ns`, `live_heap_estimate`) short-circuits to
    /// absent rather than computing from a sentinel `0`, and the renderer prints
    /// "-" rather than "0". A MEASURED zero stays `Sum(0)` and renders "0".
    Absent,
}

/// CPU-affinity aggregation result.
///
/// `uniform` is `Some(cpus)` when every thread in the group shared
/// the same allowed set; otherwise heterogeneous and the renderer
/// emits "N-M cpus (mixed)".
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct AffinitySummary {
    pub min_cpus: usize,
    pub max_cpus: usize,
    pub uniform: Option<Vec<u32>>,
}

impl Aggregated {
    /// Scalar projection for delta math. `None` when the rule
    /// produces no meaningful scalar (categorical mode, affinity
    /// with heterogeneous cpusets).
    pub fn numeric(&self) -> Option<f64> {
        match self {
            Aggregated::Sum(v) => Some(*v as f64),
            Aggregated::Max(v) => Some(*v as f64),
            Aggregated::OrdinalRange { min, max } => {
                // Midpoint: keeps a min→max shift on one end visible
                // in the delta without privileging either bound.
                Some((*min as f64 + *max as f64) / 2.0)
            }
            Aggregated::Mode { .. } => None,
            // Capture-gated family not measured for the group → no scalar, so a
            // consuming derived metric short-circuits to absent ("-").
            Aggregated::Absent => None,
            Aggregated::Affinity(s) => {
                // Number of allowed CPUs is the natural scalar. When
                // the group is uniform, `min_cpus == max_cpus`; when
                // heterogeneous, midpoint parallels OrdinalRange.
                Some((s.min_cpus as f64 + s.max_cpus as f64) / 2.0)
            }
        }
    }

    /// Construct an `Aggregated::Mode` from a single
    /// (value, count, total) triple, common in test fixtures
    /// and the single-bucket aggregate path. Equivalent to
    /// inserting `(value, count)` into a fresh tally map.
    /// `count == 0` produces an empty tally (the empty-bucket
    /// shape), preserving the historical contract that an
    /// empty Mode renders as `~` while still carrying a
    /// well-defined `total` for the rendered "(count/total)"
    /// suffix.
    pub fn mode_single(value: impl Into<String>, count: usize, total: usize) -> Aggregated {
        let mut tallies = BTreeMap::new();
        if count > 0 {
            tallies.insert(value.into(), count);
        }
        Aggregated::Mode { tallies, total }
    }

    /// The most-frequent value tracked by an `Aggregated::Mode`,
    /// or the empty string when the tallies map is empty (an
    /// empty bucket whose `total` may still be non-zero, e.g.
    /// when every contributor produced no categorical sample).
    /// Ties on count are broken by lexicographic order on the
    /// VALUE — the lex-smallest key wins. Deterministic across
    /// runs and matches the historical Mode contract under the
    /// old `value, count` shape.
    pub fn mode_value(&self) -> &str {
        match self {
            Aggregated::Mode { tallies, .. } => {
                // BTreeMap iterates keys in lex order. Use a
                // strict-greater fold so on a count tie the
                // first key encountered (lex-smallest) wins;
                // `max_by_key` returns the last-seen at a tie,
                // which would invert the lex order.
                let mut best: Option<(&str, usize)> = None;
                for (k, c) in tallies {
                    match best {
                        None => best = Some((k.as_str(), *c)),
                        Some((_, bc)) if *c > bc => best = Some((k.as_str(), *c)),
                        _ => {}
                    }
                }
                best.map(|(v, _)| v).unwrap_or("")
            }
            _ => "",
        }
    }

    /// The frequency count of the mode value. Zero when the
    /// tallies map is empty (no contributor produced a
    /// categorical sample). Same tie-break as
    /// [`Self::mode_value`].
    pub fn mode_count(&self) -> usize {
        match self {
            Aggregated::Mode { tallies, .. } => tallies.values().copied().max().unwrap_or(0),
            _ => 0,
        }
    }
}

impl fmt::Display for Aggregated {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Aggregated::Sum(v) => write!(f, "{v}"),
            Aggregated::Max(v) => write!(f, "{v}"),
            Aggregated::OrdinalRange { min, max } => {
                if min == max {
                    write!(f, "{min}")
                } else {
                    write!(f, "{min}..{max}")
                }
            }
            Aggregated::Mode { tallies, total } => {
                // Tie-break: lex-smallest value wins on count
                // ties. Mirrors `Aggregated::mode_value`'s
                // strict-greater fold.
                let mut best: Option<(&str, usize)> = None;
                for (k, c) in tallies {
                    match best {
                        None => best = Some((k.as_str(), *c)),
                        Some((_, bc)) if *c > bc => best = Some((k.as_str(), *c)),
                        _ => {}
                    }
                }
                let value = best.map(|(v, _)| v).unwrap_or("~");
                let count = best.map(|(_, c)| c).unwrap_or(0);
                if count == *total && count > 0 {
                    write!(f, "{value}")
                } else {
                    write!(f, "{value} ({count}/{total})")
                }
            }
            Aggregated::Affinity(s) => {
                if let Some(cpus) = &s.uniform {
                    let n = cpus.len();
                    let range = format_cpu_range(cpus);
                    write!(f, "{n} cpus ({range})")
                } else if s.min_cpus == s.max_cpus {
                    write!(f, "{} cpus (mixed)", s.min_cpus)
                } else {
                    write!(f, "{}-{} cpus (mixed)", s.min_cpus, s.max_cpus)
                }
            }
            // Not-measured marker. The unit-suffixed render path
            // (`scale::format_value_cell`) special-cases Absent to a bare "-"
            // (no unit ladder); this bare Display is the fallback for any other
            // consumer, so it carries no unit either.
            Aggregated::Absent => write!(f, "-"),
        }
    }
}

/// Merge `val` into `existing` for the N:1 fudge aggregation
/// (multiple candidate groups collapsed into one baseline-keyed
/// row). Mirrors the canonical [fn@super::aggregate] semantics on the
/// merged-set rather than re-aggregating the per-thread inputs:
/// per-group summaries are all the merge has access to, so the
/// merge composes summaries directly.
///
/// Per variant:
/// - [`Aggregated::Sum`]: monotone counters add.
/// - [`Aggregated::Max`]: peaks take max-of-maxes (NOT a sum —
///   wait_max / sleep_max / exec_max are per-thread peaks; summing
///   would invent peaks taller than any thread observed).
/// - [`Aggregated::OrdinalRange`]: union the bounds.
/// - [`Aggregated::Mode`]: union the per-value tally maps and
///   sum the totals. The mode (most-frequent value) is derived
///   on demand from the merged tallies, so cross-bucket
///   frequencies stay accurate — a value that appears in N
///   buckets accumulates its true total count, not just the
///   largest single-bucket count.
/// - [`Aggregated::Affinity`]: cardinality bounds widen —
///   `min_cpus` becomes the smallest of any merged summary's
///   `min_cpus`, `max_cpus` the largest. The exact CPU list
///   (`uniform`) survives only when every merged summary
///   already carried the same `Some(cpus)` list; mismatch or
///   any `None` collapses to `None`.
///
/// Variant mismatches between `existing` and `val` are silently
/// dropped — a fudged pair must agree on the metric's typed
/// rule because every group's metrics map shares the same
/// [`super::CTPROF_METRICS`] keys, so the fall-through arm is
/// defense-in-depth rather than a runtime-reachable case.
pub(super) fn merge_aggregated_into(existing: &mut Aggregated, val: &Aggregated) {
    match (existing, val) {
        (Aggregated::Sum(s), Aggregated::Sum(v)) => {
            // saturating_add to match the canonical aggregation
            // (`Summable::sum_across`, metric_types.rs): a corrupt or
            // hostile reading that would push the merged sum past
            // u64::MAX saturates rather than wrapping (release) or
            // panicking (debug). This N:1 fudge merge composes
            // already-aggregated Sum values, so the overflow is
            // reachable on adversarial input.
            *s = (*s).saturating_add(*v);
        }
        (Aggregated::Max(m), Aggregated::Max(v)) => {
            *m = (*m).max(*v);
        }
        (
            Aggregated::OrdinalRange { min, max },
            Aggregated::OrdinalRange {
                min: vmin,
                max: vmax,
            },
        ) => {
            *min = (*min).min(*vmin);
            *max = (*max).max(*vmax);
        }
        (
            Aggregated::Mode { tallies: et, total },
            Aggregated::Mode {
                tallies: vt,
                total: vtot,
            },
        ) => {
            // saturating_add on the tally totals/counts for parity
            // with the codebase's other counter merges (frac_pair's
            // saturating_add, aggregate_finite's checked_add) — a
            // tally is bounded by the sample population today, but
            // unchecked `+=` would panic in debug / wrap in release if
            // that ever changed, inconsistent with sibling sites.
            *total = (*total).saturating_add(*vtot);
            for (k, c) in vt {
                let e = et.entry(k.clone()).or_insert(0);
                *e = (*e).saturating_add(*c);
            }
        }
        (Aggregated::Affinity(es), Aggregated::Affinity(vs)) => {
            es.min_cpus = es.min_cpus.min(vs.min_cpus);
            es.max_cpus = es.max_cpus.max(vs.max_cpus);
            es.uniform = match (&es.uniform, &vs.uniform) {
                (Some(a), Some(b)) if a == b => Some(a.clone()),
                _ => None,
            };
        }
        // Not-measured marker — measured wins. An Absent `existing` is replaced
        // by `val` (so a measured group's value survives a fudge merge with an
        // unmeasured one; both-Absent stays Absent since `val` is also Absent);
        // an Absent `val` contributes nothing to a measured `existing`. This
        // keeps a fudged row's measured-ness as "any contributor measured it",
        // matching the aggregate-level rule.
        (existing_slot @ Aggregated::Absent, v) => *existing_slot = v.clone(),
        (_, Aggregated::Absent) => {}
        _ => {}
    }
}

/// Render a CPU set as a comma-separated list of contiguous-run
/// ranges (`a-b`), matching the kernel's cpuset display
/// convention (`cat cpuset.cpus` emits `0-3,8`). Assumes the
/// input is sorted ascending — capture layer
/// (`crate::ctprof::ThreadState::cpu_affinity`) stores
/// sorted cpusets. Empty input returns an empty string. Used
/// only by `Aggregated`'s `Display::fmt` Affinity arm; kept here so the
/// renderer and its helper travel together.
pub(super) fn format_cpu_range(cpus: &[u32]) -> String {
    // Collapse contiguous runs to `a-b`, join with commas. Assumes
    // sorted ascending; capture layer stores sorted cpusets.
    if cpus.is_empty() {
        return String::new();
    }
    let mut out = String::new();
    let mut start = cpus[0];
    let mut prev = cpus[0];
    for &c in &cpus[1..] {
        if c == prev + 1 {
            prev = c;
            continue;
        }
        if !out.is_empty() {
            out.push(',');
        }
        if start == prev {
            out.push_str(&start.to_string());
        } else {
            out.push_str(&format!("{start}-{prev}"));
        }
        start = c;
        prev = c;
    }
    if !out.is_empty() {
        out.push(',');
    }
    if start == prev {
        out.push_str(&start.to_string());
    } else {
        out.push_str(&format!("{start}-{prev}"));
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The N:1 fudge merge composes already-aggregated `Sum(u64)`
    /// values; merging two `u64::MAX` must saturate (matching
    /// `Summable::sum_across`), not wrap in release or panic in debug.
    #[test]
    fn merge_sum_saturates_at_u64_max() {
        let mut existing = Aggregated::Sum(u64::MAX);
        merge_aggregated_into(&mut existing, &Aggregated::Sum(u64::MAX));
        let Aggregated::Sum(s) = existing else {
            panic!("expected Sum variant");
        };
        assert_eq!(s, u64::MAX, "merged Sum must saturate, not wrap");
    }
}