use std::cmp::Reverse;
use std::collections::HashMap;
use std::fmt;
use std::time::{Duration, Instant};
#[derive(Debug, Clone)]
pub struct TraceEntry {
pub entry_id: usize,
pub node_id: usize,
pub operation: String,
pub start_time: Instant,
pub duration: Duration,
pub input_ids: Vec<usize>,
pub output_ids: Vec<usize>,
pub metadata: HashMap<String, String>,
}
impl TraceEntry {
pub fn duration_ms(&self) -> f64 {
self.duration.as_secs_f64() * 1000.0
}
pub fn duration_us(&self) -> f64 {
self.duration.as_secs_f64() * 1_000_000.0
}
}
#[derive(Debug, Clone)]
pub struct ExecutionTrace {
entries: Vec<TraceEntry>,
total_duration: Duration,
graph_id: Option<usize>,
}
impl ExecutionTrace {
pub fn new() -> Self {
Self {
entries: Vec::new(),
total_duration: Duration::ZERO,
graph_id: None,
}
}
pub fn with_graph_id(mut self, graph_id: usize) -> Self {
self.graph_id = Some(graph_id);
self
}
pub fn add_entry(&mut self, entry: TraceEntry) {
self.total_duration += entry.duration;
self.entries.push(entry);
}
pub fn entries(&self) -> &[TraceEntry] {
&self.entries
}
pub fn total_duration(&self) -> Duration {
self.total_duration
}
pub fn total_duration_ms(&self) -> f64 {
self.total_duration.as_secs_f64() * 1000.0
}
pub fn entries_for_node(&self, node_id: usize) -> Vec<&TraceEntry> {
self.entries
.iter()
.filter(|e| e.node_id == node_id)
.collect()
}
pub fn critical_path(&self) -> Vec<&TraceEntry> {
if self.entries.is_empty() {
return Vec::new();
}
let n = self.entries.len();
let mut tensor_producers: HashMap<usize, usize> = HashMap::new();
for (idx, entry) in self.entries.iter().enumerate() {
for &output_id in &entry.output_ids {
tensor_producers.insert(output_id, idx);
}
}
let mut predecessors: Vec<Vec<usize>> = vec![Vec::new(); n];
for (idx, entry) in self.entries.iter().enumerate() {
for &input_id in &entry.input_ids {
if let Some(&producer_idx) = tensor_producers.get(&input_id) {
if producer_idx < n {
predecessors[idx].push(producer_idx);
}
}
}
}
let mut eft = vec![Duration::ZERO; n];
let mut predecessor_on_critical_path = vec![None; n];
let mut changed = true;
for _ in 0..n {
if !changed {
break;
}
changed = false;
for idx in 0..n {
let mut max_pred_eft = Duration::ZERO;
let mut critical_pred = None;
for &pred_idx in &predecessors[idx] {
if eft[pred_idx] > max_pred_eft {
max_pred_eft = eft[pred_idx];
critical_pred = Some(pred_idx);
}
}
let new_eft = max_pred_eft + self.entries[idx].duration;
if new_eft > eft[idx] {
eft[idx] = new_eft;
predecessor_on_critical_path[idx] = critical_pred;
changed = true;
}
}
}
let critical_end_idx = eft
.iter()
.enumerate()
.max_by_key(|(_, &time)| time)
.map(|(idx, _)| idx)
.unwrap_or(0);
let mut critical_path_indices = Vec::new();
let mut current = Some(critical_end_idx);
while let Some(idx) = current {
critical_path_indices.push(idx);
current = predecessor_on_critical_path[idx];
}
critical_path_indices.reverse();
critical_path_indices
.iter()
.map(|&idx| &self.entries[idx])
.collect()
}
pub fn critical_path_duration(&self) -> Duration {
self.critical_path().iter().map(|e| e.duration).sum()
}
pub fn parallelism_factor(&self) -> f64 {
let critical_time = self.critical_path_duration();
if critical_time.as_secs_f64() == 0.0 {
return 1.0;
}
self.total_duration.as_secs_f64() / critical_time.as_secs_f64()
}
pub fn slowest_operations(&self, limit: usize) -> Vec<&TraceEntry> {
let mut sorted: Vec<_> = self.entries.iter().collect();
sorted.sort_by_key(|b| Reverse(b.duration));
sorted.into_iter().take(limit).collect()
}
pub fn summary(&self) -> TraceSummary {
TraceSummary::from_trace(self)
}
}
impl Default for ExecutionTrace {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct TraceSummary {
pub total_operations: usize,
pub total_time_ms: f64,
pub avg_time_ms: f64,
pub max_time_ms: f64,
pub min_time_ms: f64,
pub operation_counts: HashMap<String, usize>,
}
impl TraceSummary {
pub fn from_trace(trace: &ExecutionTrace) -> Self {
let entries = trace.entries();
let total_operations = entries.len();
let total_time_ms = trace.total_duration_ms();
let avg_time_ms = if total_operations > 0 {
total_time_ms / total_operations as f64
} else {
0.0
};
let max_time_ms = entries.iter().map(|e| e.duration_ms()).fold(0.0, f64::max);
let min_time_ms = entries
.iter()
.map(|e| e.duration_ms())
.fold(f64::MAX, f64::min);
let mut operation_counts: HashMap<String, usize> = HashMap::new();
for entry in entries {
*operation_counts.entry(entry.operation.clone()).or_insert(0) += 1;
}
Self {
total_operations,
total_time_ms,
avg_time_ms,
max_time_ms,
min_time_ms,
operation_counts,
}
}
}
impl fmt::Display for TraceSummary {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(f, "Execution Trace Summary")?;
writeln!(f, "=======================")?;
writeln!(f, "Total operations: {}", self.total_operations)?;
writeln!(f, "Total time: {:.2} ms", self.total_time_ms)?;
writeln!(f, "Average time: {:.2} ms", self.avg_time_ms)?;
writeln!(f, "Max time: {:.2} ms", self.max_time_ms)?;
writeln!(f, "Min time: {:.2} ms", self.min_time_ms)?;
writeln!(f, "\nOperation Counts:")?;
let mut sorted_ops: Vec<_> = self.operation_counts.iter().collect();
sorted_ops.sort_by_key(|(_, count)| std::cmp::Reverse(**count));
for (op, count) in sorted_ops {
writeln!(f, " {}: {}", op, count)?;
}
Ok(())
}
}
pub struct ExecutionTracer {
enabled: bool,
current_trace: ExecutionTrace,
traces: Vec<ExecutionTrace>,
next_entry_id: usize,
}
impl ExecutionTracer {
pub fn new() -> Self {
Self {
enabled: false,
current_trace: ExecutionTrace::new(),
traces: Vec::new(),
next_entry_id: 0,
}
}
pub fn enable(&mut self) {
self.enabled = true;
}
pub fn disable(&mut self) {
self.enabled = false;
}
pub fn is_enabled(&self) -> bool {
self.enabled
}
pub fn start_trace(&mut self, graph_id: Option<usize>) {
if !self.current_trace.entries.is_empty() {
self.finalize_trace();
}
self.current_trace = ExecutionTrace::new();
if let Some(id) = graph_id {
self.current_trace.graph_id = Some(id);
}
}
pub fn finalize_trace(&mut self) {
if !self.current_trace.entries.is_empty() {
let trace = std::mem::take(&mut self.current_trace);
self.traces.push(trace);
}
}
pub fn record_operation_start(
&mut self,
_node_id: usize,
_operation: impl Into<String>,
_input_ids: Vec<usize>,
) -> OperationHandle {
if !self.enabled {
return OperationHandle {
entry_id: None,
start_time: Instant::now(),
};
}
let entry_id = self.next_entry_id;
self.next_entry_id += 1;
OperationHandle {
entry_id: Some(entry_id),
start_time: Instant::now(),
}
}
pub fn record_operation_end(
&mut self,
handle: OperationHandle,
node_id: usize,
operation: impl Into<String>,
input_ids: Vec<usize>,
output_ids: Vec<usize>,
metadata: HashMap<String, String>,
) {
if !self.enabled || handle.entry_id.is_none() {
return;
}
let duration = handle.start_time.elapsed();
let entry = TraceEntry {
entry_id: handle
.entry_id
.expect("entry_id is Some after is_none guard above"),
node_id,
operation: operation.into(),
start_time: handle.start_time,
duration,
input_ids,
output_ids,
metadata,
};
self.current_trace.add_entry(entry);
}
pub fn get_trace(&self) -> &ExecutionTrace {
&self.current_trace
}
pub fn get_all_traces(&self) -> &[ExecutionTrace] {
&self.traces
}
pub fn clear(&mut self) {
self.current_trace = ExecutionTrace::new();
self.traces.clear();
self.next_entry_id = 0;
}
}
impl Default for ExecutionTracer {
fn default() -> Self {
Self::new()
}
}
pub struct OperationHandle {
entry_id: Option<usize>,
start_time: Instant,
}
#[derive(Debug, Clone)]
pub struct TensorStats {
pub tensor_id: usize,
pub shape: Vec<usize>,
pub num_elements: usize,
pub dtype: String,
pub min_value: Option<f64>,
pub max_value: Option<f64>,
pub mean_value: Option<f64>,
pub std_dev: Option<f64>,
pub num_nans: Option<usize>,
pub num_infs: Option<usize>,
}
impl TensorStats {
pub fn new(tensor_id: usize, shape: Vec<usize>, dtype: impl Into<String>) -> Self {
let num_elements = shape.iter().product();
Self {
tensor_id,
shape,
num_elements,
dtype: dtype.into(),
min_value: None,
max_value: None,
mean_value: None,
std_dev: None,
num_nans: None,
num_infs: None,
}
}
pub fn with_statistics(
mut self,
min: f64,
max: f64,
mean: f64,
std_dev: f64,
num_nans: usize,
num_infs: usize,
) -> Self {
self.min_value = Some(min);
self.max_value = Some(max);
self.mean_value = Some(mean);
self.std_dev = Some(std_dev);
self.num_nans = Some(num_nans);
self.num_infs = Some(num_infs);
self
}
pub fn has_numerical_issues(&self) -> bool {
self.num_nans.unwrap_or(0) > 0 || self.num_infs.unwrap_or(0) > 0
}
}
impl fmt::Display for TensorStats {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(f, "Tensor {} Stats:", self.tensor_id)?;
writeln!(f, " Shape: {:?}", self.shape)?;
writeln!(f, " Elements: {}", self.num_elements)?;
writeln!(f, " DType: {}", self.dtype)?;
if let Some(min) = self.min_value {
writeln!(f, " Min: {:.6}", min)?;
}
if let Some(max) = self.max_value {
writeln!(f, " Max: {:.6}", max)?;
}
if let Some(mean) = self.mean_value {
writeln!(f, " Mean: {:.6}", mean)?;
}
if let Some(std) = self.std_dev {
writeln!(f, " Std Dev: {:.6}", std)?;
}
if let Some(nans) = self.num_nans {
if nans > 0 {
writeln!(f, " ⚠️ NaNs: {}", nans)?;
}
}
if let Some(infs) = self.num_infs {
if infs > 0 {
writeln!(f, " ⚠️ Infs: {}", infs)?;
}
}
Ok(())
}
}
pub struct TensorInspector {
enabled: bool,
tensor_stats: HashMap<usize, TensorStats>,
watch_list: Vec<usize>,
}
impl TensorInspector {
pub fn new() -> Self {
Self {
enabled: false,
tensor_stats: HashMap::new(),
watch_list: Vec::new(),
}
}
pub fn enable(&mut self) {
self.enabled = true;
}
pub fn disable(&mut self) {
self.enabled = false;
}
pub fn is_enabled(&self) -> bool {
self.enabled
}
pub fn watch(&mut self, tensor_id: usize) {
if !self.watch_list.contains(&tensor_id) {
self.watch_list.push(tensor_id);
}
}
pub fn unwatch(&mut self, tensor_id: usize) {
self.watch_list.retain(|&id| id != tensor_id);
}
pub fn clear_watch_list(&mut self) {
self.watch_list.clear();
}
pub fn should_inspect(&self, tensor_id: usize) -> bool {
self.enabled && (self.watch_list.is_empty() || self.watch_list.contains(&tensor_id))
}
pub fn record_stats(&mut self, stats: TensorStats) {
if !self.enabled {
return;
}
self.tensor_stats.insert(stats.tensor_id, stats);
}
pub fn get_stats(&self, tensor_id: usize) -> Option<&TensorStats> {
self.tensor_stats.get(&tensor_id)
}
pub fn get_all_stats(&self) -> &HashMap<usize, TensorStats> {
&self.tensor_stats
}
pub fn find_problematic_tensors(&self) -> Vec<&TensorStats> {
self.tensor_stats
.values()
.filter(|stats| stats.has_numerical_issues())
.collect()
}
pub fn clear(&mut self) {
self.tensor_stats.clear();
}
}
impl Default for TensorInspector {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Breakpoint {
Node(usize),
Operation(String),
NumericalIssue,
TimeThreshold(u64),
Conditional(String), }
#[derive(Debug, Clone)]
pub struct BreakpointHit {
pub breakpoint: Breakpoint,
pub node_id: usize,
pub elapsed_us: u64,
pub context: HashMap<String, String>,
}
pub struct BreakpointManager {
enabled: bool,
breakpoints: Vec<Breakpoint>,
hits: Vec<BreakpointHit>,
continue_execution: bool,
}
impl BreakpointManager {
pub fn new() -> Self {
Self {
enabled: false,
breakpoints: Vec::new(),
hits: Vec::new(),
continue_execution: true,
}
}
pub fn enable(&mut self) {
self.enabled = true;
}
pub fn disable(&mut self) {
self.enabled = false;
}
pub fn is_enabled(&self) -> bool {
self.enabled
}
pub fn add_node_breakpoint(&mut self, node_id: usize) {
self.breakpoints.push(Breakpoint::Node(node_id));
}
pub fn add_operation_breakpoint(&mut self, operation: impl Into<String>) {
self.breakpoints
.push(Breakpoint::Operation(operation.into()));
}
pub fn add_numerical_issue_breakpoint(&mut self) {
self.breakpoints.push(Breakpoint::NumericalIssue);
}
pub fn add_time_threshold_breakpoint(&mut self, threshold_us: u64) {
self.breakpoints
.push(Breakpoint::TimeThreshold(threshold_us));
}
pub fn remove_breakpoint(&mut self, breakpoint: &Breakpoint) {
self.breakpoints.retain(|bp| bp != breakpoint);
}
pub fn clear_breakpoints(&mut self) {
self.breakpoints.clear();
}
pub fn get_breakpoints(&self) -> &[Breakpoint] {
&self.breakpoints
}
pub fn should_break(
&mut self,
node_id: usize,
operation: &str,
elapsed_us: u64,
has_numerical_issue: bool,
) -> Option<BreakpointHit> {
if !self.enabled || !self.continue_execution {
return None;
}
for breakpoint in &self.breakpoints {
let should_break = match breakpoint {
Breakpoint::Node(bp_node_id) => *bp_node_id == node_id,
Breakpoint::Operation(bp_op) => bp_op == operation,
Breakpoint::NumericalIssue => has_numerical_issue,
Breakpoint::TimeThreshold(threshold) => elapsed_us > *threshold,
Breakpoint::Conditional(_) => false, };
if should_break {
let hit = BreakpointHit {
breakpoint: breakpoint.clone(),
node_id,
elapsed_us,
context: HashMap::new(),
};
self.hits.push(hit.clone());
self.continue_execution = false;
return Some(hit);
}
}
None
}
pub fn continue_execution(&mut self) {
self.continue_execution = true;
}
pub fn get_hits(&self) -> &[BreakpointHit] {
&self.hits
}
pub fn clear_hits(&mut self) {
self.hits.clear();
}
}
impl Default for BreakpointManager {
fn default() -> Self {
Self::new()
}
}
pub struct ExecutionRecorder {
enabled: bool,
tracer: ExecutionTracer,
inspector: TensorInspector,
breakpoints: BreakpointManager,
}
impl ExecutionRecorder {
pub fn new() -> Self {
Self {
enabled: false,
tracer: ExecutionTracer::new(),
inspector: TensorInspector::new(),
breakpoints: BreakpointManager::new(),
}
}
pub fn enable(&mut self) {
self.enabled = true;
self.tracer.enable();
self.inspector.enable();
self.breakpoints.enable();
}
pub fn disable(&mut self) {
self.enabled = false;
self.tracer.disable();
self.inspector.disable();
self.breakpoints.disable();
}
pub fn tracer(&mut self) -> &mut ExecutionTracer {
&mut self.tracer
}
pub fn inspector(&mut self) -> &mut TensorInspector {
&mut self.inspector
}
pub fn breakpoints(&mut self) -> &mut BreakpointManager {
&mut self.breakpoints
}
pub fn clear(&mut self) {
self.tracer.clear();
self.inspector.clear();
self.breakpoints.clear_hits();
}
pub fn generate_report(&self) -> ExecutionReport {
ExecutionReport {
trace_summary: self.tracer.get_trace().summary(),
problematic_tensors: self.inspector.find_problematic_tensors().len(),
breakpoint_hits: self.breakpoints.get_hits().len(),
}
}
}
impl Default for ExecutionRecorder {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct ExecutionReport {
pub trace_summary: TraceSummary,
pub problematic_tensors: usize,
pub breakpoint_hits: usize,
}
impl fmt::Display for ExecutionReport {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(f, "{}", self.trace_summary)?;
writeln!(f, "\nDebug Information:")?;
writeln!(f, " Problematic tensors: {}", self.problematic_tensors)?;
writeln!(f, " Breakpoint hits: {}", self.breakpoint_hits)?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_execution_tracer() {
let mut tracer = ExecutionTracer::new();
assert!(!tracer.is_enabled());
tracer.enable();
assert!(tracer.is_enabled());
tracer.start_trace(Some(1));
let handle = tracer.record_operation_start(0, "einsum", vec![0, 1]);
std::thread::sleep(Duration::from_millis(10));
tracer.record_operation_end(handle, 0, "einsum", vec![0, 1], vec![2], HashMap::new());
let trace = tracer.get_trace();
assert_eq!(trace.entries().len(), 1);
assert!(trace.total_duration_ms() >= 10.0);
}
#[test]
fn test_trace_summary() {
let mut trace = ExecutionTrace::new();
let entry = TraceEntry {
entry_id: 0,
node_id: 0,
operation: "einsum".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![0],
output_ids: vec![1],
metadata: HashMap::new(),
};
trace.add_entry(entry);
let summary = trace.summary();
assert_eq!(summary.total_operations, 1);
assert!(summary.total_time_ms >= 10.0);
}
#[test]
fn test_tensor_inspector() {
let mut inspector = TensorInspector::new();
inspector.enable();
let stats =
TensorStats::new(0, vec![2, 3], "f64").with_statistics(0.0, 1.0, 0.5, 0.25, 0, 0);
inspector.record_stats(stats.clone());
assert_eq!(inspector.get_stats(0).expect("unwrap").tensor_id, 0);
assert!(!stats.has_numerical_issues());
}
#[test]
fn test_tensor_numerical_issues() {
let stats = TensorStats::new(0, vec![2, 3], "f64").with_statistics(
0.0,
f64::INFINITY,
0.5,
0.25,
1,
1,
);
assert!(stats.has_numerical_issues());
}
#[test]
fn test_breakpoint_manager() {
let mut manager = BreakpointManager::new();
manager.enable();
manager.add_node_breakpoint(5);
let hit = manager.should_break(5, "einsum", 1000, false);
assert!(hit.is_some());
assert_eq!(hit.expect("unwrap").node_id, 5);
manager.continue_execution();
let hit2 = manager.should_break(5, "einsum", 1000, false);
assert!(hit2.is_some());
}
#[test]
fn test_operation_breakpoint() {
let mut manager = BreakpointManager::new();
manager.enable();
manager.add_operation_breakpoint("matmul");
let hit = manager.should_break(1, "matmul", 1000, false);
assert!(hit.is_some());
let no_hit = manager.should_break(2, "add", 1000, false);
assert!(no_hit.is_none());
}
#[test]
fn test_time_threshold_breakpoint() {
let mut manager = BreakpointManager::new();
manager.enable();
manager.add_time_threshold_breakpoint(5000);
let no_hit = manager.should_break(1, "op", 4000, false);
assert!(no_hit.is_none());
let hit = manager.should_break(1, "op", 6000, false);
assert!(hit.is_some());
}
#[test]
fn test_numerical_issue_breakpoint() {
let mut manager = BreakpointManager::new();
manager.enable();
manager.add_numerical_issue_breakpoint();
let no_hit = manager.should_break(1, "op", 1000, false);
assert!(no_hit.is_none());
let hit = manager.should_break(1, "op", 1000, true);
assert!(hit.is_some());
}
#[test]
fn test_execution_recorder() {
let mut recorder = ExecutionRecorder::new();
recorder.enable();
assert!(recorder.tracer().is_enabled());
assert!(recorder.inspector().is_enabled());
assert!(recorder.breakpoints().is_enabled());
recorder.clear();
let report = recorder.generate_report();
assert_eq!(report.trace_summary.total_operations, 0);
}
#[test]
fn test_slowest_operations() {
let mut trace = ExecutionTrace::new();
for i in 0..5 {
let entry = TraceEntry {
entry_id: i,
node_id: i,
operation: format!("op{}", i),
start_time: Instant::now(),
duration: Duration::from_millis((i as u64 + 1) * 10),
input_ids: vec![],
output_ids: vec![],
metadata: HashMap::new(),
};
trace.add_entry(entry);
}
let slowest = trace.slowest_operations(3);
assert_eq!(slowest.len(), 3);
assert_eq!(slowest[0].node_id, 4); assert_eq!(slowest[1].node_id, 3);
assert_eq!(slowest[2].node_id, 2);
}
#[test]
fn test_watch_list() {
let mut inspector = TensorInspector::new();
inspector.enable();
inspector.watch(1);
inspector.watch(2);
assert!(inspector.should_inspect(1));
assert!(inspector.should_inspect(2));
assert!(!inspector.should_inspect(3));
inspector.unwatch(1);
assert!(!inspector.should_inspect(1));
assert!(inspector.should_inspect(2));
inspector.clear_watch_list();
assert!(inspector.should_inspect(5));
}
#[test]
fn test_trace_entries_for_node() {
let mut trace = ExecutionTrace::new();
for i in 0..3 {
let entry = TraceEntry {
entry_id: i,
node_id: i % 2,
operation: "op".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![],
output_ids: vec![],
metadata: HashMap::new(),
};
trace.add_entry(entry);
}
let node_0_entries = trace.entries_for_node(0);
assert_eq!(node_0_entries.len(), 2);
let node_1_entries = trace.entries_for_node(1);
assert_eq!(node_1_entries.len(), 1);
}
#[test]
fn test_critical_path_linear_chain() {
let mut trace = ExecutionTrace::new();
trace.add_entry(TraceEntry {
entry_id: 0,
node_id: 0,
operation: "op0".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![],
output_ids: vec![0],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 1,
node_id: 1,
operation: "op1".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(20),
input_ids: vec![0],
output_ids: vec![1],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 2,
node_id: 2,
operation: "op2".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(15),
input_ids: vec![1],
output_ids: vec![2],
metadata: HashMap::new(),
});
let critical_path = trace.critical_path();
assert_eq!(critical_path.len(), 3); assert_eq!(critical_path[0].node_id, 0);
assert_eq!(critical_path[1].node_id, 1);
assert_eq!(critical_path[2].node_id, 2);
let cp_duration = trace.critical_path_duration();
assert_eq!(cp_duration, Duration::from_millis(45)); }
#[test]
fn test_critical_path_parallel_operations() {
let mut trace = ExecutionTrace::new();
trace.add_entry(TraceEntry {
entry_id: 0,
node_id: 0,
operation: "op0".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![],
output_ids: vec![0, 1],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 1,
node_id: 1,
operation: "op1".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(5),
input_ids: vec![0],
output_ids: vec![2],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 2,
node_id: 2,
operation: "op2".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(20),
input_ids: vec![1],
output_ids: vec![3],
metadata: HashMap::new(),
});
let critical_path = trace.critical_path();
assert_eq!(critical_path.len(), 2);
assert_eq!(critical_path[0].node_id, 0);
assert_eq!(critical_path[1].node_id, 2);
let cp_duration = trace.critical_path_duration();
assert_eq!(cp_duration, Duration::from_millis(30)); }
#[test]
fn test_critical_path_diamond_pattern() {
let mut trace = ExecutionTrace::new();
trace.add_entry(TraceEntry {
entry_id: 0,
node_id: 0,
operation: "op0".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![],
output_ids: vec![0],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 1,
node_id: 1,
operation: "op1".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(5),
input_ids: vec![0],
output_ids: vec![1],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 2,
node_id: 2,
operation: "op2".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(25),
input_ids: vec![0],
output_ids: vec![2],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 3,
node_id: 3,
operation: "op3".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(15),
input_ids: vec![1, 2],
output_ids: vec![3],
metadata: HashMap::new(),
});
let critical_path = trace.critical_path();
assert_eq!(critical_path.len(), 3);
assert_eq!(critical_path[0].node_id, 0);
assert_eq!(critical_path[1].node_id, 2);
assert_eq!(critical_path[2].node_id, 3);
let cp_duration = trace.critical_path_duration();
assert_eq!(cp_duration, Duration::from_millis(50)); }
#[test]
fn test_critical_path_empty() {
let trace = ExecutionTrace::new();
let critical_path = trace.critical_path();
assert_eq!(critical_path.len(), 0);
assert_eq!(trace.critical_path_duration(), Duration::ZERO);
}
#[test]
fn test_critical_path_single_operation() {
let mut trace = ExecutionTrace::new();
trace.add_entry(TraceEntry {
entry_id: 0,
node_id: 0,
operation: "op0".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![],
output_ids: vec![0],
metadata: HashMap::new(),
});
let critical_path = trace.critical_path();
assert_eq!(critical_path.len(), 1);
assert_eq!(critical_path[0].node_id, 0);
}
#[test]
fn test_parallelism_factor() {
let mut trace = ExecutionTrace::new();
trace.add_entry(TraceEntry {
entry_id: 0,
node_id: 0,
operation: "op0".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(10),
input_ids: vec![],
output_ids: vec![0],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 1,
node_id: 1,
operation: "op1".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(20),
input_ids: vec![0],
output_ids: vec![1],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 2,
node_id: 2,
operation: "op2".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(30),
input_ids: vec![0],
output_ids: vec![2],
metadata: HashMap::new(),
});
trace.add_entry(TraceEntry {
entry_id: 3,
node_id: 3,
operation: "op3".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(40),
input_ids: vec![1, 2],
output_ids: vec![3],
metadata: HashMap::new(),
});
let parallelism = trace.parallelism_factor();
assert!((parallelism - 1.25).abs() < 0.01);
}
#[test]
fn test_critical_path_complex_graph() {
let mut trace = ExecutionTrace::new();
trace.add_entry(TraceEntry {
entry_id: 0,
node_id: 0,
operation: "root".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(5),
input_ids: vec![],
output_ids: vec![0],
metadata: HashMap::new(),
});
for i in 1..=3 {
trace.add_entry(TraceEntry {
entry_id: i,
node_id: i,
operation: format!("branch{}", i),
start_time: Instant::now(),
duration: Duration::from_millis((i as u64) * 10),
input_ids: vec![0],
output_ids: vec![i],
metadata: HashMap::new(),
});
}
trace.add_entry(TraceEntry {
entry_id: 4,
node_id: 4,
operation: "merge".to_string(),
start_time: Instant::now(),
duration: Duration::from_millis(15),
input_ids: vec![2, 3],
output_ids: vec![4],
metadata: HashMap::new(),
});
let critical_path = trace.critical_path();
assert!(critical_path.len() >= 3);
assert_eq!(critical_path[0].operation, "root");
assert_eq!(critical_path[critical_path.len() - 1].operation, "merge");
}
}