Add execution order divergence analysis library module (#140)

What This PR Does
- Adds a new execution_order library module that parses graph execution
sequences from PyTorch logs and detects divergences across ranks using
existing collective schedule and cache analysis infrastructure.

Key Components
- parse_graph_execution_order(): Extracts compile_id sequences from
graph_execution JSON artifacts
- analyze_execution_order(): Aligns ranks by execution index and flags
ScheduleMismatch/CacheMismatch issues
- Structured reporting: In-memory ExecOrderReport with per-index
divergence details

Integration
- Reuses existing read_collective_schedules() and cache analysis data


Next Steps
- Wire this analysis into --all-ranks-html landing page to display
"Execution-Order Diagnostics" section when artifacts are present.

Author: skarjala

src/lib.rs

@@ -29,6 +29,11 @@ pub use types::{
     RankMetaData, RuntimeAnalysis, RuntimeRankDetail,
 };
 
+pub use execution_order::{
+    analyze_execution_order, parse_graph_execution_order, ExecOrderIndexRow, ExecOrderIssue,
+    ExecOrderReport,
+};
+
 #[derive(Debug)]
 enum ParserResult {
     NoPayload,
@@ -1795,3 +1800,131 @@ fn convert_node_mappings_to_line_numbers(
         "postToCppCode": hashmap_to_json_map(line_post_to_cpp_code)
     })
 }
+
+pub mod execution_order {
+    use fxhash::FxHashMap;
+
+    /// Issue types detected at a given execution index across ranks
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    pub enum ExecOrderIssue {
+        ScheduleMismatch,
+        CacheMismatch,
+    }
+
+    /// One row in the execution-order report
+    #[derive(Debug, Clone)]
+    pub struct ExecOrderIndexRow {
+        /// Zero-based index into per-rank execution orders
+        pub index: usize,
+        /// Mapping: rank -> compile_id (compile directory name)
+        pub by_rank: FxHashMap<u32, String>,
+        /// Issues found for this index
+        pub issues: Vec<ExecOrderIssue>,
+    }
+
+    /// Final report for execution-order diagnostics
+    #[derive(Debug, Clone, Default)]
+    pub struct ExecOrderReport {
+        pub by_index: Vec<ExecOrderIndexRow>,
+    }
+
+    /// Analyze per-rank execution orders, aligning entries by index and flagging issues
+    /// using provided per-graph properties.
+    pub fn analyze_execution_order(
+        exec_orders: &FxHashMap<u32, Vec<String>>,
+        collective_schedule_by_graph: &FxHashMap<(u32, String), Vec<String>>,
+        cache_status: &FxHashMap<(u32, String), String>,
+    ) -> ExecOrderReport {
+        // Determine max length across ranks (N)
+        let max_len = exec_orders.values().map(|v| v.len()).max().unwrap_or(0);
+
+        if max_len == 0 || exec_orders.is_empty() {
+            return ExecOrderReport::default();
+        }
+
+        // Memoize property lookups per (rank, compile_id)
+        let mut sched_memo: FxHashMap<(u32, String), Option<Vec<String>>> = FxHashMap::default();
+        let mut cache_memo: FxHashMap<(u32, String), Option<String>> = FxHashMap::default();
+
+        let mut rows: Vec<ExecOrderIndexRow> = Vec::with_capacity(max_len);
+
+        for idx in 0..max_len {
+            // Gather present ranks and their compile_ids at this index
+            let by_rank: FxHashMap<u32, String> = exec_orders
+                .iter()
+                .filter_map(|(&rank, order)| order.get(idx).cloned().map(|cid| (rank, cid)))
+                .collect();
+
+            if by_rank.is_empty() {
+                continue;
+            }
+
+            // Evaluate issues among present ranks
+            let mut issues: Vec<ExecOrderIssue> = Vec::new();
+
+            // Schedule mismatch: compare collective op sequences
+            {
+                let schedules: Vec<Vec<String>> = by_rank
+                    .iter()
+                    .filter_map(|(&rank, cid)| {
+                        let key = (rank, cid.clone());
+                        let entry = sched_memo
+                            .entry((rank, cid.clone()))
+                            .or_insert_with(|| collective_schedule_by_graph.get(&key).cloned());
+                        entry.clone()
+                    })
+                    .collect();
+                if schedules.len() >= 2 && schedules[1..].iter().any(|s| s != &schedules[0]) {
+                    issues.push(ExecOrderIssue::ScheduleMismatch);
+                }
+            }
+
+            // Cache skew: compare cache status markers
+            {
+                let statuses: Vec<String> = by_rank
+                    .iter()
+                    .filter_map(|(&rank, cid)| {
+                        let key = (rank, cid.clone());
+                        let entry = cache_memo
+                            .entry((rank, cid.clone()))
+                            .or_insert_with(|| cache_status.get(&key).cloned());
+                        entry.clone().filter(|s| !s.is_empty())
+                    })
+                    .collect();
+                if statuses.len() >= 2 && statuses[1..].iter().any(|s| s != &statuses[0]) {
+                    issues.push(ExecOrderIssue::CacheMismatch);
+                }
+            }
+
+            rows.push(ExecOrderIndexRow {
+                index: idx,
+                by_rank,
+                issues,
+            });
+        }
+
+        ExecOrderReport { by_index: rows }
+    }
+
+    pub fn parse_graph_execution_order(payload: &str) -> anyhow::Result<Vec<String>> {
+        let value: serde_json::Value = serde_json::from_str(payload)?;
+        let arr = value
+            .get("graph_execution_order")
+            .and_then(|v| v.as_array())
+            .ok_or_else(|| anyhow::anyhow!("missing graph_execution_order array"))?;
+
+        let mut out = Vec::with_capacity(arr.len());
+        for item in arr {
+            match item {
+                serde_json::Value::String(s) => out.push(s.clone()),
+                serde_json::Value::Object(map) => {
+                    if let Some(s) = map.get("compile_id").and_then(|v| v.as_str()) {
+                        out.push(s.to_string());
+                    }
+                }
+                _ => {}
+            }
+        }
+        Ok(out)
+    }
+}

