diff --git a/server/Cargo.toml b/server/Cargo.toml index 617e3ca6e..d3857ef7d 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -6,7 +6,11 @@ edition = "2021" [dependencies] bevy_ecs = "0.18" bevy_app = "0.18" +# multi_threaded enables Bevy's parallel system executor (no explicit import needed — +# bevy_ecs detects the feature on its bevy_tasks dependency at compile time). bevy_tasks = { version = "0.18", features = ["multi_threaded"] } +# par_iter infrastructure for data-parallel systems. Not yet called — candidate +# systems marked with TODO comments. Active use begins when profiling shows bottlenecks. rayon = "1" serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" diff --git a/server/src/simulation/monologue.rs b/server/src/simulation/monologue.rs index ea0793bfe..d8490b0bf 100644 --- a/server/src/simulation/monologue.rs +++ b/server/src/simulation/monologue.rs @@ -588,11 +588,12 @@ fn has_hear_sound_event( pub fn trigger_monologue( _time: Res, _query: Query< - (&TilePosition, &mut MonologueState, &mut MonologueBuffer), + (&TilePosition, &MonologueState, &MonologueBuffer), With, >, ) { // Intentional no-op — kept as schedule ordering anchor. See doc comment. + // Uses shared refs (not &mut) to avoid blocking parallel systems. } // --------------------------------------------------------------------------- @@ -790,7 +791,7 @@ mod tests { fn setup_anomaly_world() -> World { let mut world = World::new(); world.init_resource::(); - world.insert_resource(SimRng::new(42)); + // SimRng no longer needed — migrated systems use EntityRng world } @@ -991,7 +992,7 @@ mod tests { fn setup_recognition_world() -> World { let mut world = World::new(); world.init_resource::(); - world.insert_resource(SimRng::new(42)); + // SimRng no longer needed — migrated systems use EntityRng world } @@ -1353,7 +1354,7 @@ mod tests { fn setup_event_world() -> World { let mut world = World::new(); world.init_resource::(); - world.insert_resource(SimRng::new(42)); + // SimRng no longer needed — migrated systems use EntityRng world.init_resource::(); world.init_resource::(); world.init_resource::(); @@ -1923,7 +1924,7 @@ mod tests { fn setup_contradiction_world() -> bevy_ecs::world::World { let mut world = bevy_ecs::world::World::new(); world.init_resource::(); - world.insert_resource(SimRng::new(42)); + // SimRng no longer needed — migrated systems use EntityRng world.insert_resource(ContradictionDetectedQueue::default()); world.init_resource::(); world diff --git a/server/src/simulation/rng.rs b/server/src/simulation/rng.rs index b2198b824..2bbf17188 100644 --- a/server/src/simulation/rng.rs +++ b/server/src/simulation/rng.rs @@ -52,7 +52,9 @@ impl EntityRng { /// Uses splitmix64 mixing to combine the two seeds with good avalanche /// properties — avoids correlated sequences for entities with similar IDs. pub fn from_seed_and_id(world_seed: u64, stable_id: u64) -> Self { - let mixed = splitmix64(world_seed.wrapping_add(stable_id)); + // Mix each input independently then XOR — avoids the collision class + // where (seed=0, id=N) == (seed=1, id=N-1) that wrapping_add creates. + let mixed = splitmix64(world_seed) ^ splitmix64(stable_id); Self { rng: ChaCha20Rng::seed_from_u64(mixed), } @@ -91,4 +93,33 @@ mod tests { let val2: u32 = rng2.rng.random(); assert_ne!(val1, val2); } + + #[test] + fn entity_rng_same_seed_same_id_same_sequence() { + let mut a = EntityRng::from_seed_and_id(42, 7); + let mut b = EntityRng::from_seed_and_id(42, 7); + let va: Vec = (0..100).map(|_| a.rng.random()).collect(); + let vb: Vec = (0..100).map(|_| b.rng.random()).collect(); + assert_eq!(va, vb); + } + + #[test] + fn entity_rng_same_seed_different_id_different_sequence() { + let mut a = EntityRng::from_seed_and_id(42, 0); + let mut b = EntityRng::from_seed_and_id(42, 1); + let va: u32 = a.rng.random(); + let vb: u32 = b.rng.random(); + assert_ne!(va, vb); + } + + #[test] + fn entity_rng_adjacent_seeds_no_collision() { + // Verify (seed=0, id=N) != (seed=1, id=N-1) — the collision class + // that wrapping_add would create. + let mut a = EntityRng::from_seed_and_id(0, 5); + let mut b = EntityRng::from_seed_and_id(1, 4); + let va: u32 = a.rng.random(); + let vb: u32 = b.rng.random(); + assert_ne!(va, vb); + } } diff --git a/server/src/workers/pool.rs b/server/src/workers/pool.rs index daf9f62ef..2376fd6dc 100644 --- a/server/src/workers/pool.rs +++ b/server/src/workers/pool.rs @@ -54,7 +54,7 @@ pub struct BackgroundWorkerPool { tx: Sender>, rx: Receiver<(Resp, DeliveryStrategy)>, shutdown: Arc, - _workers: Vec>, + workers: Vec>, } impl BackgroundWorkerPool @@ -84,7 +84,7 @@ where let handle = thread::Builder::new() .name(format!("worker-{id}")) .spawn(move || { - while !shutdown.load(Ordering::Relaxed) { + while !shutdown.load(Ordering::SeqCst) { match rx.recv_timeout(std::time::Duration::from_millis(100)) { Ok(work) => { let strategy = work.strategy; @@ -107,7 +107,7 @@ where tx: req_tx, rx: resp_rx, shutdown, - _workers: workers, + workers, } } @@ -146,13 +146,16 @@ where /// Signal all workers to shut down gracefully. pub fn shutdown(&self) { - self.shutdown.store(true, Ordering::Relaxed); + self.shutdown.store(true, Ordering::SeqCst); } } impl Drop for BackgroundWorkerPool { fn drop(&mut self) { - self.shutdown.store(true, Ordering::Relaxed); - // Workers will exit on next recv_timeout cycle (100ms max) + self.shutdown.store(true, Ordering::SeqCst); + // Join all worker threads — wait for clean exit (100ms max per thread) + for handle in self.workers.drain(..) { + let _ = handle.join(); + } } } diff --git a/server/src/workers/stubs.rs b/server/src/workers/stubs.rs index 5149c2e06..626c44c52 100644 --- a/server/src/workers/stubs.rs +++ b/server/src/workers/stubs.rs @@ -8,7 +8,7 @@ use bevy_ecs::prelude::*; -use super::pool::{BackgroundWorkerPool, DeliveryStrategy, WorkRequest}; +use super::pool::BackgroundWorkerPool; // --------------------------------------------------------------------------- // ChunkGen — terrain, buildings, props, navmesh for a world chunk @@ -53,17 +53,6 @@ impl ChunkGenWorker { }), } } - - pub fn submit(&self, req: ChunkGenRequest, strategy: DeliveryStrategy) { - self.pool.submit(WorkRequest { - payload: req, - strategy, - }); - } - - pub fn drain(&self) -> Vec<(ChunkGenResult, DeliveryStrategy)> { - self.pool.drain() - } } // --------------------------------------------------------------------------- diff --git a/server/src/workers/systems.rs b/server/src/workers/systems.rs index e877f00c7..35d6c967c 100644 --- a/server/src/workers/systems.rs +++ b/server/src/workers/systems.rs @@ -12,16 +12,25 @@ use super::stubs::{ChunkGenWorker, NpcPrepWorker, OffscreenTickWorker}; /// PreInput: poll all worker pools for completed results. /// -/// Incorporates results into the world state. Currently logs completions -/// at debug level — real incorporation logic (spawning chunks, activating -/// NPCs) plugs in at Phase 5+. +/// Currently logs completions at debug level — real incorporation logic +/// (spawning chunks, activating NPCs) plugs in at Phase 5+. +/// +/// **DETERMINISM GAP (D-010):** Results are incorporated at the tick they +/// happen to be ready, not at a deterministic tick. Two runs with the same +/// seed may incorporate a chunk at tick 500 vs tick 502 depending on CPU +/// load — diverging world state from that point. +/// +/// This is acceptable while all handlers are no-ops (no simulation state +/// is modified). Before any handler produces real state changes, implement +/// a `WorkerResultLog` resource that records `(tick, worker_type, result)`. +/// Replay reads from the log instead of polling channels. See plan #843. pub fn poll_worker_results( chunk_worker: Option>, npc_worker: Option>, offscreen_worker: Option>, ) { if let Some(worker) = chunk_worker { - for (result, _strategy) in worker.drain() { + for (result, _strategy) in worker.pool.drain() { tracing::debug!( chunk_x = result.chunk_x, chunk_y = result.chunk_y,