From c9156e69d8b838b2cbffbf6608c15860b7cfaf60 Mon Sep 17 00:00:00 2001 From: Guillaume Binet Date: Mon, 3 Aug 2026 08:27:27 -0500 Subject: [PATCH] feat(logstats): export observed schedule timing --- api/v1/cu29-export.txt | 29 ++ core/cu29_export/src/logstats.rs | 603 ++++++++++++++++++++++++++--- core/cu29_runtime/src/rendercfg.rs | 12 +- 3 files changed, 572 insertions(+), 72 deletions(-) diff --git a/api/v1/cu29-export.txt b/api/v1/cu29-export.txt index 06f08cc906c..a1b5a4eca87 100644 --- a/api/v1/cu29-export.txt +++ b/api/v1/cu29-export.txt @@ -4,6 +4,8 @@ pub struct cu29_export::logstats::DurationStats pub cu29_export::logstats::DurationStats::max_ns: core::option::Option pub cu29_export::logstats::DurationStats::mean_ns: core::option::Option pub cu29_export::logstats::DurationStats::min_ns: core::option::Option +pub cu29_export::logstats::DurationStats::p50_ns: core::option::Option +pub cu29_export::logstats::DurationStats::p95_ns: core::option::Option pub cu29_export::logstats::DurationStats::stddev_ns: core::option::Option pub struct cu29_export::logstats::EdgeLogStats pub cu29_export::logstats::EdgeLogStats::avg_raw_bytes: core::option::Option @@ -18,17 +20,44 @@ pub cu29_export::logstats::EdgeLogStats::src_channel: core::option::Option pub cu29_export::logstats::EdgeLogStats::total_raw_bytes: u64 pub cu29_export::logstats::EdgeLogStats::valid_time_samples: u64 +pub struct cu29_export::logstats::ExecutionInterval +pub cu29_export::logstats::ExecutionInterval::end_ns: u64 +pub cu29_export::logstats::ExecutionInterval::origin: alloc::string::String +pub cu29_export::logstats::ExecutionInterval::start_ns: u64 +pub struct cu29_export::logstats::ExecutionTrace +pub cu29_export::logstats::ExecutionTrace::culist_id: u64 +pub cu29_export::logstats::ExecutionTrace::duration_ns: u64 +pub cu29_export::logstats::ExecutionTrace::excluded_intervals: u32 +pub cu29_export::logstats::ExecutionTrace::intervals: alloc::vec::Vec +pub cu29_export::logstats::ExecutionTrace::kind: alloc::string::String +pub cu29_export::logstats::ExecutionTrace::residual_before_ns: core::option::Option +pub cu29_export::logstats::ExecutionTrace::wall_span_ns: u64 pub struct cu29_export::logstats::LogStats pub cu29_export::logstats::LogStats::config_signature: alloc::string::String pub cu29_export::logstats::LogStats::edges: alloc::vec::Vec pub cu29_export::logstats::LogStats::mission: core::option::Option pub cu29_export::logstats::LogStats::perf: cu29_export::logstats::PerfStats +pub cu29_export::logstats::LogStats::schedule: core::option::Option pub cu29_export::logstats::LogStats::schema_version: u32 pub struct cu29_export::logstats::PerfStats pub cu29_export::logstats::PerfStats::end_to_end: cu29_export::logstats::DurationStats pub cu29_export::logstats::PerfStats::jitter: cu29_export::logstats::DurationStats pub cu29_export::logstats::PerfStats::samples: u64 pub cu29_export::logstats::PerfStats::valid_time_samples: u64 +pub struct cu29_export::logstats::ResourceOverlapStats +pub cu29_export::logstats::ResourceOverlapStats::left: alloc::string::String +pub cu29_export::logstats::ResourceOverlapStats::occurrences: u64 +pub cu29_export::logstats::ResourceOverlapStats::resource: alloc::string::String +pub cu29_export::logstats::ResourceOverlapStats::right: alloc::string::String +pub struct cu29_export::logstats::ScheduleLogStats +pub cu29_export::logstats::ScheduleLogStats::residual_before: cu29_export::logstats::DurationStats +pub cu29_export::logstats::ScheduleLogStats::resource_overlaps: alloc::vec::Vec +pub cu29_export::logstats::ScheduleLogStats::stages: alloc::vec::Vec +pub cu29_export::logstats::ScheduleLogStats::traces: alloc::vec::Vec +pub struct cu29_export::logstats::StageLogStats +pub cu29_export::logstats::StageLogStats::durations: cu29_export::logstats::DurationStats +pub cu29_export::logstats::StageLogStats::origin: alloc::string::String +pub cu29_export::logstats::StageLogStats::samples: u64 pub fn cu29_export::logstats::compute_logstats