tests/integration_test.rs

@@ -2365,3 +2365,74 @@ fn test_tensor_meta_divergence_groups() -> Result<(), Box<dyn std::error::Error>
 
     Ok(())
 }
+
+#[test]
+fn test_execution_order_multi_rank_divergence() -> anyhow::Result<()> {
+    // Test data: Two ranks with different execution orders
+    let exec_orders: fxhash::FxHashMap<u32, Vec<String>> = vec![
+        (
+            0_u32,
+            vec!["0/0".to_string(), "0/1".to_string(), "0/2".to_string()],
+        ),
+        (
+            1_u32,
+            vec!["0/1".to_string(), "0/0".to_string(), "0/2".to_string()],
+        ),
+    ]
+    .into_iter()
+    .collect();
+
+    // Mock collective schedules - different at index 0
+    let collective_schedule_by_graph: fxhash::FxHashMap<(u32, String), Vec<String>> = vec![
+        ((0_u32, "0/0".to_string()), vec!["all_reduce".to_string()]),
+        ((0_u32, "0/1".to_string()), vec!["all_gather".to_string()]),
+        ((1_u32, "0/0".to_string()), vec!["all_reduce".to_string()]),
+        ((1_u32, "0/1".to_string()), vec!["broadcast".to_string()]), // Different!
+    ]
+    .into_iter()
+    .collect();
+
+    // Mock cache status - different at index 1
+    let cache_status: fxhash::FxHashMap<(u32, String), String> = vec![
+        ((0_u32, "0/0".to_string()), "✅".to_string()),
+        ((0_u32, "0/1".to_string()), "❌".to_string()),
+        ((1_u32, "0/0".to_string()), "✅".to_string()),
+        ((1_u32, "0/1".to_string()), "✅".to_string()), // Different!
+    ]
+    .into_iter()
+    .collect();
+
+    let report = tlparse::execution_order::analyze_execution_order(
+        &exec_orders,
+        &collective_schedule_by_graph,
+        &cache_status,
+    );
+
+    assert_eq!(report.by_index.len(), 3);
+
+    // Index 0: Rank 0 has "0/0", Rank 1 has "0/1"
+    // Should detect collective schedule mismatch
+    let row_0 = &report.by_index[0];
+    assert_eq!(row_0.by_rank.get(&0_u32), Some(&"0/0".to_string()));
+    assert_eq!(row_0.by_rank.get(&1_u32), Some(&"0/1".to_string()));
+    assert!(row_0
+        .issues
+        .contains(&tlparse::execution_order::ExecOrderIssue::ScheduleMismatch));
+
+    // Index 1: Rank 0 has "0/1", Rank 1 has "0/0"
+    // Should detect cache mismatch
+    let row_1 = &report.by_index[1];
+    assert_eq!(row_1.by_rank.get(&0_u32), Some(&"0/1".to_string()));
+    assert_eq!(row_1.by_rank.get(&1_u32), Some(&"0/0".to_string()));
+    assert!(row_1
+        .issues
+        .contains(&tlparse::execution_order::ExecOrderIssue::CacheMismatch));
+
+    // Index 2: Both ranks have "0/2" - no issues
+    let row_2 = &report.by_index[2];
+    assert_eq!(row_2.by_rank.get(&0_u32), Some(&"0/2".to_string()));
+    assert_eq!(row_2.by_rank.get(&1_u32), Some(&"0/2".to_string()));
+    assert!(row_2.issues.is_empty());
+
+    Ok(())
+}