wowlab_engine_application/

chunk_executor.rs

//! Chunk execution loop. `target_error` is a fraction (0.05 = 5%), not a percent.

use std::time::Duration;

use wowlab_common::time::Instant;
use wowlab_engine_ports::{
    ChunkAssignment, ChunkReport, DecisionTraceSink, EngineError, ProgressDone, ProgressSink,
    ProgressStart, ProgressTick, SimState,
};
use wowlab_engine_rng::seed_prefix;
use wowlab_engine_sim::{EventQueue, SimEngine, SimEngineRefs, TelemetryAccumulator};
use wowlab_engine_telemetry::TelemetrySink;
use wowlab_types::{constants::MS_PER_SECOND, sim::SimTime};

use crate::{ApplicationError, ApplicationStage, ResolvedHandler, error::EngineResultExt as _};
const CHECK_INTERVAL: u32 = 100;

const MIN_SAMPLES_FOR_STDERR: u32 = 2;

#[derive(Clone, Copy, Debug)]
pub struct ChunkRun<'a> {
    pub assignment: &'a ChunkAssignment,
    pub duration_s: f64,
    pub seed_base: u64,
}

trait ProgressTimer {
    fn elapsed(&self) -> Duration;
}

struct SystemProgressTimer {
    started_at: Instant,
}

impl SystemProgressTimer {
    fn start() -> Self {
        Self {
            // #t(rust_ambient_syscall) isolated system-clock adapter for the injected progress timer
            started_at: Instant::now(),
        }
    }
}

impl ProgressTimer for SystemProgressTimer {
    fn elapsed(&self) -> Duration {
        self.started_at.elapsed()
    }
}

struct ExecutionObservers<'a> {
    progress: &'a dyn ProgressSink,
    timer: &'a dyn ProgressTimer,
    trace_extras: TraceExtras,
}