(reader: impl std::io::Read, config: &cu29_runtime::config::CuConfig, mission: core::option::Option<&str>) -> cu29_traits::CuResult where P: cu29_traits::CopperListTuple + cu29_traits::CuPayloadRawBytes pub fn cu29_export::logstats::write_logstats(stats: &cu29_export::logstats::LogStats, path: &std::path::Path) -> cu29_traits::CuResult<()> pub enum cu29_export::Command diff --git a/core/cu29_export/src/logstats.rs b/core/cu29_export/src/logstats.rs index 3b4cbeea9ab..bcbe5c72332 100644 --- a/core/cu29_export/src/logstats.rs +++ b/core/cu29_export/src/logstats.rs @@ -6,13 +6,16 @@ use cu29::monitoring::CuDurationStatistics; use cu29::prelude::{CopperListTuple, CuMsgMetadataTrait, CuPayloadRawBytes}; use cu29::{CuError, CuResult}; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::fs::File; use std::io::Read; use std::path::Path; -const LOGSTATS_SCHEMA_VERSION: u32 = 1; +const LOGSTATS_SCHEMA_VERSION: u32 = 2; const MAX_LATENCY_NS: u64 = 10_000_000_000; +const MAX_QUANTILE_SAMPLES: usize = 2_048; +const MAX_REPRESENTATIVE_TRACES: usize = 256; +const MIN_CURRENT_CLUSTER_GAP_NS: u64 = 100_000; #[derive(Debug, Serialize, Deserialize)] pub struct LogStats { @@ -21,6 +24,52 @@ pub struct LogStats { pub mission: Option, pub edges: Vec, pub perf: PerfStats, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub schedule: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ScheduleLogStats { + pub stages: Vec, + pub traces: Vec, + pub residual_before: DurationStats, + pub resource_overlaps: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct StageLogStats { + pub origin: String, + pub samples: u64, + pub durations: DurationStats, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ExecutionTrace { + pub kind: String, + pub culist_id: u64, + /// Sum of recorded process intervals in this CopperList. + pub duration_ns: u64, + /// Wall-clock span from the first through last retained interval. + pub wall_span_ns: u64, + /// Message slots carrying timestamps outside the current execution cluster. + pub excluded_intervals: u32, + pub residual_before_ns: Option, + pub intervals: Vec, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ExecutionInterval { + pub origin: String, + pub start_ns: u64, + pub end_ns: u64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ResourceOverlapStats { + pub resource: String, + pub left: String, + pub right: String, + pub occurrences: u64, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -55,6 +104,78 @@ pub struct DurationStats { pub max_ns: Option, pub mean_ns: Option, pub stddev_ns: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub p50_ns: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub p95_ns: Option, +} + +#[derive(Debug, Default)] +struct TimingAccumulator { + samples: u64, + min: Option, + max: Option, + sum: f64, + sum_squares: f64, + quantiles: WeightedQuantiles, +} + +impl TimingAccumulator { + fn record(&mut self, value: u64) { + self.samples = self.samples.saturating_add(1); + self.min = Some(self.min.map_or(value, |current| current.min(value))); + self.max = Some(self.max.map_or(value, |current| current.max(value))); + let value_f64 = value as f64; + self.sum += value_f64; + self.sum_squares += value_f64 * value_f64; + self.quantiles.record(value); + } + + fn stats(&self) -> DurationStats { + if self.samples == 0 { + return DurationStats::default(); + } + let mean = self.sum / self.samples as f64; + let variance = (self.sum_squares / self.samples as f64 - mean * mean).max(0.0); + DurationStats { + min_ns: self.min, + max_ns: self.max, + mean_ns: Some(mean), + stddev_ns: Some(variance.sqrt()), + p50_ns: self.quantiles.quantile(0.50), + p95_ns: self.quantiles.quantile(0.95), + } + } +} + +#[derive(Debug, Default)] +struct WeightedQuantiles { + samples: Vec, + seen: u64, +} + +impl WeightedQuantiles { + fn record(&mut self, value: u64) { + self.seen = self.seen.saturating_add(1); + if self.samples.len() < MAX_QUANTILE_SAMPLES { + self.samples.push(value); + return; + } + let candidate = splitmix64(self.seen) % self.seen; + if candidate < MAX_QUANTILE_SAMPLES as u64 { + self.samples[candidate as usize] = value; + } + } + + fn quantile(&self, quantile: f64) -> Option { + if self.samples.is_empty() { + return None; + } + let mut values = self.samples.clone(); + values.sort_unstable(); + let index = ((values.len().saturating_sub(1)) as f64 * quantile).round() as usize; + values.get(index).copied() + } } #[derive(Clone, Debug, Eq, Hash, PartialEq)] @@ -66,12 +187,6 @@ struct EdgeKey { msg: String, } -#[derive(Clone, Debug, Eq, Hash, PartialEq)] -struct SrcMsgKey { - src: String, - msg: String, -} - #[derive(Clone, Debug)] struct OutputSlot { edges: Vec, @@ -192,9 +307,19 @@ where { let graph = config.get_graph(mission)?; let signature = build_graph_signature(graph, mission); - let output_slots = build_output_slots(graph)?; + let output_slots = build_output_slots::

