Add execution timeouts for simulations

Author: jbrodovsky

book/RBPF.md

@@ -460,7 +460,7 @@ fn test_rbpf_closed_loop_on_real_data() {
     let stream = build_event_stream(&records, &cfg);
 
     // Run closed-loop
-    let results = run_closed_loop(&mut rbpf, stream, None)
+    let results = run_closed_loop(&mut rbpf, stream, None, None)
         .expect("RBPF closed-loop should complete");
 
     // Compute error metrics

core/src/sim.rs

@@ -44,6 +44,7 @@ use log::{debug, info, warn};
 use std::fmt::{Debug, Display};
 use std::io::{self, Read, Write};
 use std::path::Path;
+use std::time::{Duration as StdDuration, Instant};
 
 use anyhow::{Result, bail};
 use chrono::{DateTime, Duration, Utc};
@@ -60,9 +61,11 @@ use crate::messages::{Event, EventStream, GnssFaultModel, GnssScheduler};
 
 use crate::{IMUData, StrapdownState, forward};
 use health::HealthMonitor;
+use execution::ExecutionMonitor;
 
 // Re-export HealthLimits for easier access in tests and external users
 pub use health::HealthLimits;
+pub use execution::{ExecutionLimits, ExecutionMonitor};
 
 pub const DEFAULT_PROCESS_NOISE: [f64; 15] = [
     // Default process noise if not provided
@@ -83,6 +86,10 @@ pub const DEFAULT_PROCESS_NOISE: [f64; 15] = [
     1e-8, // gyro bias z noise
 ];
 
+pub const DEFAULT_MAX_WALL_CLOCK_RATIO: f64 = 0.25;
+pub const DEFAULT_MAX_WALL_CLOCK_S: f64 = 1200.0;
+pub const DEFAULT_MAX_NO_PROGRESS_S: f64 = 600.0;
+
 fn de_f64_nan<'de, D>(deserializer: D) -> Result<f64, D::Error>
 where
     D: Deserializer<'de>,
@@ -1903,19 +1910,31 @@ pub fn dead_reckoning(records: &[TestDataRecord]) -> Vec<NavigationResult> {
 /// * `filter` - Mutable reference to a type implementing NavigationFilter
 /// * `stream` - Event stream containing IMU and measurement events
 /// * `health_limits` - Optional health limits for monitoring
+/// * `execution_limits` - Optional wall-clock and no-progress limits
 ///
 /// # Returns
 /// * `Vec<NavigationResult>` - A vector of navigation results
 pub fn run_closed_loop<F: NavigationFilter>(
     filter: &mut F,
     stream: EventStream,
     health_limits: Option<HealthLimits>,
+    execution_limits: Option<ExecutionLimits>,
 ) -> anyhow::Result<Vec<NavigationResult>> {
     let start_time = stream.start_time;
     let mut results: Vec<NavigationResult> = Vec::with_capacity(stream.events.len());
     let total = stream.events.len();
     let mut last_ts: Option<DateTime<Utc>> = None;
     let mut monitor = HealthMonitor::new(health_limits.unwrap_or_default());
+    let sim_duration_s = stream
+        .events
+        .last()
+        .map(|event| match event {
+            Event::Imu { elapsed_s, .. } => *elapsed_s,
+            Event::Measurement { elapsed_s, .. } => *elapsed_s,
+        })
+        .unwrap_or(0.0);
+    let mut execution_monitor =
+        execution_limits.map(|limits| ExecutionMonitor::new(limits, sim_duration_s));
 
     info!(
         "Starting closed-loop navigation filter with {} events",
@@ -2005,6 +2024,10 @@ pub fn run_closed_loop<F: NavigationFilter>(
             debug!("Filter state at {}: {:?}", ts, mean);
             results.push(NavigationResult::from((&ts, &mean, &cov)));
         }
+
+        if let Some(ref mut monitor) = execution_monitor {
+            monitor.check("closed-loop")?;
+        }
     }
     debug!("Closed-loop simulation complete");
     Ok(results)
@@ -2525,6 +2548,125 @@ pub fn print_sim_status<F: NavigationFilter>(filter: &F) {
     );
 }
 
+pub mod execution {
+    use super::*;
+
+    #[derive(Clone, Debug, Serialize, Deserialize)]
+    pub struct ExecutionLimits {
+        /// Max wall-clock time as a ratio of simulated duration (<= 0 disables).
+        #[serde(default = "default_max_wall_clock_ratio")]
+        pub max_wall_clock_ratio: f64,
+        /// Max wall-clock time per trajectory in seconds (<= 0 disables).
+        #[serde(default = "default_max_wall_clock_s")]
+        pub max_wall_clock_s: f64,
+        /// Max wall-clock time without progress in seconds (<= 0 disables).
+        #[serde(default = "default_max_no_progress_s")]
+        pub max_no_progress_s: f64,
+    }
+
+    fn default_max_wall_clock_ratio() -> f64 {
+        DEFAULT_MAX_WALL_CLOCK_RATIO
+    }
+
+    fn default_max_wall_clock_s() -> f64 {
+        DEFAULT_MAX_WALL_CLOCK_S
+    }
+
+    fn default_max_no_progress_s() -> f64 {
+        DEFAULT_MAX_NO_PROGRESS_S
+    }
+
+    impl Default for ExecutionLimits {
+        fn default() -> Self {
+            Self {
+                max_wall_clock_ratio: default_max_wall_clock_ratio(),
+                max_wall_clock_s: default_max_wall_clock_s(),
+                max_no_progress_s: default_max_no_progress_s(),
+            }
+        }
+    }
+
+    #[derive(Clone, Debug)]
+    pub struct ExecutionMonitor {
+        start_time: Instant,
+        last_progress: Instant,
+        max_wall_clock: Option<StdDuration>,
+        max_no_progress: Option<StdDuration>,
+    }
+
+    impl ExecutionMonitor {
+        pub fn new(limits: ExecutionLimits, sim_duration_s: f64) -> Self {
+            let max_wall_clock = compute_max_wall_clock(
+                sim_duration_s,
+                limits.max_wall_clock_ratio,
+                limits.max_wall_clock_s,
+            );
+            let max_no_progress = if limits.max_no_progress_s > 0.0 {
+                Some(StdDuration::from_secs_f64(limits.max_no_progress_s))
+            } else {
+                None
+            };
+            let now = Instant::now();
+
+            Self {
+                start_time: now,
+                last_progress: now,
+                max_wall_clock,
+                max_no_progress,
+            }
+        }
+
+        pub fn check(&mut self, context: &str) -> Result<()> {
+            let now = Instant::now();
+            if let Some(max_wall_clock) = self.max_wall_clock {
+                if now.duration_since(self.start_time) > max_wall_clock {
+                    bail!(
+                        "Execution timeout ({context}): exceeded wall-clock limit of {:.2} s",
+                        max_wall_clock.as_secs_f64()
+                    );
+                }
+            }
+            if let Some(max_no_progress) = self.max_no_progress {
+                let since_progress = now.duration_since(self.last_progress);
+                if since_progress > max_no_progress {
+                    bail!(
+                        "Execution timeout ({context}): no progress for {:.2} s (limit {:.2} s)",
+                        since_progress.as_secs_f64(),
+                        max_no_progress.as_secs_f64()
+                    );
+                }
+            }
+            self.last_progress = now;
+            Ok(())
+        }
+    }
+
+    fn compute_max_wall_clock(
+        sim_duration_s: f64,
+        max_ratio: f64,
+        max_wall_clock_s: f64,
+    ) -> Option<StdDuration> {
+        let mut max_s = None;
+        if sim_duration_s > 0.0 && max_ratio > 0.0 {
+            max_s = Some(sim_duration_s * max_ratio);
+        }
+        if max_wall_clock_s > 0.0 {
+            max_s = Some(match max_s {
+                Some(current) => current.min(max_wall_clock_s),
+                None => max_wall_clock_s,
+            });
+        }
+
+        max_s.and_then(|s| {
+            if s > 0.0 {
+                Some(StdDuration::from_secs_f64(s))
+            } else {
+                None
+            }
+        })
+    }
+}
+
 pub mod health {
     use super::*;
 
@@ -3004,6 +3146,9 @@ pub struct SimulationConfig {
     /// Generate performance plot comparing navigation output to GPS measurements
     #[serde(default)]
     pub generate_plot: bool,
+    /// Execution time limits (wall-clock and no-progress)
+    #[serde(default)]
+    pub execution_limits: ExecutionLimits,
     /// Logging configuration
     #[serde(default)]
     pub logging: LoggingConfig,
@@ -3034,6 +3179,7 @@ impl Default for SimulationConfig {
             seed: default_seed(),
             parallel: false,
             generate_plot: false,
+            execution_limits: ExecutionLimits::default(),
             logging: LoggingConfig::default(),
             closed_loop: Some(ClosedLoopConfig::default()),
             particle_filter: None,
@@ -3704,7 +3850,7 @@ mod tests {
             events: vec![event],
         };
 
-        let res = run_closed_loop(&mut ukf, stream, None);
+        let res = run_closed_loop(&mut ukf, stream, None, None);
         assert!(!res.unwrap().is_empty());
     }
     #[test]
@@ -4084,6 +4230,20 @@ mod tests {
         assert!(limits.cov_diag_max > 0.0);
     }
 
+    #[test]
+    fn test_execution_monitor_no_progress_timeout() {
+        let limits = ExecutionLimits {
+            max_wall_clock_ratio: 0.0,
+            max_wall_clock_s: 0.0,
+            max_no_progress_s: 0.01,
+        };
+        let mut monitor = ExecutionMonitor::new(limits, 1.0);
+
+        std::thread::sleep(std::time::Duration::from_millis(20));
+        let result = monitor.check("test");
+        assert!(result.is_err());
+    }
+
     #[test]
     fn test_health_monitor_check_valid_state() {
         let limits = HealthLimits::default();
@@ -4574,7 +4734,7 @@ mod tests {
         };
 
         let health_limits = HealthLimits::default();
-        let result = run_closed_loop(&mut ukf, stream, Some(health_limits));
+        let result = run_closed_loop(&mut ukf, stream, Some(health_limits), None);
         assert!(result.is_ok());
     }
 

core/tests/integration_tests.rs

@@ -590,7 +590,7 @@ fn test_ukf_closed_loop_on_real_data() {
 
     // Run closed-loop filter
     let results =
-        run_closed_loop(&mut ukf, stream, None).expect("Closed-loop filter should complete");
+        run_closed_loop(&mut ukf, stream, None, None).expect("Closed-loop filter should complete");
 
     // Verify results
     assert!(
@@ -743,7 +743,7 @@ fn test_ukf_with_degraded_gnss() {
     let stream = build_event_stream(&records, &cfg);
 
     // Run closed-loop filter
-    let results = run_closed_loop(&mut ukf, stream, None)
+    let results = run_closed_loop(&mut ukf, stream, None, None)
         .expect("Closed-loop filter with degraded GNSS should complete");
 
     // Compute error metrics
@@ -838,7 +838,7 @@ fn test_ukf_outperforms_dead_reckoning() {
     };
     let stream = build_event_stream(&records, &cfg);
 
-    let ukf_results = run_closed_loop(&mut ukf, stream, None).expect("UKF should complete");
+    let ukf_results = run_closed_loop(&mut ukf, stream, None, None).expect("UKF should complete");
     let ukf_stats = compute_error_metrics(&ukf_results, &records);
 
     // Print comparison
@@ -918,7 +918,7 @@ fn test_ekf_closed_loop_on_real_data() {
 
     // Run closed-loop filter
     let results =
-        run_closed_loop(&mut ekf, stream, None).expect("Closed-loop EKF filter should complete");
+        run_closed_loop(&mut ekf, stream, None, None).expect("Closed-loop EKF filter should complete");
 
     // Verify results
     assert!(
@@ -1068,7 +1068,7 @@ fn test_ekf_with_degraded_gnss() {
     let stream = build_event_stream(&records, &cfg);
 
     // Run closed-loop filter
-    let results = run_closed_loop(&mut ekf, stream, None)
+    let results = run_closed_loop(&mut ekf, stream, None, None)
         .expect("Closed-loop EKF filter with degraded GNSS should complete");
 
     // Compute error metrics
@@ -1176,7 +1176,7 @@ fn test_ekf_outperforms_dead_reckoning() {
     };
     let stream = build_event_stream(&records, &cfg);
 
-    let ekf_results = run_closed_loop(&mut ekf, stream, None).expect("EKF should complete");
+    let ekf_results = run_closed_loop(&mut ekf, stream, None, None).expect("EKF should complete");
     let ekf_stats = compute_error_metrics(&ekf_results, &records);
 
     // Print comparison
@@ -1256,7 +1256,7 @@ fn test_eskf_closed_loop_on_real_data() {
 
     // Run closed-loop filter
     let results =
-        run_closed_loop(&mut eskf, stream, None).expect("Closed-loop ESKF filter should complete");
+        run_closed_loop(&mut eskf, stream, None, None).expect("Closed-loop ESKF filter should complete");
 
     // Verify results
     assert!(
@@ -1405,7 +1405,7 @@ fn test_eskf_with_degraded_gnss() {
     let stream = build_event_stream(&records, &cfg);
 
     // Run closed-loop filter
-    let results = run_closed_loop(&mut eskf, stream, None)
+    let results = run_closed_loop(&mut eskf, stream, None, None)
         .expect("Closed-loop ESKF filter with degraded GNSS should complete");
 
     // Compute error metrics
@@ -1505,7 +1505,7 @@ fn test_eskf_outperforms_dead_reckoning() {
     };
     let stream = build_event_stream(&records, &cfg);
 
-    let eskf_results = run_closed_loop(&mut eskf, stream, None).expect("ESKF should complete");
+    let eskf_results = run_closed_loop(&mut eskf, stream, None, None).expect("ESKF should complete");
     let eskf_stats = compute_error_metrics(&eskf_results, &records);
 
     // Print comparison
@@ -1596,7 +1596,7 @@ fn test_eskf_stability_high_dynamics() {
 
     // Run closed-loop filter
     let results =
-        run_closed_loop(&mut eskf, stream, None).expect("ESKF with high dynamics should complete");
+        run_closed_loop(&mut eskf, stream, None, None).expect("ESKF with high dynamics should complete");
 
     // Verify all results are valid (no NaN or Inf)
     for (i, result) in results.iter().enumerate() {
@@ -1709,7 +1709,7 @@ fn test_filter_comparison() {
         0.0,
     );
     let stream_ukf = build_event_stream(&records, &cfg);
-    let ukf_results = run_closed_loop(&mut ukf, stream_ukf, None).expect("UKF should complete");
+    let ukf_results = run_closed_loop(&mut ukf, stream_ukf, None, None).expect("UKF should complete");
     let ukf_stats = compute_error_metrics(&ukf_results, &records);
 
     // Run EKF
@@ -1721,7 +1721,7 @@ fn test_filter_comparison() {
         true,
     );
     let stream_ekf = build_event_stream(&records, &cfg);
-    let ekf_results = run_closed_loop(&mut ekf, stream_ekf, None).expect("EKF should complete");
+    let ekf_results = run_closed_loop(&mut ekf, stream_ekf, None, None).expect("EKF should complete");
     let ekf_stats = compute_error_metrics(&ekf_results, &records);
 
     // Run ESKF
@@ -1736,7 +1736,7 @@ fn test_filter_comparison() {
     );
 
     let stream_eskf = build_event_stream(&records, &cfg);
-    let eskf_results = run_closed_loop(&mut eskf, stream_eskf, None).expect("ESKF should complete");
+    let eskf_results = run_closed_loop(&mut eskf, stream_eskf, None, None).expect("ESKF should complete");
     let eskf_stats = compute_error_metrics(&eskf_results, &records);
 
     // Print comparison