Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 125 additions & 2 deletions core/src/sim.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1944,7 +1944,7 @@ pub fn run_closed_loop<F: NavigationFilter>(
let cov = filter.get_certainty();
results.push(NavigationResult::from((&start_time, &mean, &cov)));
debug!("Initial filter state at {}: {:?}", start_time, mean);
let mut last_ts = start_time;
let mut last_ts = Some(start_time);

for (i, event) in stream.events.into_iter().enumerate() {
// Print detailed progress every 100 iterations or at key milestones
Expand Down Expand Up @@ -2034,7 +2034,7 @@ pub fn run_closed_loop<F: NavigationFilter>(
let cov = filter.get_certainty();
results.push(NavigationResult::from((&ts, &mean, &cov)));
debug!("Filter state at {}: {:?}", ts, mean);
last_ts = ts;
last_ts = Some(ts);
}
}
debug!("Closed-loop simulation complete");
Expand Down Expand Up @@ -2576,6 +2576,17 @@ pub fn print_sim_status<F: NavigationFilter>(filter: &F) {
pub mod execution {
use super::*;

/// Configuration for execution timeout limits in simulations.
///
/// This struct provides multiple timeout mechanisms to prevent runaway computations
/// and detect performance issues during simulation execution:
///
/// - **Wall-clock ratio timeout**: Limits execution time relative to simulation duration
/// - **Absolute wall-clock timeout**: Enforces a hard limit in real-world seconds
/// - **No-progress timeout**: Detects when simulation makes no progress (possible hang)
///
/// All timeout values <= 0 are treated as disabled. The most restrictive active timeout
/// will trigger first. These limits are enforced by [`ExecutionMonitor`] during simulation.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ExecutionLimits {
/// Max wall-clock time as a ratio of simulated duration (<= 0 disables).
Expand Down Expand Up @@ -2611,6 +2622,17 @@ pub mod execution {
}
}

/// Monitors execution time and progress during simulation runs.
///
/// This struct tracks wall-clock time and detects stalled simulations by monitoring
/// when progress was last reported. It enforces timeout limits specified in [`ExecutionLimits`].
///
/// The monitor maintains two types of timeouts:
/// - **Wall-clock timeout**: Maximum total execution time (computed from simulation duration and limits)
/// - **No-progress timeout**: Maximum time without calling `mark_progress()`
///
/// Use [`check`](ExecutionMonitor::check) periodically during simulation to verify execution
/// stays within limits, and call `mark_progress()` after each significant simulation step.
#[derive(Clone, Debug)]
pub struct ExecutionMonitor {
start_time: Instant,
Expand All @@ -2620,6 +2642,17 @@ pub mod execution {
}

impl ExecutionMonitor {
/// Creates a new execution monitor with the specified limits.
///
/// # Arguments
///
/// * `limits` - Timeout configuration including wall-clock and no-progress limits
/// * `sim_duration_s` - Expected simulation duration in seconds, used to compute
/// wall-clock timeout when `max_wall_clock_ratio` is enabled
///
/// # Returns
///
/// A new `ExecutionMonitor` instance initialized with the current time.
pub fn new(limits: ExecutionLimits, sim_duration_s: f64) -> Self {
let max_wall_clock = compute_max_wall_clock(
sim_duration_s,
Expand All @@ -2641,6 +2674,33 @@ pub mod execution {
}
}

/// Checks if execution is within timeout limits.
///
/// This method verifies that the simulation has not exceeded wall-clock or no-progress
/// timeout limits. It should be called periodically during simulation execution.
///
/// # Arguments
///
/// * `context` - A string describing the current execution context, included in error messages
///
/// # Returns
///
/// * `Ok(())` if execution is within limits
/// * `Err(...)` with a descriptive message if any timeout has been exceeded
///
/// # Example
///
/// ```no_run
/// # use strapdown_core::sim::{ExecutionMonitor, ExecutionLimits};
/// let limits = ExecutionLimits::default();
/// let mut monitor = ExecutionMonitor::new(limits, 100.0);
///
/// // Check timeout before processing
/// monitor.check("data processing")?;
/// // ... do work ...
/// monitor.mark_progress();
/// # Ok::<(), anyhow::Error>(())
/// ```
pub fn check(&mut self, context: &str) -> Result<()> {
let now = Instant::now();
if let Some(max_wall_clock) = self.max_wall_clock
Expand Down Expand Up @@ -4303,6 +4363,69 @@ mod tests {
assert!(result.is_err());
}

#[test]
fn test_execution_monitor_wall_clock_timeout() {
let limits = ExecutionLimits {
max_wall_clock_ratio: 0.0,
max_wall_clock_s: 0.01,
max_no_progress_s: 0.0,
};
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());
assert!(result.unwrap_err().to_string().contains("wall-clock limit"));
}

#[test]
fn test_execution_monitor_successful_execution_with_progress() {
let limits = ExecutionLimits {
max_wall_clock_ratio: 0.0,
max_wall_clock_s: 1.0,
max_no_progress_s: 0.05,
};
let mut monitor = ExecutionMonitor::new(limits, 1.0);

// Simulate work with regular progress updates
for _ in 0..5 {
std::thread::sleep(std::time::Duration::from_millis(10));
let result = monitor.check("test");
assert!(result.is_ok());
monitor.mark_progress();
}
}

#[test]
fn test_execution_monitor_zero_timeout_disabled() {
let limits = ExecutionLimits {
max_wall_clock_ratio: 0.0,
max_wall_clock_s: 0.0,
max_no_progress_s: 0.0,
};
let mut monitor = ExecutionMonitor::new(limits, 1.0);

// With all timeouts disabled, sleep and check should succeed
std::thread::sleep(std::time::Duration::from_millis(50));
let result = monitor.check("test");
assert!(result.is_ok());
}

#[test]
fn test_execution_monitor_negative_timeout_disabled() {
let limits = ExecutionLimits {
max_wall_clock_ratio: -1.0,
max_wall_clock_s: -1.0,
max_no_progress_s: -1.0,
};
let mut monitor = ExecutionMonitor::new(limits, 1.0);

// With negative timeouts (disabled), sleep and check should succeed
std::thread::sleep(std::time::Duration::from_millis(50));
let result = monitor.check("test");
assert!(result.is_ok());
}

#[test]
fn test_health_monitor_check_valid_state() {
let limits = HealthLimits::default();
Expand Down
6 changes: 3 additions & 3 deletions core/tests/integration_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1977,7 +1977,7 @@ fn test_filter_output_length_matches_input() {
);

let event_stream = build_event_stream(&records, &degradation);
let ukf_results = run_closed_loop(&mut ukf, event_stream, None)
let ukf_results = run_closed_loop(&mut ukf, event_stream, None, None)
.expect("UKF closed loop should complete successfully");

assert_eq!(
Expand All @@ -2003,7 +2003,7 @@ fn test_filter_output_length_matches_input() {
);

let event_stream = build_event_stream(&records, &degradation);
let ekf_results = run_closed_loop(&mut ekf, event_stream, None)
let ekf_results = run_closed_loop(&mut ekf, event_stream, None, None)
.expect("EKF closed loop should complete successfully");

assert_eq!(
Expand All @@ -2024,7 +2024,7 @@ fn test_filter_output_length_matches_input() {
ErrorStateKalmanFilter::new(initial_state, imu_biases, initial_covariance, process_noise);

let event_stream = build_event_stream(&records, &degradation);
let eskf_results = run_closed_loop(&mut eskf, event_stream, None)
let eskf_results = run_closed_loop(&mut eskf, event_stream, None, None)
.expect("ESKF closed loop should complete successfully");

assert_eq!(
Expand Down
16 changes: 10 additions & 6 deletions sim/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -56,8 +56,9 @@ use strapdown::sim::health::HealthMonitor;
use strapdown::sim::{DEFAULT_PROCESS_NOISE, GeoResolution};
use strapdown::sim::{
ExecutionLimits, ExecutionMonitor, FaultArgs, FilterType, NavigationResult, ParticleFilterType,
SchedulerArgs, SimulationConfig, SimulationMode, TestDataRecord, build_fault, build_scheduler,
dead_reckoning, initialize_ekf, initialize_eskf, initialize_ukf, run_closed_loop,
SchedulerArgs, SimulationConfig, SimulationMode, TestDataRecord, UkfConfig, build_fault,
build_scheduler, dead_reckoning, initialize_ekf, initialize_eskf, initialize_ukf,
run_closed_loop,
};

const LONG_ABOUT: &str =
Expand Down Expand Up @@ -416,7 +417,7 @@ fn process_file(
let results = match filter_config.filter {
FilterType::Ukf => {
let mut ukf =
initialize_ukf(records[0].clone(), None, None, None, None, None, None);
initialize_ukf(records[0].clone(), UkfConfig::default());
info!("Initialized UKF");
run_closed_loop(&mut ukf, event_stream, None, Some(execution_limits.clone()))
}
Expand Down Expand Up @@ -635,7 +636,7 @@ fn process_file(
Event::Imu { dt_s, imu, .. } => {
rbpf.predict(&imu, dt_s);
}
Event::Measurement { mut meas, .. } => {
Event::Measurement { meas, .. } => {
#[cfg(feature = "geonav")]
if gravity_map.is_some() || magnetic_map.is_some() {
let (mean, _) = rbpf.estimate();
Expand Down Expand Up @@ -807,6 +808,7 @@ fn run_from_config(
/// This is a helper function that extracts the common logic for running closed-loop simulations
/// with either UKF or EKF filters. It handles event stream creation, filter initialization,
/// simulation execution, and results writing.
#[allow(clippy::too_many_arguments)]
fn run_single_closed_loop_simulation(
filter_type: FilterType,
records: &[TestDataRecord],
Expand Down Expand Up @@ -1135,7 +1137,9 @@ fn run_geo_closed_loop_cli(args: &ClosedLoopSimArgs) -> Result<(), Box<dyn Error
// Get all CSV files to process
let csv_files = get_csv_files(&args.sim.input)?;
let is_multiple = csv_files.len() > 1;
let _execution_limits = execution_limits_from_args(&args.sim);
// NOTE: Execution limits are not yet applied in geophysical closed-loop simulations.
// We still parse the arguments here to validate them and keep CLI behavior consistent.
let _ = execution_limits_from_args(&args.sim);

if is_multiple {
info!("Processing {} CSV files from directory", csv_files.len());
Expand Down Expand Up @@ -1543,7 +1547,7 @@ fn run_particle_filter(args: &ParticleFilterSimArgs) -> Result<(), Box<dyn Error
Event::Imu { dt_s, imu, .. } => {
rbpf.predict(&imu, dt_s);
}
Event::Measurement { mut meas, .. } => {
Event::Measurement { meas, .. } => {
#[cfg(feature = "geonav")]
if args.geo.geo {
let (mean, _) = rbpf.estimate();
Expand Down