fn enrich_progress_tick(tick: &mut ProgressTick, timer: &dyn ProgressTimer) {
    let elapsed = timer.elapsed();
    let elapsed_s = elapsed.as_secs_f64();

    tick.elapsed_ms = Some(u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX));
    tick.throughput_sps =
        (elapsed_s > 0.0).then(|| f64::from(tick.iterations_completed) / elapsed_s);
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum TraceExtras {
    On,
    Off,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ExecutionOptions {
    trace_extras: TraceExtras,
    capture_representative: bool,
}

fn accumulator_to_report(
    accumulator: TelemetryAccumulator,
    assignment: &ChunkAssignment,
) -> ChunkReport {
    let telemetry_bytes = accumulator.encode(assignment.chunk_index, assignment.permutation_index);

    ChunkReport {
        job_id: assignment.job_id.clone(),
        chunk_id: assignment.chunk_id.clone(),
        chunk_index: assignment.chunk_index,
        telemetry_bytes,
    }
}

fn encounter_end_time(duration_s: f64) -> Result<SimTime, ApplicationError> {
    let millis = u32::try_from(wowlab_types::numeric::f64_to_u64_saturating_trunc(
        duration_s * MS_PER_SECOND,
    ))
    .map_err(|e| {
        EngineError::chunk_validation(format!(
            "duration_s {duration_s} overflows the millisecond clock: {e}"
        ))
    })
    .in_application_stage(ApplicationStage::ChunkExecution)?;

    Ok(SimTime::from_millis(millis))
}

fn relative_stderr(mean_dps: f64, std_dps: f64, n: u32) -> Option<f64> {
    if n < MIN_SAMPLES_FOR_STDERR
        || !mean_dps.is_finite()
        || mean_dps <= 0.0
        || !std_dps.is_finite()
    {
        return None;
    }

    let stderr = std_dps / (f64::from(n)).sqrt();

    Some(stderr / mean_dps)
}

/// Run N iterations and return a `ChunkReport`.
pub fn run_chunk(
    run: ChunkRun<'_>,
    progress: &dyn ProgressSink,
    handler_factory: impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<ChunkReport, ApplicationError> {
    let accumulator = run_chunk_into_accumulator_impl(
        run,
        progress,
        ExecutionOptions {
            trace_extras: TraceExtras::Off,
            capture_representative: true,
        },
        handler_factory,
    )?;

    Ok(accumulator_to_report(accumulator, run.assignment))
}

/// Trace-runner variant of [`run_chunk`] that attaches a decision sink and trace extras.
pub(crate) fn run_chunk_with_trace<T>(
    run: ChunkRun<'_>,
    progress: &dyn ProgressSink,
    trace_sink: T,
    handler_factory: impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<ChunkReport, ApplicationError>
where
    T: DecisionTraceSink + 'static,
{
    let trace_sink: std::sync::Arc<dyn DecisionTraceSink> = std::sync::Arc::new(trace_sink);
    let accumulator = run_chunk_into_accumulator_impl(
        run,
        progress,
        ExecutionOptions {
            trace_extras: TraceExtras::On,
            capture_representative: true,
        },
        move || {
            let mut handler = handler_factory()?;

            handler
                .handler_mut()
                .attach_decision_trace(std::sync::Arc::clone(&trace_sink));

            Ok(handler)
        },
    )?;

    Ok(accumulator_to_report(accumulator, run.assignment))
}

/// Like [`run_chunk`] but yields the raw `TelemetryAccumulator` for merging.
pub fn run_chunk_into_accumulator(
    run: ChunkRun<'_>,
    progress: &dyn ProgressSink,
    handler_factory: impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<TelemetryAccumulator, ApplicationError> {
    run_chunk_into_accumulator_impl(
        run,
        progress,
        ExecutionOptions {
            trace_extras: TraceExtras::Off,
            capture_representative: true,
        },
        handler_factory,
    )
}

pub(crate) fn run_chunk_into_accumulator_for_merge(
    run: ChunkRun<'_>,
    progress: &dyn ProgressSink,
    handler_factory: impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<TelemetryAccumulator, ApplicationError> {
    run_chunk_into_accumulator_impl(
        run,
        progress,
        ExecutionOptions {
            trace_extras: TraceExtras::Off,
            capture_representative: false,
        },
        handler_factory,
    )
}

fn run_chunk_into_accumulator_impl(
    run: ChunkRun<'_>,
    progress: &dyn ProgressSink,
    options: ExecutionOptions,
    handler_factory: impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<TelemetryAccumulator, ApplicationError> {
    let timer = SystemProgressTimer::start();

    run_chunk_into_accumulator_with_timer(
        run,
        &ExecutionObservers {
            progress,
            timer: &timer,
            trace_extras: options.trace_extras,
        },
        options.capture_representative,
        &handler_factory,
    )
}

fn run_chunk_into_accumulator_with_timer(
    run: ChunkRun<'_>,
    observers: &ExecutionObservers<'_>,
    capture_representative: bool,
    handler_factory: &impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<TelemetryAccumulator, ApplicationError> {
    let ChunkRun {
        assignment,
        duration_s,
        seed_base,
    } = run;
    let encounter_end = encounter_end_time(duration_s)?;

    let max_iters = if assignment.target_error.is_some() {
        assignment
            .max_iterations
            .unwrap_or(assignment.iterations)
            .max(1)
    } else {
        assignment.iterations
    };
    let min_iters = assignment.min_iterations.unwrap_or(1).max(1);

    observers.progress.on_start(ProgressStart {
        job_id: assignment.job_id.clone(),
        chunk_id: assignment.chunk_id.clone(),
        iterations: max_iters,
    });

    let mut accumulator = TelemetryAccumulator::new();

    if observers.trace_extras == TraceExtras::On {
        accumulator.enable_trace_extras();
    }

    let mut handler = handler_factory()?.into_inner();
    let mut queue = EventQueue::new();
    let mut sink = TelemetrySink::new();

    handler.install_target_metadata(&mut sink);

    // docref:start chunk-seed-derivation
    let seed_prefix = seed_prefix(seed_base, &assignment.chunk_id);
    // docref:end chunk-seed-derivation

    tracing::debug!(
        chunk_id = %assignment.chunk_id,
        seed_base,
        seed_prefix,
        "chunk seed derived",
    );

    let mut completed: u32 = 0;
    // docref:start orchestration-iteration-reset
    for i in 0..max_iters {
        wowlab_types::bench_metrics::measure_reset(|| handler.reset());
        queue.clear();
        // docref:end orchestration-iteration-reset

        let state = SimState::new(i, encounter_end, seed_prefix);

        let mut engine = SimEngine::new(
            state,
            SimEngineRefs {
                telemetry: &mut accumulator,
                handler: handler.as_mut(),
                queue: &mut queue,
                sink: &mut sink,
            },
        );

        engine
            .run()
            .in_application_stage(ApplicationStage::ChunkExecution)?;
        completed = i + 1;

        let is_check_tick = completed.is_multiple_of(CHECK_INTERVAL) || completed == max_iters;

        if !is_check_tick {
            continue;
        }

        let stats = accumulator.running_stats();
        let rel_stderr = stats
            .as_ref()
            .and_then(|s| relative_stderr(s.mean_dps, s.std_dps, completed));

        let should_stop = match assignment.target_error {
            Some(te) if completed >= min_iters => {
                matches!(rel_stderr, Some(re) if re < te)
            }
            _ => false,
        };

        // #t(block: rust_clone_in_loop) ProgressTick requires owned String; only cloned every CHECK_INTERVAL iterations
        let mut tick = ProgressTick::new(assignment.chunk_id.clone(), completed, max_iters);

        tick.running_mean_dps = stats.as_ref().map(|s| s.mean_dps);
        tick.running_std_dps = stats.as_ref().map(|s| s.std_dps);
        tick.running_min_dps = stats.as_ref().map(|s| s.min_dps);
        tick.running_max_dps = stats.as_ref().map(|s| s.max_dps);
        tick.relative_stderr = rel_stderr;
        enrich_progress_tick(&mut tick, observers.timer);
        observers.progress.on_tick(tick);

        if should_stop {
            break;
        }
    }

    if capture_representative {
        capture_representative_with_handler(
            &mut accumulator,
            encounter_end,
            handler.as_mut(),
            &mut queue,
            &mut sink,
        )?;
    }

    observers.progress.on_complete(ProgressDone {
        chunk_id: assignment.chunk_id.clone(),
        iterations_completed: completed,
    });

    Ok(accumulator)
}

fn capture_representative_with_handler(
    accumulator: &mut TelemetryAccumulator,
    encounter_end: SimTime,
    handler: &mut dyn wowlab_engine_ports::SpecHandler,
    queue: &mut EventQueue,
    sink: &mut TelemetrySink,
) -> Result<(), ApplicationError> {
    let Some(representative) = accumulator.representative_iteration() else {
        return Ok(());
    };

    wowlab_types::bench_metrics::measure_reset(|| handler.reset());
    queue.clear();

    SimEngine::new(
        SimState::new(
            representative.iteration,
            encounter_end,
            representative.seed_prefix,
        ),
        SimEngineRefs {
            telemetry: accumulator,
            handler,
            queue,
            sink,
        },
    )
    .capture_representative()
    .in_application_stage(ApplicationStage::ChunkExecution)
}

pub(crate) fn capture_exact_representative(
    accumulator: &mut TelemetryAccumulator,
    duration_s: f64,
    handler_factory: impl Fn() -> Result<ResolvedHandler, ApplicationError>,
) -> Result<(), ApplicationError> {
    if accumulator.representative_iteration().is_none() {
        return Ok(());
    }

    let encounter_end = encounter_end_time(duration_s)?;
    let mut handler = handler_factory()?.into_inner();
    let mut queue = EventQueue::new();
    let mut sink = TelemetrySink::new();

    handler.install_target_metadata(&mut sink);

    capture_representative_with_handler(
        accumulator,
        encounter_end,
        handler.as_mut(),
        &mut queue,
        &mut sink,
    )
}

#[cfg(test)]
mod tests {
    use std::error::Error as _;

    use googletest::prelude::*;
    use rstest::rstest;

    use super::*;

    struct FixedProgressTimer(Duration);

    #[gtest]
    fn simulation_run_failure_retains_ports_source_and_exact_display() -> Result<()> {
        let error = EngineError::from(wowlab_engine_ports::SimRunError::event_budget_exceeded(
            500_001,
        ));

        verify_that!(
            error.to_string(),
            eq(
                "simulation runtime error: event budget exceeded: simulator processed 500001 events without converging"
            )
        )?;
        let source = error
            .source()
            .and_then(std::error::Error::source)
            .and_then(|source| source.downcast_ref::<wowlab_engine_ports::SimRunError>())
            .or_fail()?;

        verify_that!(source.event_count(), eq(Some(500_001)))?;

        Ok(())
    }

    impl ProgressTimer for FixedProgressTimer {
        fn elapsed(&self) -> Duration {
            self.0
        }
    }

    #[gtest]
    fn progress_tick_enrichment_uses_injected_elapsed_time() -> Result<()> {
        let mut tick = ProgressTick::new("test".to_string(), 100, 200);

        enrich_progress_tick(&mut tick, &FixedProgressTimer(Duration::from_millis(1_250)));

        verify_that!(tick.elapsed_ms, some(eq(1_250)))?;

        verify_that!(tick.throughput_sps, some(near(80.0, 1e-12)))
    }

    #[gtest]
    fn zero_elapsed_time_omits_undefined_throughput() -> Result<()> {
        let mut tick = ProgressTick::new("test".to_string(), 100, 200);

        enrich_progress_tick(&mut tick, &FixedProgressTimer(Duration::ZERO));

        verify_that!(tick.elapsed_ms, some(eq(0)))?;

        verify_that!(tick.throughput_sps, none())
    }

    #[gtest]
    #[rstest]
    #[case::whole_second(10.0, 10_000)]
    #[case::fractional(1.5, 1_500)]
    #[case::zero(0.0, 0)]
    fn encounter_end_time_cases(#[case] duration_s: f64, #[case] expected_ms: u32) -> Result<()> {
        verify_that!(
            encounter_end_time(duration_s).map(|time| time.as_millis()),
            ok(eq(&expected_ms))
        )
    }

    #[gtest]
    fn encounter_end_time_overflow_is_chunk_validation_err() -> Result<()> {
        let result = encounter_end_time(5_000_000.0);

        verify_that!(
            result.as_ref().err().or_fail()?.engine_error(),
            some(predicate(EngineError::is_chunk_validation))
        )?;

        verify_that!(
            result.err().or_fail()?.to_string(),
            contains_substring("overflows the millisecond clock")
        )
    }

    #[gtest]
    #[rstest]
    #[case::n_below_two_none(100.0, 10.0, 1, None)]
    #[case::known_value(1000.0, 100.0, 100, Some(0.01))]
    #[case::mean_zero_none(0.0, 10.0, 10, None)]
    #[case::mean_negative_none(-5.0, 10.0, 10, None)]
    #[case::mean_nan_none(f64::NAN, 10.0, 10, None)]
    #[case::std_infinite_none(100.0, f64::INFINITY, 10, None)]
    #[case::n_two_boundary_some(100.0, 0.0, 2, Some(0.0))]
    fn relative_stderr_cases(
        #[case] mean_dps: f64,
        #[case] std_dps: f64,
        #[case] n: u32,
        #[case] expected: Option<f64>,
    ) -> Result<()> {
        let actual = relative_stderr(mean_dps, std_dps, n);

        if let Some(e) = expected {
            verify_that!(actual, some(near(e, 1e-12)))
        } else {
            verify_that!(actual, none())
        }
    }
}