use super::*;
pub fn build_phase_buckets_with_stimulus(
samples: &crate::scenario::sample::SampleSeries,
stimulus_events: &[crate::timeline::StimulusEvent],
) -> Vec<PhaseBucket> {
let monitor_samples: &[crate::monitor::MonitorSample] =
samples.monitor().map(|m| m.samples()).unwrap_or(&[]);
let preemption_threshold_ns: u64 = samples
.monitor()
.map(|m| m.preemption_threshold_ns())
.unwrap_or(0);
let by_phase = samples.by_stimulus_phase(stimulus_events);
let monitor_to_window_offset_ms = monitor_clock_offset(stimulus_events, monitor_samples);
let mut buckets = buckets_from_grouped(
by_phase,
monitor_samples,
monitor_to_window_offset_ms,
preemption_threshold_ns,
);
let step_starts = crate::scenario::sample::step_starts_from_stimulus(stimulus_events);
synthesize_missing_step_buckets(
&mut buckets,
&step_starts,
stimulus_events,
monitor_samples,
monitor_to_window_offset_ms,
preemption_threshold_ns,
);
buckets.sort_by_key(|b| b.step_index);
fill_phase_iteration_rates(&mut buckets, stimulus_events);
buckets
}
fn synthesize_missing_step_buckets(
buckets: &mut Vec<PhaseBucket>,
step_starts: &[(u64, u16)],
stimulus_events: &[crate::timeline::StimulusEvent],
monitor_samples: &[crate::monitor::MonitorSample],
monitor_to_window_offset_ms: i64,
preemption_threshold_ns: u64,
) {
for &(start_ms, k) in step_starts {
if buckets.iter().any(|b| b.step_index == k) {
continue;
}
let step_end = stimulus_events
.iter()
.filter(|e| e.is_step_end && e.step_index == Some(k))
.map(|e| e.elapsed_ms)
.min();
let next_start = step_starts
.iter()
.filter(|&&(ms, _)| ms > start_ms)
.map(|&(ms, _)| ms)
.min();
let terminal = stimulus_events
.iter()
.find(|e| e.is_terminal)
.map(|e| e.elapsed_ms);
let end_ms = match (step_end, next_start) {
(Some(se), Some(ns)) => se.min(ns),
(Some(se), None) => se,
(None, Some(ns)) => ns,
(None, None) => terminal.unwrap_or(u64::MAX),
};
let label = if k == 0 {
"BASELINE".to_string()
} else {
format!("Step[{}]", k.saturating_sub(1))
};
let mut bucket = PhaseBucket {
step_index: k,
label,
start_ms,
end_ms,
sample_count: 0,
metrics: std::collections::BTreeMap::new(),
per_cgroup: std::collections::BTreeMap::new(),
};
fold_monitor_into_bucket(
&mut bucket,
monitor_samples,
monitor_to_window_offset_ms,
preemption_threshold_ns,
);
buckets.push(bucket);
}
}
fn fill_phase_iteration_rates(
buckets: &mut [PhaseBucket],
stimulus_events: &[crate::timeline::StimulusEvent],
) {
let mut sorted_events: Vec<&crate::timeline::StimulusEvent> = stimulus_events.iter().collect();
sorted_events.sort_by_key(|e| (e.elapsed_ms, !e.is_step_end));
for w in sorted_events.windows(2) {
let prev = w[0];
let curr = w[1];
if prev.is_terminal || prev.is_step_end {
continue;
}
let Some((iters, secs)) = prev.rate_components(curr) else {
continue;
};
for bucket in buckets.iter_mut() {
let in_bucket = match prev.step_index {
Some(k) => bucket.step_index == k,
None => {
if bucket.start_ms == bucket.end_ms {
prev.elapsed_ms == bucket.start_ms
} else {
prev.elapsed_ms >= bucket.start_ms && prev.elapsed_ms < bucket.end_ms
}
}
};
if in_bucket {
bucket
.metrics
.entry("total_phase_iterations".to_string())
.or_insert(iters);
bucket
.metrics
.entry("total_phase_duration_sec".to_string())
.or_insert(secs);
break;
}
}
}
for bucket in buckets.iter_mut() {
crate::stats::derive_rate_metrics(&mut bucket.metrics);
}
}
pub fn build_phase_buckets(samples: &crate::scenario::sample::SampleSeries) -> Vec<PhaseBucket> {
let monitor_samples: &[crate::monitor::MonitorSample] =
samples.monitor().map(|m| m.samples()).unwrap_or(&[]);
let preemption_threshold_ns: u64 = samples
.monitor()
.map(|m| m.preemption_threshold_ns())
.unwrap_or(0);
buckets_from_grouped(
samples.by_stamped_phase(),
monitor_samples,
0,
preemption_threshold_ns,
)
}
fn phase_group_cpu_delta(
samples: &[crate::scenario::sample::Sample<'_>],
task_field: impl Fn(&crate::monitor::task_enrichment::TaskEnrichment) -> u64,
signal_field: impl Fn(&crate::monitor::task_enrichment::TaskEnrichment) -> Option<u64>,
) -> Option<f64> {
let group_totals = |s: &crate::scenario::sample::Sample<'_>| {
let mut live: std::collections::HashMap<i32, u128> = std::collections::HashMap::new();
let mut signal: std::collections::HashMap<i32, Option<u128>> =
std::collections::HashMap::new();
for t in s.snapshot.task_enrichments() {
*live.entry(t.tgid).or_insert(0) += u128::from(task_field(t));
match signal_field(t) {
Some(v) => {
signal.insert(t.tgid, Some(u128::from(v)));
}
None => {
signal.entry(t.tgid).or_insert(None);
}
}
}
live.into_iter()
.filter_map(|(tgid, l)| {
signal
.get(&tgid)
.copied()
.flatten()
.map(|sig| (tgid, l + sig))
})
.collect::<std::collections::HashMap<i32, u128>>()
};
let mut ordered: Vec<&crate::scenario::sample::Sample<'_>> = samples.iter().collect();
ordered.sort_by_key(|s| s.boundary_offset_ms.or(s.elapsed_ms).unwrap_or(u64::MAX));
let mut first_seen: std::collections::HashMap<i32, u128> = std::collections::HashMap::new();
let mut last_seen: std::collections::HashMap<i32, u128> = std::collections::HashMap::new();
let mut readable_count: std::collections::HashMap<i32, u32> = std::collections::HashMap::new();
for s in ordered {
for (tgid, total) in group_totals(s) {
first_seen.entry(tgid).or_insert(total);
last_seen.insert(tgid, total);
*readable_count.entry(tgid).or_insert(0) += 1;
}
}
let mut sum: u128 = 0;
let mut measured = false;
for (tgid, last) in &last_seen {
if readable_count.get(tgid).copied().unwrap_or(0) < 2 {
continue;
}
measured = true;
let first = first_seen.get(tgid).copied().unwrap_or(*last);
sum += last.saturating_sub(first);
}
measured.then_some(sum as f64)
}
fn fold_per_cpu_spatial_max(
metrics: &mut std::collections::BTreeMap<String, f64>,
samples_in_phase: &[crate::scenario::sample::Sample<'_>],
win: impl Fn(&crate::scenario::sample::Sample<'_>) -> Option<u64>,
field: impl Fn(&crate::monitor::dump::PerCpuTimeStats) -> u64,
max_key: &str,
concentration_key: &str,
) {
let placed: Vec<(&crate::scenario::sample::Sample<'_>, u64)> = samples_in_phase
.iter()
.filter(|s| !s.snapshot.per_cpu_time().is_empty())
.filter_map(|s| win(s).map(|w| (s, w)))
.collect();
if let (Some(&(first_s, fw)), Some(&(last_s, lw))) = (
placed.iter().min_by_key(|(_, w)| *w),
placed.iter().max_by_key(|(_, w)| *w),
) && fw < lw
{
let deltas: Vec<f64> = first_s
.snapshot
.per_cpu_time()
.iter()
.filter_map(|c0| {
last_s
.snapshot
.per_cpu_time_at(c0.cpu)
.map(|c1| field(c1).saturating_sub(field(c0)) as f64)
})
.collect();
if let Some(max) = deltas.iter().copied().reduce(f64::max) {
metrics.entry(max_key.to_string()).or_insert(max);
if deltas.len() >= 2 {
let mean = deltas.iter().sum::<f64>() / deltas.len() as f64;
if mean > 0.0 {
metrics
.entry(concentration_key.to_string())
.or_insert(max / mean);
}
}
}
}
}
pub(crate) fn cpu_util_comp_scale(
first: &crate::monitor::dump::PerCpuTimeStats,
last: &crate::monitor::dump::PerCpuTimeStats,
) -> f64 {
let user = last.cpustat_user_ns.saturating_sub(first.cpustat_user_ns);
let nice = last.cpustat_nice_ns.saturating_sub(first.cpustat_nice_ns);
let system = last
.cpustat_system_ns
.saturating_sub(first.cpustat_system_ns);
let idle = last.cpustat_idle_ns.saturating_sub(first.cpustat_idle_ns);
let iowait = last
.cpustat_iowait_ns
.saturating_sub(first.cpustat_iowait_ns);
let irq = last.cpustat_irq_ns.saturating_sub(first.cpustat_irq_ns);
let softirq = last
.cpustat_softirq_ns
.saturating_sub(first.cpustat_softirq_ns);
let steal = last.cpustat_steal_ns.saturating_sub(first.cpustat_steal_ns);
let delta_total = [user, nice, system, idle, iowait, irq, softirq, steal]
.into_iter()
.fold(0u64, u64::saturating_add);
let overhead = irq.saturating_add(softirq).saturating_add(steal);
let available = delta_total.saturating_sub(overhead);
if available == 0 {
return 1.0;
}
(delta_total as f64 / available as f64).clamp(1.0, 20.0)
}
fn fold_util_comp_scale(
metrics: &mut std::collections::BTreeMap<String, f64>,
samples_in_phase: &[crate::scenario::sample::Sample<'_>],
win: impl Fn(&crate::scenario::sample::Sample<'_>) -> Option<u64>,
) {
let placed: Vec<(&crate::scenario::sample::Sample<'_>, u64)> = samples_in_phase
.iter()
.filter(|s| !s.snapshot.per_cpu_time().is_empty())
.filter_map(|s| win(s).map(|w| (s, w)))
.collect();
if let (Some(&(first_s, fw)), Some(&(last_s, lw))) = (
placed.iter().min_by_key(|(_, w)| *w),
placed.iter().max_by_key(|(_, w)| *w),
) && fw < lw
{
let scales: Vec<f64> = first_s
.snapshot
.per_cpu_time()
.iter()
.filter_map(|c0| {
last_s
.snapshot
.per_cpu_time_at(c0.cpu)
.map(|c1| cpu_util_comp_scale(c0, c1))
})
.collect();
if !scales.is_empty() {
let mean = scales.iter().sum::<f64>() / scales.len() as f64;
metrics
.entry("avg_cpu_util_comp_scale".to_string())
.or_insert(mean);
}
}
}
fn fold_lat_cri(
metrics: &mut std::collections::BTreeMap<String, f64>,
samples_in_phase: &[crate::scenario::sample::Sample<'_>],
) {
let mut sum = 0.0f64;
let mut count = 0usize;
let mut max = f64::MIN;
for s in samples_in_phase {
for alloc in &s.snapshot.report().sdt_allocations {
for entry in &alloc.entries {
if let Some(v) = entry
.payload
.get("normalized_lat_cri")
.and_then(|r| r.as_u64())
{
let v = v as f64;
sum += v;
count += 1;
if v > max {
max = v;
}
}
}
}
}
if count > 0 {
metrics
.entry("avg_task_lat_cri".to_string())
.or_insert(sum / count as f64);
metrics.entry("max_task_lat_cri".to_string()).or_insert(max);
}
}
fn fold_per_cgroup_psi(
metrics: &mut std::collections::BTreeMap<String, f64>,
samples_in_phase: &[crate::scenario::sample::Sample<'_>],
win: impl Fn(&crate::scenario::sample::Sample<'_>) -> Option<u64>,
) {
use crate::monitor::btf_offsets::{decode_avg10_percent, decode_total_us};
let avg10_peak = samples_in_phase
.iter()
.filter_map(|s| {
s.snapshot
.cgroup_psi()
.iter()
.map(|c| decode_avg10_percent(c.avg10_raw))
.reduce(f64::max)
})
.reduce(f64::max);
if let Some(v) = avg10_peak {
metrics
.entry("max_cgroup_psi_irq_avg10".to_string())
.or_insert(v);
}
let placed: Vec<(&crate::scenario::sample::Sample<'_>, u64)> = samples_in_phase
.iter()
.filter(|s| !s.snapshot.cgroup_psi().is_empty())
.filter_map(|s| win(s).map(|w| (s, w)))
.collect();
if let (Some(&(first_s, fw)), Some(&(last_s, lw))) = (
placed.iter().min_by_key(|(_, w)| *w),
placed.iter().max_by_key(|(_, w)| *w),
) && fw < lw
{
let deltas: Vec<f64> = first_s
.snapshot
.cgroup_psi()
.iter()
.filter_map(|c0| {
last_s
.snapshot
.cgroup_psi()
.iter()
.find(|c1| c1.cgroup_kva == c0.cgroup_kva && c1.serial_nr == c0.serial_nr)
.map(|c1| decode_total_us(c1.total_ns.saturating_sub(c0.total_ns)))
})
.collect();
if let Some(max) = deltas.iter().copied().reduce(f64::max) {
metrics
.entry("max_cgroup_irq_pressure".to_string())
.or_insert(max);
if deltas.len() >= 2 {
let mean = deltas.iter().sum::<f64>() / deltas.len() as f64;
if mean > 0.0 {
metrics
.entry("max_cgroup_irq_pressure_concentration".to_string())
.or_insert(max / mean);
}
}
}
}
}
fn buckets_from_grouped(
by_phase: std::collections::BTreeMap<u16, Vec<crate::scenario::sample::Sample<'_>>>,
monitor_samples: &[crate::monitor::MonitorSample],
monitor_to_window_offset_ms: i64,
preemption_threshold_ns: u64,
) -> Vec<PhaseBucket> {
let mut out: Vec<PhaseBucket> = Vec::with_capacity(by_phase.len());
for (step_index, samples_in_phase) in by_phase {
let label = if step_index == 0 {
"BASELINE".to_string()
} else {
format!("Step[{}]", step_index.saturating_sub(1))
};
let sample_count = samples_in_phase.len();
let win = |s: &crate::scenario::sample::Sample<'_>| s.boundary_offset_ms.or(s.elapsed_ms);
let (start_ms, end_ms) = if samples_in_phase.is_empty() {
(0, u64::MAX)
} else {
let mut lo = u64::MAX;
let mut hi = 0u64;
for s in &samples_in_phase {
if let Some(w) = win(s) {
lo = lo.min(w);
hi = hi.max(w);
}
}
(lo, hi)
};
let mut metrics: std::collections::BTreeMap<String, f64> =
std::collections::BTreeMap::new();
for metric_def in crate::stats::METRICS {
let per_sample_readings: Vec<f64> = samples_in_phase
.iter()
.filter_map(|s| metric_def.read_sample(s))
.collect();
if per_sample_readings.is_empty() {
continue;
}
if let Some(reduced) =
crate::stats::aggregate_samples_for_phase(metric_def, &per_sample_readings)
{
metrics.insert(metric_def.name.to_string(), reduced);
}
}
if let Some(v) = phase_group_cpu_delta(&samples_in_phase, |t| t.stime, |t| t.signal_stime) {
metrics.entry("system_time_ns".to_string()).or_insert(v);
}
if let Some(v) = phase_group_cpu_delta(&samples_in_phase, |t| t.utime, |t| t.signal_utime) {
metrics.entry("user_time_ns".to_string()).or_insert(v);
}
fold_per_cpu_spatial_max(
&mut metrics,
&samples_in_phase,
win,
|c| c.irqs_sum,
"max_cpu_hardirqs",
"max_cpu_hardirq_concentration",
);
fold_per_cpu_spatial_max(
&mut metrics,
&samples_in_phase,
win,
|c| c.softirqs[crate::monitor::btf_offsets::SOFTIRQ_NET_RX],
"max_cpu_softirq_net_rx",
"max_cpu_softirq_net_rx_concentration",
);
fold_util_comp_scale(&mut metrics, &samples_in_phase, win);
fold_lat_cri(&mut metrics, &samples_in_phase);
fold_per_cgroup_psi(&mut metrics, &samples_in_phase, win);
let mut bucket = PhaseBucket {
step_index,
label,
start_ms,
end_ms,
sample_count,
metrics,
per_cgroup: std::collections::BTreeMap::new(),
};
fold_monitor_into_bucket(
&mut bucket,
monitor_samples,
monitor_to_window_offset_ms,
preemption_threshold_ns,
);
if bucket.metrics.contains_key("total_hardirqs") && bucket.end_ms > bucket.start_ms {
let wall_ms = (bucket.end_ms - bucket.start_ms) as f64;
bucket
.metrics
.entry("total_phase_wall_ns".to_string())
.or_insert(wall_ms * 1_000_000.0);
bucket
.metrics
.entry("total_phase_wall_sec".to_string())
.or_insert(wall_ms / 1000.0);
}
crate::stats::derive_rate_metrics(&mut bucket.metrics);
out.push(bucket);
}
out
}
fn fold_monitor_into_bucket(
bucket: &mut PhaseBucket,
monitor_samples: &[crate::monitor::MonitorSample],
monitor_to_window_offset_ms: i64,
preemption_threshold_ns: u64,
) {
let start_ms = bucket.start_ms;
let end_ms = bucket.end_ms;
let in_window = |monitor_ms: u64| -> bool {
let shifted = monitor_ms as i64 - monitor_to_window_offset_ms;
if shifted < 0 {
return false;
}
let m = shifted as u64;
if start_ms == end_ms {
m == start_ms
} else {
m >= start_ms && m < end_ms
}
};
let phase_monitor_samples: Vec<&crate::monitor::MonitorSample> = monitor_samples
.iter()
.filter(|s| in_window(s.elapsed_ms))
.filter(|s| crate::monitor::sample_looks_valid(s))
.collect();
if phase_monitor_samples.is_empty() {
return;
}
let pm = crate::timeline::compute_metrics(&phase_monitor_samples, preemption_threshold_ns);
let synthesized = bucket.sample_count == 0;
let mut put = |key: &str, v: f64| {
if v.is_finite() {
bucket.metrics.entry(key.to_string()).or_insert(v);
}
};
if let Some(v) = pm.avg_imbalance {
put("avg_imbalance_ratio", v);
}
if let Some(v) = pm.avg_nr_running {
put("avg_nr_running", v);
}
if let Some(v) = pm.max_imbalance {
put("max_imbalance_ratio", v);
}
if pm.stall_count > 0 {
put("stuck_count", pm.stall_count as f64);
}
if synthesized {
if let Some(v) = pm.avg_dsq_depth {
put("avg_dsq_depth", v);
}
put("max_dsq_depth", pm.max_dsq_depth as f64);
let has_events =
|s: &&crate::monitor::MonitorSample| s.cpus.iter().any(|c| c.event_counters.is_some());
let first_ev = phase_monitor_samples.iter().copied().find(has_events);
let last_ev = phase_monitor_samples.iter().copied().rev().find(has_events);
if let (Some(first), Some(last)) = (first_ev, last_ev) {
let fb = crate::monitor::counter_delta(
last.sum_event_field(|e| e.select_cpu_fallback).unwrap_or(0),
first
.sum_event_field(|e| e.select_cpu_fallback)
.unwrap_or(0),
);
let kl = crate::monitor::counter_delta(
last.sum_event_field(|e| e.dispatch_keep_last).unwrap_or(0),
first.sum_event_field(|e| e.dispatch_keep_last).unwrap_or(0),
);
put("total_fallback", fb as f64);
put("total_keep_last", kl as f64);
}
}
}
fn monitor_clock_offset(
stimulus_events: &[crate::timeline::StimulusEvent],
monitor_samples: &[crate::monitor::MonitorSample],
) -> i64 {
if stimulus_events.is_empty() || monitor_samples.is_empty() {
return 0;
}
let first_stimulus_ms = stimulus_events
.iter()
.map(|e| e.elapsed_ms)
.min()
.unwrap_or(0);
let first_monitor_ms = monitor_samples
.iter()
.find(|s| s.elapsed_ms > 500 && !s.cpus.is_empty())
.map(|s| s.elapsed_ms)
.unwrap_or_else(|| monitor_samples.first().map(|s| s.elapsed_ms).unwrap_or(0));
first_monitor_ms as i64 - first_stimulus_ms as i64
}