diff --git a/server/src/atlas/district_mix.rs b/server/src/atlas/district_mix.rs index 7577d82bc..a6d3e5504 100644 --- a/server/src/atlas/district_mix.rs +++ b/server/src/atlas/district_mix.rs @@ -197,7 +197,7 @@ pub fn compute_district_mix( // We avoid f32 by using integer weighted random selection. let weight_sum: u32 = weights.iter().sum(); let mut counts: [u32; 9] = [0; 9]; - let mut lcg = AtlasRng::new(seed); + let mut lcg = AtlasRng::new(seed.wrapping_add(1)); for _ in 0..total_districts { let mut pick = lcg.next_u32() % weight_sum; diff --git a/server/src/atlas/gen_queue.rs b/server/src/atlas/gen_queue.rs index 5268e6e4b..95e99c563 100644 --- a/server/src/atlas/gen_queue.rs +++ b/server/src/atlas/gen_queue.rs @@ -113,10 +113,10 @@ struct QueuedWork { /// Submit work with `submit()`. Drain completions with `drain_completions()` /// once per tick. The Rayon thread pool runs tasks in priority order. /// -/// Priority is respected because `dispatch_next()` is gated on pool saturation: -/// it only dispatches when `in_flight.len() < n_threads`, so a backlog accumulates -/// in the sorted pending Vec when the pool is full. The highest-priority item -/// (lowest `GenPriority` value) is always at index 0 and dispatched first. +/// Priority is respected because `dispatch_next()` is gated on pool saturation +/// via `in_flight_count`: it only dispatches when fewer than `n_threads` tasks +/// are running. This applies to all work item types — `in_flight` (body-id set) +/// is only for AnalyzeBody dedup; `in_flight_count` is the general saturation gate. #[derive(Resource)] pub struct GenerationQueue { /// Pending work items, sorted by priority (index 0 = highest priority). @@ -126,8 +126,11 @@ pub struct GenerationQueue { completion_rx: Receiver, /// Rayon thread pool dedicated to generation work. pool: rayon::ThreadPool, - /// Set of body_ids currently in-flight to avoid duplicate submissions. + /// Set of body_ids currently in-flight — used only for AnalyzeBody dedup. in_flight: Arc>>, + /// Count of all work items currently executing in the Rayon pool. + /// This is the saturation gate — all work item types increment/decrement it. + in_flight_count: Arc>, /// Thread count — caps concurrent dispatches so pending items accumulate /// and priority ordering is consulted before the pool has free threads. n_threads: usize, @@ -168,6 +171,7 @@ impl GenerationQueue { completion_rx: rx, pool, in_flight: Arc::new(Mutex::new(std::collections::BTreeSet::new())), + in_flight_count: Arc::new(Mutex::new(0)), n_threads, } } @@ -228,16 +232,16 @@ impl GenerationQueue { // Dispatch the highest-priority pending item to the Rayon pool. // - // Gated on pool saturation: only dispatches when in_flight.len() < n_threads. - // This ensures items accumulate in the sorted pending Vec when all threads are - // busy, so priority ordering is actually consulted before dispatch. + // Gated on in_flight_count < n_threads — applies to all work item types, + // not just AnalyzeBody. When the pool is full, items stay in the sorted + // pending Vec so priority ordering is consulted on the next free slot. fn dispatch_next(&self) { let item = { - let in_flight = self.in_flight.lock().unwrap(); - if in_flight.len() >= self.n_threads { + let count = self.in_flight_count.lock().unwrap(); + if *count >= self.n_threads { return; } - drop(in_flight); + drop(count); let mut pending = self.pending.lock().unwrap(); if pending.is_empty() { @@ -246,29 +250,28 @@ impl GenerationQueue { pending.remove(0).item }; - // Mark body as in-flight. + // Mark body as in-flight (AnalyzeBody dedup). if let Some(body_id) = item.body_id() { self.in_flight.lock().unwrap().insert(body_id.to_string()); } + // Increment general in-flight counter for all item types. + *self.in_flight_count.lock().unwrap() += 1; let tx = self.completion_tx.clone(); let in_flight = Arc::clone(&self.in_flight); - let pending = Arc::clone(&self.pending); + let in_flight_count = Arc::clone(&self.in_flight_count); self.pool.spawn(move || { let completion = run_work_item(&item); - // Un-mark in-flight. + // Un-mark body dedup set (AnalyzeBody only). if let Some(body_id) = item.body_id() { in_flight.lock().unwrap().remove(body_id); } + // Decrement general counter for all item types. + *in_flight_count.lock().unwrap() -= 1; let _ = tx.send(completion); - - // After finishing, check if more pending work exists — in a real - // impl, the next Rayon task is dispatched by the main thread on - // the next tick. We don't self-recurse here to avoid pool saturation. - let _ = pending; // keep Arc alive until task exits }); } } @@ -363,7 +366,8 @@ mod tests { #[test] fn priority_ordering() { // Submit three items rapidly; Immediate should be dispatched first. - let q = make_queue(); + // Uses 3 threads so all items can dispatch without hitting saturation. + let q = GenerationQueue::with_threads(3); // Using GenerateSkeleton (no dedup logic) to test ordering directly. q.submit( GenWorkItem::GenerateSkeleton { city_id: 1 }, @@ -384,37 +388,43 @@ mod tests { #[test] fn priority_ordering_respected_under_saturation() { - // Single-thread queue: pool saturates after 1 dispatch, so remaining - // items queue up and are dispatched in priority order. + // Single-thread queue: in_flight_count saturates at 1, so the second + // item stays in the pending Vec and is dispatched in priority order. + // Uses AnalyzeBody (distinct body_ids) so all paths — dedup set AND + // in_flight_count — are exercised. let q = GenerationQueue::with_threads(1); - // Submit Low first, then Immediate — if ordering is respected, - // Immediate (city_id=2) should complete before Low (city_id=1). + // Submit Low first, then Immediate. With 1 thread: + // - "BodyA" (Low) dispatches immediately (pool empty). + // - "BodyB" (Immediate) is inserted at index 0 of the sorted pending + // Vec while "BodyA" is in-flight (in_flight_count = 1 = n_threads). + // - When "BodyA" completes, drain_completions() calls dispatch_next() + // which picks index 0 = "BodyB" (Immediate). q.submit( - GenWorkItem::GenerateSkeleton { city_id: 1 }, + GenWorkItem::AnalyzeBody { + body_id: "BodyA".to_string(), + }, GenPriority::Low, ); q.submit( - GenWorkItem::GenerateSkeleton { city_id: 2 }, + GenWorkItem::AnalyzeBody { + body_id: "BodyB".to_string(), + }, GenPriority::Immediate, ); - // Wait for both to complete. - std::thread::sleep(Duration::from_millis(200)); - let completions = q.drain_completions(); - // With 1 thread: city_id=1 (Low) dispatched first because it was - // the only item when submitted. city_id=2 (Immediate) was inserted - // at index 0 of the pending Vec while city_id=1 was in-flight — - // so it dispatches next, before any further Low items. - assert_eq!(completions.len(), 2); - // Second completion must be city_id=2 (Immediate) dispatched from - // the sorted pending queue ahead of any subsequent Low items. - assert!(matches!( - &completions[0], - GenCompletion::SkeletonGenerated { city_id: 1 } - )); - assert!(matches!( - &completions[1], - GenCompletion::SkeletonGenerated { city_id: 2 } - )); + // Wait for BodyA to complete. + std::thread::sleep(Duration::from_millis(50)); + // drain_completions dispatches BodyB (Immediate, index 0 of pending). + let first = q.drain_completions(); + // Wait for BodyB to complete. + std::thread::sleep(Duration::from_millis(50)); + let second = q.drain_completions(); + + assert_eq!(first.len(), 1); + assert_eq!(second.len(), 1); + assert!(matches!(&first[0], GenCompletion::BodyAnalyzed { body_id } if body_id == "BodyA")); + assert!( + matches!(&second[0], GenCompletion::BodyAnalyzed { body_id } if body_id == "BodyB") + ); } #[test] diff --git a/server/src/atlas/rng.rs b/server/src/atlas/rng.rs index 42b87a95a..1d90c4d0e 100644 --- a/server/src/atlas/rng.rs +++ b/server/src/atlas/rng.rs @@ -3,10 +3,7 @@ //! Shared by all atlas generation modules that need seeded randomness. //! Uses Knuth's LCG parameters — integer-only arithmetic, no f32, D-010 compliant. //! -//! Two construction options: -//! - [`AtlasRng::new`] — adds 1 to seed (district_mix convention). -//! - [`AtlasRng::new_mixed`] — mixes seed with a Fibonacci hash constant to -//! avoid degenerate state at 0 (skeleton_gen convention). +//! Callers are responsible for any seed pre-mixing before calling `AtlasRng::new`. /// Seeded linear congruential generator (D-010). pub struct AtlasRng { @@ -14,18 +11,12 @@ pub struct AtlasRng { } impl AtlasRng { - /// Seed by adding 1 — matches the district_mix convention. + /// Create a new RNG from a pre-mixed seed. + /// + /// Callers must ensure the seed is non-degenerate (avoid passing 0 directly + /// if the seed could realistically be 0 — add a constant before calling). pub fn new(seed: u64) -> Self { - Self { - state: seed.wrapping_add(1), - } - } - - /// Seed with Fibonacci hash mix — avoids degenerate state at seed=0. - pub fn new_mixed(seed: u64) -> Self { - Self { - state: seed.wrapping_add(0x9e37_79b9_7f4a_7c15), - } + Self { state: seed } } fn next_u64(&mut self) -> u64 { @@ -63,12 +54,4 @@ mod tests { let vals_b: Vec = (0..10).map(|_| b.next_u32()).collect(); assert_ne!(vals_a, vals_b); } - - #[test] - fn mixed_seed_avoids_zero_state() { - let mut r = AtlasRng::new_mixed(0); - // Should not produce a constant zero sequence. - let vals: Vec = (0..5).map(|_| r.next_u32()).collect(); - assert!(vals.iter().any(|&v| v != 0)); - } } diff --git a/server/src/atlas/skeleton_gen.rs b/server/src/atlas/skeleton_gen.rs index d07c60ca0..cb5009f71 100644 --- a/server/src/atlas/skeleton_gen.rs +++ b/server/src/atlas/skeleton_gen.rs @@ -245,7 +245,7 @@ fn derive_layout_mode( /// scaled to the ±16 sim tile maximum from `block_irregularity::max_offset_sim_tiles`. fn organic_placements(irregularity: f32, seed: u64) -> [[BlockPlacement; 4]; 4] { let max_offset = (irregularity * 16.0) as i16; - let mut lcg = AtlasRng::new_mixed(seed); + let mut lcg = AtlasRng::new(seed.wrapping_add(0x9e37_79b9_7f4a_7c15)); // Build the 2D array using a flat closure to keep things readable. let mut flat: [BlockPlacement; 16] = core::array::from_fn(|_| BlockPlacement { diff --git a/tooling/planet-gen/import_province_boundaries.py b/tooling/planet-gen/import_province_boundaries.py index fa348cebc..3da0f25f6 100644 --- a/tooling/planet-gen/import_province_boundaries.py +++ b/tooling/planet-gen/import_province_boundaries.py @@ -444,7 +444,7 @@ def import_body_provinces( print(f" {body_id}: {len(basins)} basins — {', '.join(areas)}") if not dry_run: - with conn: # per-body savepoint: rolls back this body on exception, keeps prior commits + with conn: # per-body transaction (BEGIN/COMMIT): rolls back this body on exception, keeps prior commits conn.execute( "DELETE FROM atlas_province_boundaries WHERE body_id = ?", (body_id,),