wowlab_engine_application/
simulate_parallel.rs

1//! Unified parallel simulation driver.
2
3use rayon::prelude::*;
4use wowlab_engine_ports::{
5    ChunkAssignment, ContentCatalog, DynDataResolver, EngineError, ProgressSink,
6};
7use wowlab_engine_sim::TelemetryAccumulator;
8
9use crate::{
10    ApplicationError, ApplicationStage,
11    error::EngineResultExt as _,
12    intent_builder::IntentOverrides,
13    simulate_intent::{build_handler_from_setup, resolve_setup},
14};
15
16const PARALLEL_JOB_LABEL: &str = "sim";
17
18/// Parallel execution settings for one resolved simulation intent.
19#[derive(Clone, Copy, Debug)]
20pub struct ParallelRun {
21    pub seed_base: u64,
22    pub total_iterations: u32,
23    pub num_threads: usize,
24}
25
26/// Resolve `sim_config` once, run `total_iters` iterations across `num_threads`, encode the result.
27pub async fn simulate_intent_parallel(
28    catalog: &ContentCatalog,
29    sim_config: &str,
30    run: ParallelRun,
31    resolver: &DynDataResolver<'_>,
32    progress: &dyn ProgressSink,
33) -> Result<Vec<u8>, ApplicationError> {
34    let num_threads = run.num_threads.max(1);
35
36    let setup = resolve_setup(catalog, sim_config, resolver, &IntentOverrides::default()).await?;
37    let duration = setup.duration_s;
38
39    let assignments = chunk_assignments(run.total_iterations, num_threads);
40
41    let pool = rayon::ThreadPoolBuilder::new()
42        .num_threads(num_threads)
43        .build()
44        .map_err(|e| EngineError::simulation_runtime(format!("failed to build thread pool: {e}")))
45        .in_application_stage(ApplicationStage::ParallelSimulation)?;
46
47    let accumulators: Vec<Result<TelemetryAccumulator, ApplicationError>> = pool.install(|| {
48        assignments
49            .par_iter()
50            .map(|assignment| {
51                let factory = || build_handler_from_setup(&setup);
52
53                crate::chunk_executor::run_chunk_into_accumulator_for_merge(
54                    crate::ChunkRun {
55                        assignment,
56                        duration_s: duration,
57                        seed_base: run.seed_base,
58                    },
59                    progress,
60                    factory,
61                )
62            })
63            .collect()
64    });
65
66    let mut merged = merge_accumulators(accumulators)?;
67
68    crate::chunk_executor::capture_exact_representative(&mut merged, duration, || {
69        build_handler_from_setup(&setup)
70    })?;
71
72    Ok(merged.encode(0, None))
73}
74
75fn merge_accumulators(
76    accumulators: Vec<Result<TelemetryAccumulator, ApplicationError>>,
77) -> Result<TelemetryAccumulator, ApplicationError> {
78    let mut merged: Option<TelemetryAccumulator> = None;
79
80    for accumulator in accumulators {
81        let accumulator = accumulator?;
82
83        match &mut merged {
84            Some(current) => current.merge(accumulator),
85            None => merged = Some(accumulator),
86        }
87    }
88
89    Ok(merged.unwrap_or_default())
90}
91
92fn chunk_assignments(total_iters: u32, num_threads: usize) -> Vec<ChunkAssignment> {
93    let threads = wowlab_types::numeric::usize_to_u32_saturating(num_threads);
94    let base_count = total_iters / threads;
95    let remainder = total_iters % threads;
96
97    (0..threads)
98        .map(|t| {
99            let my_count = base_count + u32::from(t < remainder);
100
101            ChunkAssignment::single_indexed(PARALLEL_JOB_LABEL, t, my_count)
102        })
103        .collect()
104}
105
106#[cfg(test)]
107mod tests {
108    use googletest::prelude::*;
109    use wowlab_engine_ports::{NoopProgress, SimContext, test_spec_handler};
110
111    use super::*;
112
113    struct PrefixDamageHandler {
114        high_damage_prefix: u64,
115        damage: f64,
116    }
117
118    test_spec_handler! {
119    impl PrefixDamageHandler {
120        defaults [on_player_ready, on_cast_complete, cast_time_ms]
121
122        fn on_sim_start(&mut self, ctx: &mut SimContext) {
123            self.damage = if ctx.state.seed_prefix == self.high_damage_prefix {
124                200.0
125            } else {
126                f64::from(ctx.state.iteration) * 100.0
127            };
128        }
129
130        fn total_damage(&self) -> f64 {
131            self.damage
132        }
133
134    }
135    }
136
137    #[gtest]
138    fn chunk_assignments_partition_iterations_exactly() -> Result<()> {
139        let chunks = chunk_assignments(103, 4);
140
141        verify_that!(chunks.len(), eq(4))
142            .with_failure_message(|| "one chunk per thread".to_string())?;
143        let total: u32 = chunks.iter().map(|c| c.iterations).sum();
144
145        verify_that!(total, eq(103))
146            .with_failure_message(|| "chunks sum to the requested iteration count".to_string())?;
147        verify_that!(
148            (chunks[0].iterations, chunks[3].iterations),
149            (eq(26), eq(25))
150        )
151        .with_failure_message(|| "remainder belongs to the earliest chunks".to_string())?;
152
153        Ok(())
154    }
155
156    #[gtest]
157    fn chunk_assignments_have_distinct_ids() -> Result<()> {
158        let chunks = chunk_assignments(10, 3);
159        let ids: std::collections::BTreeSet<&str> =
160            chunks.iter().map(|c| c.chunk_id.as_str()).collect();
161
162        verify_that!(ids.len(), eq(3))
163            .with_failure_message(|| "each chunk id is distinct for seed separation".to_string())?;
164
165        Ok(())
166    }
167
168    #[gtest]
169    fn single_thread_is_one_chunk() -> Result<()> {
170        let chunks = chunk_assignments(50, 1);
171
172        verify_that!(chunks.len(), eq(1)).with_failure_message(|| {
173            "single thread is the one-chunk degenerate case".to_string()
174        })?;
175        verify_that!(chunks[0].iterations, eq(50))
176            .with_failure_message(|| "the lone chunk owns every iteration".to_string())?;
177
178        Ok(())
179    }
180
181    #[gtest]
182    fn multi_chunk_final_mean_selects_and_reruns_exact_iteration() -> Result<()> {
183        let assignments = chunk_assignments(3, 2);
184        let high_damage_prefix = wowlab_engine_rng::seed_prefix(0, &assignments[1].chunk_id);
185        let factory = || {
186            Ok(crate::ResolvedHandler::new(PrefixDamageHandler {
187                high_damage_prefix,
188                damage: 0.0,
189            }))
190        };
191        let first = crate::chunk_executor::run_chunk_into_accumulator_for_merge(
192            crate::ChunkRun {
193                assignment: &assignments[0],
194                duration_s: 1.0,
195                seed_base: 0,
196            },
197            &NoopProgress,
198            factory,
199        )?;
200        let second = crate::chunk_executor::run_chunk_into_accumulator_for_merge(
201            crate::ChunkRun {
202                assignment: &assignments[1],
203                duration_s: 1.0,
204                seed_base: 0,
205            },
206            &NoopProgress,
207            factory,
208        )?;
209        let mut merged = merge_accumulators(vec![Ok(first), Ok(second)])?;
210        let before = merged.running_stats().or_fail()?;
211        let selected = merged.representative_iteration().or_fail()?;
212
213        verify_that!(before.mean_dps, near(100.0, 1e-9))?;
214        verify_that!(selected.iteration, eq(1))?;
215
216        crate::chunk_executor::capture_exact_representative(&mut merged, 1.0, factory)?;
217
218        let after = merged.running_stats().or_fail()?;
219
220        verify_that!(after.mean_dps, near(before.mean_dps, 1e-9))?;
221        verify_that!(after.std_dps, near(before.std_dps, 1e-9))?;
222
223        verify_that!(merged.representative_dps(), some(near(100.0, 1e-9)))
224    }
225}
wowlab_engine_application/simulate_parallel.rs

wowlab_engine_application/
simulate_parallel.rs