(graph)?; let mut edge_accumulators = build_edge_accumulators(graph); let mut perf = PerfAccumulator::new(); + let resource_bindings = build_resource_bindings::

(config, graph); + let mut stage_accumulators = BTreeMap::::new(); + let mut trace_duration_accumulator = TimingAccumulator::default(); + let mut residual_accumulator = TimingAccumulator::default(); + let mut representative_traces = Vec::::new(); + let mut worst_trace: Option = None; + let mut active_intervals = Vec::::new(); + let mut overlap_counts = BTreeMap::<(String, String, String), u64>::new(); + let mut previous_end_ns = None; + let mut trace_count = 0u64; let mut warned_lengths = false; for culist in copperlists_reader::

(&mut reader) { @@ -229,6 +354,36 @@ where } perf.record_sample(compute_end_to_end_latency(&cumsgs)); + + if let Some(trace) = + build_execution_trace(culist.id, &cumsgs, P::get_all_task_ids(), previous_end_ns) + { + for interval in &trace.intervals { + stage_accumulators + .entry(interval.origin.clone()) + .or_default() + .record(interval.end_ns - interval.start_ns); + } + trace_duration_accumulator.record(trace.duration_ns); + if let Some(residual) = trace.residual_before_ns { + residual_accumulator.record(residual); + } + record_resource_overlaps( + &trace, + &resource_bindings, + &mut active_intervals, + &mut overlap_counts, + ); + previous_end_ns = trace.intervals.iter().map(|interval| interval.end_ns).max(); + trace_count = trace_count.saturating_add(1); + sample_trace(&mut representative_traces, trace.clone(), trace_count); + if worst_trace + .as_ref() + .is_none_or(|worst| trace.duration_ns > worst.duration_ns) + { + worst_trace = Some(trace); + } + } } let edges = edge_accumulators @@ -236,15 +391,245 @@ where .map(|(key, acc)| acc.finalize(key)) .collect(); + let duration_p50 = trace_duration_accumulator + .stats() + .p50_ns + .unwrap_or_default(); + let typical_trace = representative_traces + .into_iter() + .min_by_key(|trace| trace.duration_ns.abs_diff(duration_p50)); + let mut traces = Vec::new(); + if let Some(mut typical) = typical_trace { + typical.kind = "typical".to_string(); + traces.push(typical); + } + if let Some(mut worst) = worst_trace + && traces + .first() + .is_none_or(|typical| typical.culist_id != worst.culist_id) + { + worst.kind = "slowest".to_string(); + traces.push(worst); + } + + let stages = stage_accumulators + .into_iter() + .map(|(origin, accumulator)| StageLogStats { + origin, + samples: accumulator.samples, + durations: accumulator.stats(), + }) + .collect(); + let resource_overlaps = overlap_counts + .into_iter() + .map( + |((resource, left, right), occurrences)| ResourceOverlapStats { + resource, + left, + right, + occurrences, + }, + ) + .collect(); + Ok(LogStats { schema_version: LOGSTATS_SCHEMA_VERSION, config_signature: signature, mission: mission.map(|value| value.to_string()), edges, perf: perf.finalize(), + schedule: Some(ScheduleLogStats { + stages, + traces, + residual_before: residual_accumulator.stats(), + resource_overlaps, + }), }) } +fn build_execution_trace( + culist_id: u64, + msgs: &[&dyn cu29::prelude::ErasedCuStampedData], + origins: &[&str], + previous_end_ns: Option, +) -> Option { + let mut grouped = BTreeMap::::new(); + for (msg, origin) in msgs.iter().zip(origins.iter()) { + let (Some(start), Some(end)) = ( + extract_start_time_ns(msg.metadata()), + extract_end_time_ns(msg.metadata()), + ) else { + continue; + }; + if end < start { + continue; + } + grouped + .entry((*origin).to_string()) + .and_modify(|range| { + range.0 = range.0.min(start); + range.1 = range.1.max(end); + }) + .or_insert((start, end)); + } + if grouped.is_empty() { + return None; + } + let mut intervals = grouped + .into_iter() + .map(|(origin, (start_ns, end_ns))| ExecutionInterval { + origin, + start_ns, + end_ns, + }) + .collect::>(); + intervals.sort_by_key(|interval| (interval.start_ns, interval.end_ns)); + let original_count = intervals.len(); + let intervals = select_current_execution_cluster(intervals); + let start_ns = intervals.first()?.start_ns; + let end_ns = intervals.iter().map(|interval| interval.end_ns).max()?; + let duration_ns = intervals + .iter() + .map(|interval| interval.end_ns.saturating_sub(interval.start_ns)) + .sum(); + Some(ExecutionTrace { + kind: String::new(), + culist_id, + duration_ns, + wall_span_ns: end_ns.saturating_sub(start_ns), + excluded_intervals: original_count.saturating_sub(intervals.len()) as u32, + residual_before_ns: previous_end_ns.and_then(|end| start_ns.checked_sub(end)), + intervals, + }) +} + +fn select_current_execution_cluster(intervals: Vec) -> Vec { + if intervals.len() < 2 { + return intervals; + } + let total_process_ns: u64 = intervals + .iter() + .map(|interval| interval.end_ns.saturating_sub(interval.start_ns)) + .sum(); + let split_gap_ns = total_process_ns + .saturating_mul(4) + .max(MIN_CURRENT_CLUSTER_GAP_NS); + let mut ranges = Vec::new(); + let mut cluster_start = 0; + for index in 1..intervals.len() { + let gap = intervals[index] + .start_ns + .saturating_sub(intervals[index - 1].end_ns); + if gap > split_gap_ns { + ranges.push(cluster_start..index); + cluster_start = index; + } + } + ranges.push(cluster_start..intervals.len()); + let selected = ranges + .into_iter() + .max_by_key(|range| (range.len(), intervals[range.end - 1].end_ns)) + .unwrap_or(0..intervals.len()); + intervals[selected].to_vec() +} + +fn sample_trace(samples: &mut Vec, trace: ExecutionTrace, seen: u64) { + if samples.len() < MAX_REPRESENTATIVE_TRACES { + samples.push(trace); + return; + } + let hash = splitmix64(trace.culist_id); + let candidate = hash % seen; + if candidate < MAX_REPRESENTATIVE_TRACES as u64 { + samples[candidate as usize] = trace; + } +} + +fn splitmix64(mut value: u64) -> u64 { + value = value.wrapping_add(0x9e3779b97f4a7c15); + value = (value ^ (value >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + value = (value ^ (value >> 27)).wrapping_mul(0x94d049bb133111eb); + value ^ (value >> 31) +} + +fn build_resource_bindings( + config: &CuConfig, + graph: &CuGraph, +) -> HashMap> { + let mut bindings = HashMap::new(); + for origin in P::get_all_task_ids() { + if bindings.contains_key(*origin) { + continue; + } + let mut targets: Vec = if let Some(rest) = origin.strip_prefix("bridge::") { + let bridge_id = rest.split("::").next().unwrap_or_default(); + config + .bridges + .iter() + .find(|bridge| bridge.id == bridge_id) + .and_then(|bridge| bridge.resources.as_ref()) + .map(|resources| resources.values().cloned().collect()) + .unwrap_or_default() + } else { + graph + .get_node_id_by_name(origin) + .and_then(|node_id| graph.get_node(node_id)) + .and_then(|node| node.get_resources()) + .map(|resources| resources.values().cloned().collect()) + .unwrap_or_default() + }; + targets.sort(); + targets.dedup(); + bindings.insert((*origin).to_string(), targets); + } + bindings +} + +fn record_resource_overlaps( + trace: &ExecutionTrace, + bindings: &HashMap>, + active: &mut Vec, + counts: &mut BTreeMap<(String, String, String), u64>, +) { + let Some(trace_start) = trace + .intervals + .iter() + .map(|interval| interval.start_ns) + .min() + else { + return; + }; + active.retain(|interval| interval.end_ns > trace_start); + + for (index, interval) in trace.intervals.iter().enumerate() { + for other in active.iter().chain(trace.intervals[..index].iter()) { + if interval.origin == other.origin + || interval.start_ns >= other.end_ns + || other.start_ns >= interval.end_ns + { + continue; + } + let Some(left_resources) = bindings.get(&interval.origin) else { + continue; + }; + let Some(right_resources) = bindings.get(&other.origin) else { + continue; + }; + for resource in left_resources { + if right_resources.contains(resource) { + let (left, right) = if interval.origin <= other.origin { + (interval.origin.clone(), other.origin.clone()) + } else { + (other.origin.clone(), interval.origin.clone()) + }; + *counts.entry((resource.clone(), left, right)).or_default() += 1; + } + } + } + } + active.extend(trace.intervals.iter().cloned()); +} + pub fn write_logstats(stats: &LogStats, path: &Path) -> CuResult<()> { let file = File::create(path) .map_err(|e| CuError::new_with_cause("Failed to create logstats output", e))?; @@ -253,60 +638,65 @@ pub fn write_logstats(stats: &LogStats, path: &Path) -> CuResult<()> { Ok(()) } -fn build_output_slots(graph: &CuGraph) -> CuResult> { - let packs = collect_output_packs(graph)?; - let edges_by_src = build_edges_by_src_msg(graph); - let total_msgs: usize = packs.iter().map(|pack| pack.msg_types.len()).sum(); - let mut slots = Vec::with_capacity(total_msgs); - - for pack in packs { - for msg in pack.msg_types { - let edges = edges_by_src - .get(&SrcMsgKey { - src: pack.src.clone(), - msg: msg.clone(), - }) - .cloned() - .unwrap_or_default(); - slots.push(OutputSlot { edges }); - } +fn build_output_slots(graph: &CuGraph) -> CuResult> { + let specs = P::get_output_specs(); + if specs.is_empty() { + return build_output_slots_from_plan(graph); } - - Ok(slots) + Ok(specs + .iter() + .map(|spec| OutputSlot { + edges: graph + .edges() + .filter(|edge| edge.msg == spec.msg_type && edge_matches_origin(edge, spec.task_id)) + .map(edge_key_from_connection) + .collect(), + }) + .collect()) } -fn build_edge_accumulators(graph: &CuGraph) -> HashMap { - let mut acc = HashMap::new(); - for cnx in graph.edges() { - let key = EdgeKey { - src: cnx.src.clone(), - src_channel: cnx.src_channel.clone(), - dst: cnx.dst.clone(), - dst_channel: cnx.dst_channel.clone(), - msg: cnx.msg.clone(), - }; - acc.entry(key).or_default(); +fn edge_matches_origin(edge: &cu29::config::Cnx, origin: &str) -> bool { + if edge.src == origin { + return true; } - acc + let Some(rest) = origin.strip_prefix("bridge::") else { + return false; + }; + let mut parts = rest.split("::"); + let (Some(bridge), Some(direction), Some(channel), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return false; + }; + direction == "rx" && edge.src == bridge && edge.src_channel.as_deref() == Some(channel) } -fn build_edges_by_src_msg(graph: &CuGraph) -> HashMap> { - let mut map: HashMap> = HashMap::new(); - for cnx in graph.edges() { - let key = SrcMsgKey { - src: cnx.src.clone(), - msg: cnx.msg.clone(), - }; - let edge = EdgeKey { - src: cnx.src.clone(), - src_channel: cnx.src_channel.clone(), - dst: cnx.dst.clone(), - dst_channel: cnx.dst_channel.clone(), - msg: cnx.msg.clone(), - }; - map.entry(key).or_default().push(edge); +fn edge_key_from_connection(cnx: &cu29::config::Cnx) -> EdgeKey { + EdgeKey { + src: cnx.src.clone(), + src_channel: cnx.src_channel.clone(), + dst: cnx.dst.clone(), + dst_channel: cnx.dst_channel.clone(), + msg: cnx.msg.clone(), } - map +} + +fn build_output_slots_from_plan(graph: &CuGraph) -> CuResult> { + let mut packs = Vec::new(); + collect_output_packs_from_loop(&compute_runtime_plan(graph)?, graph, &mut packs)?; + packs.sort_by_key(|pack| pack.culist_index); + Ok(packs + .into_iter() + .flat_map(|pack| { + pack.msg_types.into_iter().map(move |msg| OutputSlot { + edges: graph + .edges() + .filter(|edge| edge.src == pack.src && edge.msg == msg) + .map(edge_key_from_connection) + .collect(), + }) + }) + .collect()) } #[derive(Debug)] @@ -316,14 +706,6 @@ struct OutputPackInfo { msg_types: Vec, } -fn collect_output_packs(graph: &CuGraph) -> CuResult> { - let plan = compute_runtime_plan(graph)?; - let mut packs = Vec::new(); - collect_output_packs_from_loop(&plan, graph, &mut packs)?; - packs.sort_by_key(|pack| pack.culist_index); - Ok(packs) -} - fn collect_output_packs_from_loop( loop_unit: &CuExecutionLoop, graph: &CuGraph, @@ -351,6 +733,21 @@ fn collect_output_packs_from_loop( Ok(()) } +fn build_edge_accumulators(graph: &CuGraph) -> HashMap { + let mut acc = HashMap::new(); + for cnx in graph.edges() { + let key = EdgeKey { + src: cnx.src.clone(), + src_channel: cnx.src_channel.clone(), + dst: cnx.dst.clone(), + dst_channel: cnx.dst_channel.clone(), + msg: cnx.msg.clone(), + }; + acc.entry(key).or_default(); + } + acc +} + fn compute_end_to_end_latency( msgs: &[&dyn cu29::prelude::ErasedCuStampedData], ) -> Option { @@ -384,6 +781,8 @@ fn duration_stats_from(stats: &CuDurationStatistics) -> DurationStats { max_ns: Some(stats.max().as_nanos()), mean_ns: Some(stats.mean().as_nanos() as f64), stddev_ns: Some(stats.stddev().as_nanos() as f64), + p50_ns: None, + p95_ns: None, } } @@ -396,6 +795,8 @@ fn jitter_stats_from(stats: &CuDurationStatistics) -> DurationStats { max_ns: Some(stats.jitter_max().as_nanos()), mean_ns: Some(stats.jitter_mean().as_nanos() as f64), stddev_ns: Some(stats.jitter_stddev().as_nanos() as f64), + p50_ns: None, + p95_ns: None, } } @@ -510,4 +911,78 @@ mod tests { assert_eq!(stats.end_to_end.max_ns, Some(1_000)); assert_eq!(stats.jitter.min_ns, None); } + + #[test] + fn timing_accumulator_reports_bounded_quantiles() { + let mut timings = TimingAccumulator::default(); + for value in 1..=10_000 { + timings.record(value); + } + let stats = timings.stats(); + assert_eq!(stats.min_ns, Some(1)); + assert_eq!(stats.max_ns, Some(10_000)); + assert!(stats.p50_ns.unwrap().abs_diff(5_000) < 250); + assert!(stats.p95_ns.unwrap().abs_diff(9_500) < 250); + assert!(timings.quantiles.samples.len() <= MAX_QUANTILE_SAMPLES); + } + + #[test] + fn resource_overlap_requires_time_and_declared_target_overlap() { + let trace = ExecutionTrace { + kind: String::new(), + culist_id: 1, + duration_ns: 30, + wall_span_ns: 30, + excluded_intervals: 0, + residual_before_ns: None, + intervals: vec![ + ExecutionInterval { + origin: "left".to_string(), + start_ns: 10, + end_ns: 30, + }, + ExecutionInterval { + origin: "right".to_string(), + start_ns: 20, + end_ns: 40, + }, + ], + }; + let bindings = HashMap::from([ + ("left".to_string(), vec!["gpu0".to_string()]), + ("right".to_string(), vec!["gpu0".to_string()]), + ]); + let mut active = Vec::new(); + let mut counts = BTreeMap::new(); + record_resource_overlaps(&trace, &bindings, &mut active, &mut counts); + assert_eq!( + counts.get(&("gpu0".to_string(), "left".to_string(), "right".to_string())), + Some(&1) + ); + } + + #[test] + fn current_execution_cluster_excludes_carried_forward_slots() { + let intervals = vec![ + ExecutionInterval { + origin: "stale_bridge".to_string(), + start_ns: 100, + end_ns: 200, + }, + ExecutionInterval { + origin: "source".to_string(), + start_ns: 10_000_000, + end_ns: 10_000_500, + }, + ExecutionInterval { + origin: "sink".to_string(), + start_ns: 10_000_600, + end_ns: 10_001_000, + }, + ]; + let selected = select_current_execution_cluster(intervals); + assert_eq!(selected.len(), 2); + assert_eq!(selected[0].origin, "source"); + assert_eq!(selected[1].origin, "sink"); + } } diff --git a/core/cu29_runtime/src/rendercfg.rs b/core/cu29_runtime/src/rendercfg.rs index 7c0c5012b0b..72601b1bffd 100644 --- a/core/cu29_runtime/src/rendercfg.rs +++ b/core/cu29_runtime/src/rendercfg.rs @@ -48,7 +48,7 @@ const MODULE_TRUNC_MARKER: &str = "…"; const MODULE_SEPARATOR: &str = "⠶"; const PLACEHOLDER_TEXT: &str = "\u{2014}"; const COPPER_LOGO_SVG: &str = include_str!("../assets/cu29.svg"); -const LOGSTATS_SCHEMA_VERSION: u32 = 1; +const LOGSTATS_SCHEMA_VERSION: u32 = 2; // Color palette and fills. const BORDER_COLOR: &str = "#999999"; @@ -469,7 +469,7 @@ fn load_logstats( let logstats: LogStats = serde_json::from_str(&contents) .map_err(|e| CuError::new_with_cause("Failed to parse logstats JSON", e))?; - if logstats.schema_version != LOGSTATS_SCHEMA_VERSION { + if !(1..=LOGSTATS_SCHEMA_VERSION).contains(&logstats.schema_version) { eprintln!( "Warning: logstats schema version {} does not match renderer {}", logstats.schema_version, LOGSTATS_SCHEMA_VERSION @@ -4591,14 +4591,10 @@ fn build_graph_signature(config: &config::CuConfig, mission: Option<&str>) -> Cu let mut nodes: Vec<_> = graph.get_all_nodes(); nodes.sort_by_key(|a| a.1.get_id()); - for (node_id, node) in nodes { + for (_, node) in nodes { let flavor = match node.get_flavor() { config::Flavor::Bridge => "bridge", - config::Flavor::Task => match config::resolve_task_kind_for_id(graph, node_id)? { - config::TaskKind::Source => "source", - config::TaskKind::Regular => "task", - config::TaskKind::Sink => "sink", - }, + config::Flavor::Task => "task", }; parts.push(format!( "node|{}|{}|{}",