jpmschweitzer
3e87fd5b4f
T-1151: build_district_window_layer dispatches one Rayon task per row (pure derive_window_cell via derive_at_metres), scattered row-major into the flat arrays; a cfg(test) serial path backs the bit-identical parallel-vs-serial golden. T-1150: serde-default window_granularity (1=district, 4=quarter) + window_min_wl_m on AtlasLayerRequest — additive, no sixth demux shape, old frames decode unchanged (tested). Quarter mode = full reclassification at 512m spacing over the same world rect ((4n)x(4n) cells); WIRE_CAP_CELLS=4096 enforces n*granularity <= cap (quarter clamps n to 16, the design doc's worked example). Granularity + min_wl key ALL five touch points: DistrictWindowLayer echo, server FIFO-256 cache key (now a 5-tuple), per-connection coalescing key, client request codec (omitted-at-default wire fields), client LRU key. Mandatory aliasing regressions on both ends: identical (body, center, n) at granularity 1 vs 4 produce distinct cache entries and correct per-granularity payload shapes (server, 3-thread queue to avoid the AnalyzeBody thread contention found while writing it) and distinct client cache keys (gdUnit). Replay fixture regenerated — the layer struct grew two echoed fields (231->254 bytes, content verified). Client requests stay district-granularity by default — quarter requests arrive with T-1153's rung selection.
1430 lines
61 KiB
Rust
1430 lines
61 KiB
Rust
//! Background generation queue — prioritized Rayon thread pool (D-206).
|
||
//!
|
||
//! All runtime-background generation work runs through this queue. The main
|
||
//! tick thread submits work items (non-blocking) and drains completion events
|
||
//! once per tick via a `crossbeam` channel.
|
||
//!
|
||
//! **Priority levels (D-206):**
|
||
//! - `Immediate`: player arrives within 1 game-minute. Runs first.
|
||
//! - `High`: player arrives within 5 game-minutes.
|
||
//! - `Medium`: player is in the same system.
|
||
//! - `Low`: player has heard of this location via NPC/news.
|
||
//!
|
||
//! **Work item types (D-206):**
|
||
//! - `AnalyzeBody`: D8 drainage + attractor extraction for a body.
|
||
//! - `GenerateSkeleton`: Phase 1 QuarterSkeleton for a city.
|
||
//! - `FillChunk`: Phase 2 chunk fill for a pre-loaded quarter.
|
||
//! - `DeriveWindow`: District-resolution window derive (D-226 T-1124, T-1137).
|
||
//!
|
||
//! Completion events are delivered to the main thread via
|
||
//! `GenerationQueue::drain_completions()`, called once per tick from a Bevy
|
||
//! system in `TickPhase::PreInput`.
|
||
//!
|
||
//! **Thread count (D-206):** `available_parallelism - 2`, minimum 1.
|
||
|
||
use std::path::PathBuf;
|
||
use std::sync::{Arc, Mutex};
|
||
|
||
use bevy_ecs::prelude::Resource;
|
||
use crossbeam_channel::{Receiver, Sender};
|
||
|
||
use crate::atlas::attractor_matching::CityRecord;
|
||
use crate::atlas::body_world_state::BodyWorldState;
|
||
use crate::atlas::cascade::{run_cascade_from_heightmap, CascadeLayer};
|
||
use crate::atlas::district_profile::{BodyParams, ClimateConstants, DistrictPos};
|
||
use crate::atlas::features::TerrainAnalysis;
|
||
use crate::atlas::heightmap::{load_heightmap_png, GRID_H, GRID_W};
|
||
use crate::atlas::layer_proxy::{build_district_window_layer, DistrictWindowLayer};
|
||
use crate::atlas::shell::{fill_chunk, FilledChunk};
|
||
use crate::atlas::skeleton_gen::{assign_all_block_tags, generate_quarter_skeleton};
|
||
use crate::atlas::trait_catalog_reader::ExteriorCatalog;
|
||
use crate::bridge::ConnectionId;
|
||
use crate::seed::SeedChain;
|
||
use crate::simulation::generator::{BuildingPropertyTag, CityGenerationContext, QuarterWorldState};
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Priority
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Work priority levels — lower discriminant = higher priority.
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
|
||
pub enum GenPriority {
|
||
/// Player arrives within ~1 game-minute. Runs before all other levels.
|
||
Immediate = 0,
|
||
/// Player arrives within ~5 game-minutes.
|
||
High = 1,
|
||
/// Player is in the same system.
|
||
Medium = 2,
|
||
/// Player has seen or heard of this location (NPC dialogue, news ticker).
|
||
Low = 3,
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Work item types
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// A unit of background generation work (D-206).
|
||
#[derive(Debug, Clone)]
|
||
pub enum GenWorkItem {
|
||
/// Run the Layer-1 cascade (drainage → features → sub-biome) for this body.
|
||
/// The enqueuer resolves the inputs (D-225): `heightmap_path` is the
|
||
/// mod-resolved source PNG, `body_seed` is this body's SeedChain position.
|
||
/// `run_work_item` is pure compute — it does no path/DB resolution.
|
||
AnalyzeBody {
|
||
body_id: String,
|
||
heightmap_path: PathBuf,
|
||
sea_level: f32,
|
||
body_seed: SeedChain,
|
||
/// The body's settlements (from `atlas_city_names`), pre-resolved at
|
||
/// dispatch time so the cascade stays DB-free. Fed to Layer-3 placement
|
||
/// (#955); empty if the body has no settlements (cascade stops at Layer 1).
|
||
cities: Vec<CityRecord>,
|
||
/// The body's system `dominant_faction` (D-237), pre-resolved at dispatch
|
||
/// time. Drives Layer-3 TerritorialStatus + spatial character (#956,
|
||
/// D-212/214/215). `None` → `FrontierUnclaimed`.
|
||
dominant_faction: Option<String>,
|
||
/// Body physical parameters for the DistrictProfile layer (T-1023, D-239 §1).
|
||
/// Pre-resolved at dispatch time. `None` → district layer skipped for this body.
|
||
/// Boxed: `BodyParams` is large relative to other variants (clippy
|
||
/// large_enum_variant) — boxing keeps `GenWorkItem` compact.
|
||
body_params: Option<Box<BodyParams>>,
|
||
},
|
||
/// Generate a Phase 1 QuarterSkeleton for this city.
|
||
///
|
||
/// `context` is the D-199 economic read-set pre-resolved at dispatch time.
|
||
/// All 6 required fields must be populated before this item is submitted
|
||
/// (D-199: "Missing fields abort the task … generation does not proceed with
|
||
/// partial context").
|
||
///
|
||
/// `body_id` routes the resulting `SkeletonGenerated` completion into the
|
||
/// correct `BodyWorldState` cache entry (D-230).
|
||
///
|
||
/// `quarter_id` is the stable content-addressable id for the generated
|
||
/// quarter (keyed by city position + world seed).
|
||
///
|
||
/// `economic_role`, `population`, and `founding_age_years` are D-199 fields
|
||
/// carried alongside the context because `generate_quarter_skeleton` accepts them as
|
||
/// separate parameters (its signature is not changed by this ticket).
|
||
GenerateSkeleton {
|
||
city_id: u64,
|
||
body_id: String,
|
||
/// D-199 economic read-set + all other context fields.
|
||
context: Box<CityGenerationContext>,
|
||
/// Stable content-addressable quarter id (D-194/D-230).
|
||
quarter_id: u64,
|
||
/// Quarter-level seed chain (D-224).
|
||
chain: SeedChain,
|
||
// D-199 raw fields passed to generate_quarter_skeleton separately.
|
||
economic_role: String,
|
||
population: i64,
|
||
founding_age_years: u32,
|
||
/// D-235 exterior-grammar content (T-988): the trait-template catalog's
|
||
/// `visual_bundle`s plus the two sibling content tables
|
||
/// (`architecture_zone_bias`, `color_register_bands`), pre-resolved at
|
||
/// dispatch time — mirrors `context`'s own pre-resolution rationale, so
|
||
/// `assign_block_tags` never touches `systems.db` (T-987/D-230 purity).
|
||
/// A `Vec`/`BTreeMap`-backed struct is a handful of words on the stack
|
||
/// regardless of its contents' size (same reasoning already documented
|
||
/// for `FillChunk.block_tags` below), so this needs no `Box`.
|
||
exterior_catalog: ExteriorCatalog,
|
||
},
|
||
/// Derive the building shell for one 64 m chunk of an existing quarter
|
||
/// (D-230 derive phase, T-987).
|
||
///
|
||
/// The covering block's `block_tags` are **pre-resolved into the item** at enqueue
|
||
/// time because `run_work_item` is cache-free (mirrors `GenerateSkeleton`). A 64 m
|
||
/// chunk lies wholly within one 128 m block and footprints are block-confined, so
|
||
/// the covering block's tags are exactly the relevant set. Build items with
|
||
/// [`build_fill_chunk_item`] — the D-230 "skeleton not yet processed → re-enqueue
|
||
/// at `High`" precondition is the caller's cache lookup, which only reaches this
|
||
/// constructor once the `QuarterWorldState` exists.
|
||
FillChunk {
|
||
/// Stable id of the quarter being filled (D-194/D-230).
|
||
quarter_id: u64,
|
||
/// Block grid position within the quarter (0..4, 0..4).
|
||
block_pos: (u8, u8),
|
||
/// Sub-chunk quadrant within the block (0..2, 0..2) — a block is 2×2 chunks.
|
||
sub_chunk: (u8, u8),
|
||
/// Covering block's building tags, pre-resolved from the cached
|
||
/// `QuarterWorldState`. Empty for an open/un-built block (→ empty shell).
|
||
/// A `Vec` is three words on the stack regardless of element size (the tags
|
||
/// live behind the pointer), so — unlike `AnalyzeBody`'s boxed `BodyParams` —
|
||
/// this variant needs no `Box` to stay clippy `large_enum_variant`-clean.
|
||
block_tags: Vec<BuildingPropertyTag>,
|
||
},
|
||
/// Derive a district-resolution window (D-226 T-1124 amendment, T-1137).
|
||
///
|
||
/// **Binding serving model:** window derives ride this SAME Rayon queue as
|
||
/// every other expensive atlas path — never inline on the `PreInput` drain
|
||
/// (the amendment's §1 is explicit: a window derive at n=32/64 is
|
||
/// ~7–29 ms, which would blow the "cheap channel drain" contract
|
||
/// `drain_generation_completions` documents).
|
||
///
|
||
/// **TerrainAnalysis availability (T-1137 binding decision, with numbers;
|
||
/// corrected 2026-07-21 per PR #187 review — Tyre C1):** `BodyWorldState`
|
||
/// does NOT retain `TerrainAnalysis` after cascade completion (T-1044
|
||
/// scoped its transient-carry fix to *within-cascade* reuse only —
|
||
/// `cascade.rs` drops it once `DistrictProfile`+`RoadGraph` finish; see the
|
||
/// doc on `CascadeSnapshot::terrain_analysis`). Caching it alongside every
|
||
/// `BodyWorldStateCache` entry would cost ~2 MB × 50-body capacity ≈
|
||
/// 100 MB of PERMANENT resident cost, paid by every cached body whether or
|
||
/// not a window is ever requested for it — the exact D-203 budget concern
|
||
/// T-1044's own ticket text guarded against. So `run_work_item` re-derives
|
||
/// via `run_layer1` (matching `aliveness_probe`'s existing `--render`
|
||
/// workaround) rather than persisting a field on `BodyWorldState` — but
|
||
/// NOT unconditionally on every work item: the actual model is a small
|
||
/// **per-body LRU** (`TerrainAnalysisCache`, capacity 8, ~16 MB worst
|
||
/// case), consulted before every re-derive. The FIRST `DeriveWindow` on a
|
||
/// body pays the full ~45 ms `run_layer1` cost and populates that body's
|
||
/// cache entry; EVERY SUBSEQUENT window on the SAME body (any `center`/`n`,
|
||
/// not just an exact repeat — that narrower case is what
|
||
/// `DistrictWindowCache` in `layer_proxy.rs` already catches) hits the LRU
|
||
/// and skips straight to the ~7–29 ms per-window pack in
|
||
/// `build_district_window_layer`. An LRU eviction re-pays the ~45 ms on the
|
||
/// next window for that body. Because every `DeriveWindow` work item still
|
||
/// runs off the tick thread on the Rayon queue regardless of hit or miss,
|
||
/// both costs are invisible to the main thread either way — the LRU's
|
||
/// value is throughput/worker-occupancy (bounding how many ~45 ms re-derives
|
||
/// a pan-burst across one body can force), not tick-thread latency.
|
||
DeriveWindow {
|
||
body_id: String,
|
||
/// Coalescing/routing key (D-226 T-1124 amendment §1 "recommended"
|
||
/// per-connection coalescing) — NOT used by `run_work_item` itself
|
||
/// (the derive is connection-agnostic), only by
|
||
/// `GenerationQueue::submit_window` to decide which still-pending item
|
||
/// a new one for the same connection+body supersedes.
|
||
conn_id: ConnectionId,
|
||
/// Mod-resolved source heightmap PNG (mirrors `AnalyzeBody`).
|
||
heightmap_path: PathBuf,
|
||
sea_level: f32,
|
||
/// This body's `SeedChain` position — `derive_district`'s seed input.
|
||
body_seed: SeedChain,
|
||
/// Body physical parameters. Boxed for the same `large_enum_variant`
|
||
/// reason `AnalyzeBody.body_params` is boxed.
|
||
body_params: Box<BodyParams>,
|
||
/// Window centre + side length in districts. `n` is ALREADY clamped to
|
||
/// `[1, DISTRICT_WINDOW_MAX_N]` AND the granularity-aware
|
||
/// `WIRE_CAP_CELLS` ceiling by the caller (`handle_atlas_request`)
|
||
/// before this item is built — never trusted from the wire again here.
|
||
center: DistrictPos,
|
||
n: u32,
|
||
/// Derivation granularity (T-1150) — `WINDOW_GRANULARITY_DISTRICT` (1)
|
||
/// or `WINDOW_GRANULARITY_QUARTER` (4). Already resolved via
|
||
/// `resolve_window_granularity` by the caller.
|
||
granularity: u32,
|
||
/// Octave cutoff in whole metres (T-1149/T-1150), `0` = no cutoff.
|
||
min_wl_m: u32,
|
||
},
|
||
}
|
||
|
||
impl GenWorkItem {
|
||
pub fn body_id(&self) -> Option<&str> {
|
||
match self {
|
||
GenWorkItem::AnalyzeBody { body_id, .. } => Some(body_id),
|
||
_ => None,
|
||
}
|
||
}
|
||
|
||
/// Coalescing key for `DeriveWindow` items only — `(connection, body,
|
||
/// granularity)` (T-1150, design doc §3 [SOFT] recommendation, extending
|
||
/// T-1137's `(connection, body)`). `granularity` is part of the key so an
|
||
/// in-flight district-spacing (granularity 1) pan-burst is never
|
||
/// superseded by an unrelated quarter-spacing (granularity 4) request for
|
||
/// the same connection+body, and vice versa — the two rungs are separate
|
||
/// in-flight derives, not competing updates to the same one.
|
||
/// `None` for every other variant (they don't coalesce this way).
|
||
pub fn window_supersede_key(&self) -> Option<(ConnectionId, &str, u32)> {
|
||
if let GenWorkItem::DeriveWindow {
|
||
body_id,
|
||
conn_id,
|
||
granularity,
|
||
..
|
||
} = self
|
||
{
|
||
Some((*conn_id, body_id, *granularity))
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Completion event
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Sent back to the main thread when a work item finishes (D-206).
|
||
#[derive(Debug)]
|
||
pub enum GenCompletion {
|
||
BodyAnalyzed {
|
||
body_id: String,
|
||
/// The computed world state, ready for `BodyWorldStateCache::insert`.
|
||
state: BodyWorldState,
|
||
},
|
||
SkeletonGenerated {
|
||
city_id: u64,
|
||
/// The body this skeleton belongs to — used to route state into
|
||
/// `BodyWorldState.quarters` (D-230).
|
||
body_id: String,
|
||
/// District-level world state (skeleton + block tags) produced by the plan phase (D-230).
|
||
/// Boxed to keep `GenCompletion` variant sizes balanced (D-230 skeleton is ~2.7 KB).
|
||
state: Box<QuarterWorldState>,
|
||
},
|
||
ChunkFilled {
|
||
/// The derived shell for this chunk (D-230, T-987). Sparse — only the
|
||
/// non-`Void` shell voxels. Carries its own quarter/block/sub-chunk address.
|
||
/// Boxed to keep `GenCompletion` variant sizes balanced.
|
||
filled: Box<FilledChunk>,
|
||
},
|
||
/// A district window finished deriving (D-226 T-1124 amendment, T-1137).
|
||
/// The main thread inserts `layer` into the `DistrictWindowCache` keyed by
|
||
/// `(body_id, layer.center, layer.n)` — NOT pushed directly into any
|
||
/// in-flight response (the window's requester re-polls per the existing
|
||
/// D-225 loop and hits the now-populated cache on its next request; see
|
||
/// `handle_atlas_request`'s window branch).
|
||
WindowDerived {
|
||
body_id: String,
|
||
/// Boxed to keep `GenCompletion` variant sizes balanced (six
|
||
/// `Vec`s — comparable to `SkeletonGenerated`/`ChunkFilled`'s own
|
||
/// boxing rationale).
|
||
layer: Box<DistrictWindowLayer>,
|
||
},
|
||
/// Work item failed — body_id or city_id for logging.
|
||
Failed { item: GenWorkItem, reason: String },
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Internal queued work
|
||
// ---------------------------------------------------------------------------
|
||
|
||
struct QueuedWork {
|
||
priority: GenPriority,
|
||
item: GenWorkItem,
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// GenerationQueue — Bevy Resource
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Bevy `Resource` managing the background generation queue (D-206).
|
||
///
|
||
/// Submit work with `submit()`. Drain completions with `drain_completions()`
|
||
/// once per tick. The Rayon thread pool runs tasks in priority order.
|
||
///
|
||
/// Also owns the queue-scoped `TerrainAnalysisCache` (T-1137, PR #187 review
|
||
/// C1) — a small per-body LRU consulted by every `DeriveWindow` work item
|
||
/// before paying the `run_layer1` re-derive cost. Lives here (not as a
|
||
/// separate `Resource`) because it must be reachable from `run_work_item`
|
||
/// while it executes on a Rayon worker thread, the same reason
|
||
/// `in_flight`/`in_flight_count` are `Arc<Mutex<_>>` fields on this struct
|
||
/// rather than plain fields.
|
||
///
|
||
/// 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).
|
||
pending: Arc<Mutex<Vec<QueuedWork>>>,
|
||
/// Completions channel — background tasks send here; main thread reads.
|
||
completion_tx: Sender<GenCompletion>,
|
||
completion_rx: Receiver<GenCompletion>,
|
||
/// Rayon thread pool dedicated to generation work.
|
||
pool: rayon::ThreadPool,
|
||
/// Set of body_ids currently in-flight — used only for AnalyzeBody dedup.
|
||
in_flight: Arc<Mutex<std::collections::BTreeSet<String>>>,
|
||
/// 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<Mutex<usize>>,
|
||
/// Thread count — caps concurrent dispatches so pending items accumulate
|
||
/// and priority ordering is consulted before the pool has free threads.
|
||
n_threads: usize,
|
||
/// Per-body `TerrainAnalysis` LRU shared by every `DeriveWindow` work item
|
||
/// on this queue (T-1137, PR #187 review C1) — see [`TerrainAnalysisCache`].
|
||
terrain_cache: Arc<Mutex<TerrainAnalysisCache>>,
|
||
}
|
||
|
||
impl std::fmt::Debug for GenerationQueue {
|
||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||
let pending_len = self.pending.lock().map(|p| p.len()).unwrap_or(0);
|
||
f.debug_struct("GenerationQueue")
|
||
.field("pending_count", &pending_len)
|
||
.finish()
|
||
}
|
||
}
|
||
|
||
impl GenerationQueue {
|
||
/// Create a new queue with the D-206 thread count:
|
||
/// `available_parallelism - 2`, minimum 1.
|
||
pub fn new() -> Self {
|
||
let n_threads = std::thread::available_parallelism()
|
||
.map(|p| p.get().saturating_sub(2).max(1))
|
||
.unwrap_or(1);
|
||
Self::with_threads(n_threads)
|
||
}
|
||
|
||
/// Create a queue with a specific thread count (for testing).
|
||
pub fn with_threads(n_threads: usize) -> Self {
|
||
let pool = rayon::ThreadPoolBuilder::new()
|
||
.num_threads(n_threads)
|
||
.thread_name(|i| format!("gen-worker-{i}"))
|
||
.build()
|
||
.expect("failed to build generation rayon pool");
|
||
|
||
let (tx, rx) = crossbeam_channel::unbounded();
|
||
|
||
Self {
|
||
pending: Arc::new(Mutex::new(Vec::new())),
|
||
completion_tx: tx,
|
||
completion_rx: rx,
|
||
pool,
|
||
in_flight: Arc::new(Mutex::new(std::collections::BTreeSet::new())),
|
||
in_flight_count: Arc::new(Mutex::new(0)),
|
||
n_threads,
|
||
terrain_cache: Arc::new(Mutex::new(TerrainAnalysisCache::new(
|
||
TERRAIN_ANALYSIS_CACHE_CAPACITY,
|
||
))),
|
||
}
|
||
}
|
||
|
||
/// Submit a work item at the given priority.
|
||
///
|
||
/// If an `AnalyzeBody` item for the same body_id is already in-flight or
|
||
/// pending, the submission is silently ignored (idempotent).
|
||
pub fn submit(&self, item: GenWorkItem, priority: GenPriority) {
|
||
// Dedup AnalyzeBody submissions.
|
||
if let Some(body_id) = item.body_id() {
|
||
let in_flight = self.in_flight.lock().unwrap();
|
||
if in_flight.contains(body_id) {
|
||
return;
|
||
}
|
||
drop(in_flight);
|
||
// Check pending list.
|
||
let pending = self.pending.lock().unwrap();
|
||
if pending.iter().any(|q| q.item.body_id() == Some(body_id)) {
|
||
return;
|
||
}
|
||
drop(pending);
|
||
}
|
||
|
||
let mut pending = self.pending.lock().unwrap();
|
||
let pos = pending
|
||
.iter()
|
||
.position(|q| q.priority > priority)
|
||
.unwrap_or(pending.len());
|
||
pending.insert(pos, QueuedWork { priority, item });
|
||
drop(pending);
|
||
|
||
self.dispatch_next();
|
||
}
|
||
|
||
/// Submit a `DeriveWindow` item with per-connection coalescing (D-226
|
||
/// T-1124 amendment §1, "recommended"; extended T-1150 to key on
|
||
/// granularity too): if a `DeriveWindow` item for the SAME `(connection,
|
||
/// body, granularity)` is still sitting in the pending queue (not yet
|
||
/// dispatched to a Rayon worker), it is replaced in place by the new one
|
||
/// — a pan-burst that queues several window requests for the same
|
||
/// connection+body+granularity before the first is dispatched collapses
|
||
/// to one derive. A district-spacing and quarter-spacing request for the
|
||
/// same connection+body do NOT coalesce with each other — they're
|
||
/// separate in-flight derives, not competing updates to the same rung.
|
||
///
|
||
/// Deliberately does **not** attempt to cancel an item already dispatched
|
||
/// to a Rayon worker (no cancellation channel exists, and the amendment
|
||
/// does not require it — bounding queue buildup is the goal, not
|
||
/// interrupting in-flight compute). `item` MUST be a `DeriveWindow`
|
||
/// variant; any other variant is submitted via the plain `submit` path
|
||
/// with no coalescing (this method still accepts it for caller
|
||
/// convenience, but the supersede check is a no-op when
|
||
/// `window_supersede_key()` returns `None`).
|
||
pub fn submit_window(&self, item: GenWorkItem, priority: GenPriority) {
|
||
if let Some(key) = item.window_supersede_key() {
|
||
let key = (key.0, key.1.to_string(), key.2);
|
||
let mut pending = self.pending.lock().unwrap();
|
||
pending.retain(|q| {
|
||
q.item
|
||
.window_supersede_key()
|
||
.map(|k| (k.0, k.1.to_string(), k.2) != key)
|
||
.unwrap_or(true)
|
||
});
|
||
let pos = pending
|
||
.iter()
|
||
.position(|q| q.priority > priority)
|
||
.unwrap_or(pending.len());
|
||
pending.insert(pos, QueuedWork { priority, item });
|
||
drop(pending);
|
||
self.dispatch_next();
|
||
} else {
|
||
self.submit(item, priority);
|
||
}
|
||
}
|
||
|
||
/// Drain all completed items from the channel and dispatch pending work.
|
||
///
|
||
/// Call once per tick from the main thread. Returns all completions
|
||
/// available without blocking. After draining, dispatches as many pending
|
||
/// items as there are free thread slots — this is the point where priority
|
||
/// ordering matters, since the pool was saturated when items were submitted.
|
||
pub fn drain_completions(&self) -> Vec<GenCompletion> {
|
||
let mut out = Vec::new();
|
||
while let Ok(c) = self.completion_rx.try_recv() {
|
||
out.push(c);
|
||
}
|
||
// Fill any newly-freed slots.
|
||
for _ in 0..out.len() {
|
||
self.dispatch_next();
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Number of items waiting in the pending queue.
|
||
pub fn pending_count(&self) -> usize {
|
||
self.pending.lock().unwrap().len()
|
||
}
|
||
|
||
// Dispatch the highest-priority pending item to the Rayon pool.
|
||
//
|
||
// 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 count = self.in_flight_count.lock().unwrap();
|
||
if *count >= self.n_threads {
|
||
return;
|
||
}
|
||
drop(count);
|
||
|
||
let mut pending = self.pending.lock().unwrap();
|
||
if pending.is_empty() {
|
||
return;
|
||
}
|
||
pending.remove(0).item
|
||
};
|
||
|
||
// 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 in_flight_count = Arc::clone(&self.in_flight_count);
|
||
let terrain_cache = Arc::clone(&self.terrain_cache);
|
||
|
||
self.pool.spawn(move || {
|
||
let completion = run_work_item(&item, &terrain_cache);
|
||
|
||
// 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);
|
||
});
|
||
}
|
||
}
|
||
|
||
impl Default for GenerationQueue {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// TerrainAnalysisCache (T-1137, PR #187 review — Tyre C1)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Small per-body LRU of re-derived [`TerrainAnalysis`] (~1.5–2 MB/entry),
|
||
/// consulted by the `DeriveWindow` execution path before paying the ~45 ms
|
||
/// `run_layer1` re-derive cost (T-1137 binding decision — see the
|
||
/// `DeriveWindow` variant doc on why `TerrainAnalysis` is re-derived rather
|
||
/// than cached on `BodyWorldState` at all).
|
||
///
|
||
/// **Why this exists (PR #187 review finding, binding):** the original
|
||
/// T-1137 landing called `run_layer1` unconditionally on every `DeriveWindow`
|
||
/// work item — nothing memoized it within a body, so the *dominant* usage
|
||
/// pattern (panning around ONE body, many windows) paid the ~45 ms re-derive
|
||
/// on every single window instead of just the first. This cache closes that
|
||
/// gap: first window on a body pays the full re-derive and populates the
|
||
/// entry; every subsequent window on the SAME body (until eviction) hits the
|
||
/// cache and skips straight to the ~7–29 ms `build_district_window_layer`
|
||
/// pack (§4's actual per-window number).
|
||
///
|
||
/// **Shape:** `Arc<Mutex<...>>` — lives on [`GenerationQueue`] alongside
|
||
/// `in_flight`/`in_flight_count` (the same "shared state cloned into every
|
||
/// Rayon closure" pattern), because `run_work_item` executes ON a Rayon
|
||
/// worker thread, potentially concurrently with other workers when
|
||
/// `n_threads > 1`; a queue-owned, plain (non-`Resource`) cache is the
|
||
/// correct home — `handle_atlas_request`'s `DistrictWindowCache` is main-
|
||
/// thread-only (D-225 poll loop) and cannot be reused here.
|
||
///
|
||
/// **Capacity (8, ~16 MB worst case):** deliberately small relative to
|
||
/// `BodyWorldStateCache::CACHE_CAPACITY` (50) — this caches a re-derive
|
||
/// shortcut for bodies actually being window-browsed RIGHT NOW, not a
|
||
/// body-indexed store meant to grow with session length. True LRU
|
||
/// (access-recency, mirroring `BodyWorldStateCache`'s own eviction policy)
|
||
/// rather than `DistrictWindowCache`'s FIFO-by-insertion: unlike a
|
||
/// D-227-pure derived window (valid forever, no recency signal to track),
|
||
/// which body a player keeps panning around IS a recency signal, so
|
||
/// access-order eviction is the right fit here.
|
||
#[derive(Debug)]
|
||
struct TerrainAnalysisCache {
|
||
entries: std::collections::BTreeMap<String, (TerrainAnalysis, u64)>,
|
||
/// Monotonic access counter (substitutes for `BodyWorldStateCache`'s
|
||
/// `SimTick` — there is no tick concept on a background Rayon thread).
|
||
clock: u64,
|
||
capacity: usize,
|
||
}
|
||
|
||
/// Default capacity for [`TerrainAnalysisCache`] (PR #187 review — Tyre C1
|
||
/// binding numbers: "capacity ~8, ~2MB/entry = ~16MB worst case").
|
||
const TERRAIN_ANALYSIS_CACHE_CAPACITY: usize = 8;
|
||
|
||
impl TerrainAnalysisCache {
|
||
fn new(capacity: usize) -> Self {
|
||
Self {
|
||
entries: std::collections::BTreeMap::new(),
|
||
clock: 0,
|
||
capacity,
|
||
}
|
||
}
|
||
|
||
/// Look up a cached `TerrainAnalysis` for `body_id`, re-deriving via
|
||
/// `run_layer1` on a miss and inserting the result (evicting the LRU
|
||
/// entry first if at capacity). Bumps the access clock on both a hit and
|
||
/// a fresh insert (both are "this body was just used").
|
||
fn get_or_derive(
|
||
&mut self,
|
||
body_id: &str,
|
||
heightmap: &crate::atlas::heightmap::BodyHeightmap,
|
||
) -> TerrainAnalysis {
|
||
self.clock += 1;
|
||
let now = self.clock;
|
||
if let Some((ta, last_used)) = self.entries.get_mut(body_id) {
|
||
*last_used = now;
|
||
return ta.clone();
|
||
}
|
||
|
||
let (_, ta) = crate::atlas::layer1::run_layer1(heightmap);
|
||
|
||
if self.entries.len() >= self.capacity && !self.entries.contains_key(body_id) {
|
||
if let Some(victim) = self
|
||
.entries
|
||
.iter()
|
||
.min_by_key(|(_, (_, last_used))| *last_used)
|
||
.map(|(id, _)| id.clone())
|
||
{
|
||
self.entries.remove(&victim);
|
||
}
|
||
}
|
||
self.entries.insert(body_id.to_string(), (ta.clone(), now));
|
||
ta
|
||
}
|
||
|
||
#[cfg(test)]
|
||
fn len(&self) -> usize {
|
||
self.entries.len()
|
||
}
|
||
|
||
#[cfg(test)]
|
||
fn contains(&self, body_id: &str) -> bool {
|
||
self.entries.contains_key(body_id)
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Work execution stub
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Execute one work item. This is the Rayon task body (off the tick thread).
|
||
///
|
||
/// `AnalyzeBody` runs the real Layer-1 cascade (#968, D-225); `GenerateSkeleton`
|
||
/// runs the real plan phase (#957, D-229) producing the skeleton + block tags;
|
||
/// `FillChunk` runs the real derive phase (T-987, D-230) producing the building
|
||
/// shell from the pre-resolved tags.
|
||
///
|
||
/// `terrain_cache` serves `DeriveWindow`'s `TerrainAnalysis` re-derive
|
||
/// shortcut (T-1137, PR #187 review C1) — unused by every other variant
|
||
/// (they don't touch `TerrainAnalysis` at all, or — `AnalyzeBody` — derive it
|
||
/// once already as part of the normal in-cascade path, T-1044).
|
||
fn run_work_item(
|
||
item: &GenWorkItem,
|
||
terrain_cache: &Arc<Mutex<TerrainAnalysisCache>>,
|
||
) -> GenCompletion {
|
||
match item {
|
||
GenWorkItem::AnalyzeBody {
|
||
body_id,
|
||
heightmap_path,
|
||
sea_level,
|
||
body_seed,
|
||
cities,
|
||
dominant_faction,
|
||
body_params,
|
||
} => match load_heightmap_png(heightmap_path, body_id, *sea_level) {
|
||
Ok(hm) => {
|
||
// Layer 1 runs at the GRID_W×GRID_H working resolution (D-202):
|
||
// downsample the higher-res stored heightmap first.
|
||
let working = if hm.width > GRID_W || hm.height > GRID_H {
|
||
hm.downsample(GRID_W, GRID_H)
|
||
} else {
|
||
hm
|
||
};
|
||
// Run the full cascade through Region (T-1113), the terminal
|
||
// layer. It subsumes RoadGraph (T-1038), Settlement,
|
||
// DistrictProfile (T-1023), and all prior layers.
|
||
// DistrictProfile and Region derivation both gate on
|
||
// body_params internally (skipped when absent — e.g. a body
|
||
// with no params row), but the road graph needs no body
|
||
// params, so it runs for every analyzed body.
|
||
let up_to = CascadeLayer::Region;
|
||
let snapshot = run_cascade_from_heightmap(
|
||
*body_seed,
|
||
working,
|
||
cities,
|
||
dominant_faction.as_deref(),
|
||
body_params.as_deref(),
|
||
up_to,
|
||
);
|
||
GenCompletion::BodyAnalyzed {
|
||
body_id: body_id.clone(),
|
||
state: snapshot.into_body_world_state(),
|
||
}
|
||
}
|
||
Err(e) => GenCompletion::Failed {
|
||
item: item.clone(),
|
||
reason: format!("heightmap load failed: {e}"),
|
||
},
|
||
},
|
||
GenWorkItem::GenerateSkeleton {
|
||
city_id,
|
||
body_id,
|
||
context,
|
||
quarter_id,
|
||
chain,
|
||
economic_role,
|
||
population,
|
||
founding_age_years,
|
||
exterior_catalog,
|
||
} => {
|
||
// Build the Phase 1 skeleton from the pre-resolved D-199 context.
|
||
// `economic_role`, `population`, and `founding_age_years` are the
|
||
// D-199 raw fields carried alongside the context because
|
||
// `generate_quarter_skeleton` accepts them as separate parameters.
|
||
let skeleton = generate_quarter_skeleton(
|
||
context,
|
||
*population,
|
||
economic_role,
|
||
*quarter_id,
|
||
*founding_age_years,
|
||
*chain,
|
||
);
|
||
// Step-3 building-property tags per footprint (D-229, #957): subdivide
|
||
// each block into building plots and tag them. `exterior_catalog`
|
||
// (D-235, T-988) resolves each tag's BuildingExteriorTag in the
|
||
// same pass.
|
||
let block_tags = assign_all_block_tags(
|
||
&skeleton,
|
||
context,
|
||
economic_role,
|
||
*founding_age_years,
|
||
*chain,
|
||
exterior_catalog,
|
||
);
|
||
GenCompletion::SkeletonGenerated {
|
||
city_id: *city_id,
|
||
body_id: body_id.clone(),
|
||
state: Box::new(QuarterWorldState {
|
||
skeleton,
|
||
block_tags,
|
||
}),
|
||
}
|
||
}
|
||
GenWorkItem::FillChunk {
|
||
quarter_id,
|
||
block_pos,
|
||
sub_chunk,
|
||
block_tags,
|
||
} => {
|
||
// D-230 derive phase: pure rectangle-containment + z-range shell fill over
|
||
// the pre-resolved tags. No cache read here — that is what keeps FillChunk
|
||
// trivially fast and re-derivable (D-227).
|
||
let filled = fill_chunk(*quarter_id, *block_pos, *sub_chunk, block_tags);
|
||
GenCompletion::ChunkFilled {
|
||
filled: Box::new(filled),
|
||
}
|
||
}
|
||
GenWorkItem::DeriveWindow {
|
||
body_id,
|
||
conn_id: _, // routing-only (queue-level coalescing); the derive itself is connection-agnostic
|
||
heightmap_path,
|
||
sea_level,
|
||
body_seed,
|
||
body_params,
|
||
center,
|
||
n,
|
||
granularity,
|
||
min_wl_m,
|
||
} => match load_heightmap_png(heightmap_path, body_id, *sea_level) {
|
||
Ok(hm) => {
|
||
// Same GRID_W×GRID_H downsample AnalyzeBody applies (D-202) — the
|
||
// window derive must run on the SAME working-grid resolution the
|
||
// whole-body cascade uses, or district positions between the two
|
||
// views would disagree (derive_district maps DistrictPos through
|
||
// ta.w/ta.h, T-1137 decision note).
|
||
let working = if hm.width > GRID_W || hm.height > GRID_H {
|
||
hm.downsample(GRID_W, GRID_H)
|
||
} else {
|
||
hm
|
||
};
|
||
// TerrainAnalysis via the per-body LRU (T-1137 binding decision +
|
||
// PR #187 review C1): first window on a body pays the ~45 ms
|
||
// run_layer1 re-derive and populates the cache entry; every
|
||
// subsequent window on the SAME body (until eviction) hits the
|
||
// cache and skips straight to the ~7–29 ms per-window pack below.
|
||
// This is the memoized form of the SAME workaround
|
||
// aliveness_probe --render uses when CascadeSnapshot.terrain_analysis
|
||
// is None (it has no cache — a one-shot CLI run doesn't need one).
|
||
let ta = terrain_cache
|
||
.lock()
|
||
.unwrap()
|
||
.get_or_derive(body_id, &working);
|
||
let climate = ClimateConstants::default();
|
||
let layer = build_district_window_layer(
|
||
*body_seed,
|
||
body_id,
|
||
body_params,
|
||
&ta,
|
||
*center,
|
||
*n,
|
||
&climate,
|
||
*granularity,
|
||
*min_wl_m,
|
||
);
|
||
GenCompletion::WindowDerived {
|
||
body_id: body_id.clone(),
|
||
layer: Box::new(layer),
|
||
}
|
||
}
|
||
Err(e) => GenCompletion::Failed {
|
||
item: item.clone(),
|
||
reason: format!("heightmap load failed: {e}"),
|
||
},
|
||
},
|
||
}
|
||
}
|
||
|
||
/// Build a [`GenWorkItem::FillChunk`] for one 64 m sub-chunk of a quarter, pulling the
|
||
/// covering block's tags out of the cached `QuarterWorldState` (D-230 derive phase, T-987).
|
||
///
|
||
/// Pure (no queue/cache handle), so it unit-tests without a running app. The D-230
|
||
/// precondition — "`FillChunk` is only dispatched after `SkeletonGenerated` for that
|
||
/// district has been processed; if absent, re-enqueue at `High`" — is the caller's
|
||
/// cache lookup: this constructor only runs once the `QuarterWorldState` exists. A
|
||
/// block with no buildings yields empty `block_tags` (→ an empty, terrain-only shell),
|
||
/// which is a valid ready state, not a not-yet-generated one.
|
||
pub fn build_fill_chunk_item(
|
||
quarter: &QuarterWorldState,
|
||
block_pos: (u8, u8),
|
||
sub_chunk: (u8, u8),
|
||
) -> GenWorkItem {
|
||
let block_tags = quarter
|
||
.block_tags
|
||
.get(&block_pos)
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
GenWorkItem::FillChunk {
|
||
quarter_id: quarter.skeleton.quarter_id,
|
||
block_pos,
|
||
sub_chunk,
|
||
block_tags,
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Tests
|
||
// ---------------------------------------------------------------------------
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use std::time::Duration;
|
||
|
||
fn make_queue() -> GenerationQueue {
|
||
GenerationQueue::with_threads(2)
|
||
}
|
||
|
||
/// Write a tiny 16-bit grayscale heightmap PNG to a unique temp path so the
|
||
/// real cascade can run in `run_work_item` without a committed fixture.
|
||
fn test_heightmap_path() -> std::path::PathBuf {
|
||
use std::io::BufWriter;
|
||
use std::sync::atomic::{AtomicU32, Ordering};
|
||
static SEQ: AtomicU32 = AtomicU32::new(0);
|
||
let n = SEQ.fetch_add(1, Ordering::Relaxed);
|
||
let path = std::env::temp_dir().join(format!("sr_genq_{}_{n}.png", std::process::id()));
|
||
let file = std::fs::File::create(&path).expect("create test heightmap");
|
||
let mut enc = png::Encoder::new(BufWriter::new(file), 32, 16);
|
||
enc.set_color(png::ColorType::Grayscale);
|
||
enc.set_depth(png::BitDepth::Sixteen);
|
||
let mut w = enc.write_header().expect("png header");
|
||
let data: Vec<u8> = (0..32u32 * 16)
|
||
.flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
|
||
.collect();
|
||
w.write_image_data(&data).expect("png data");
|
||
path
|
||
}
|
||
|
||
/// Build an `AnalyzeBody` work item pointing at a tiny test heightmap.
|
||
fn analyze(body_id: &str) -> GenWorkItem {
|
||
GenWorkItem::AnalyzeBody {
|
||
body_id: body_id.to_string(),
|
||
heightmap_path: test_heightmap_path(),
|
||
sea_level: 0.3,
|
||
body_seed: SeedChain::for_body(42, body_id),
|
||
cities: vec![],
|
||
dominant_faction: None,
|
||
body_params: None, // T-1023: no body params in queue-mechanic unit tests
|
||
}
|
||
}
|
||
|
||
/// Build a minimal `GenerateSkeleton` work item with a stub context.
|
||
///
|
||
/// The stub context uses Commission/Regional/Urban defaults — the same
|
||
/// values the existing skeleton_gen tests use. These tests exercise queue
|
||
/// mechanics (ordering, saturation, drain), not economic read-set content.
|
||
fn gen_skeleton(city_id: u64) -> GenWorkItem {
|
||
use crate::simulation::generator::{
|
||
BulkClass, CityGenerationContext, FoundingOrientation, MorphologyZone,
|
||
PoliticalArchetype, ProductionUbiquity, SettingType, WorldTier,
|
||
};
|
||
GenWorkItem::GenerateSkeleton {
|
||
city_id,
|
||
body_id: format!("TestBody{city_id}"),
|
||
context: Box::new(CityGenerationContext {
|
||
city_id,
|
||
political_archetype: PoliticalArchetype::Commission,
|
||
prosperity_baseline_bps: 6_000,
|
||
surrounding_biome: SettingType::Urban,
|
||
road_entry_directions: vec![],
|
||
footprint_radius_km: 5.0,
|
||
founding_orientation: FoundingOrientation::Cardinal,
|
||
world_tier: WorldTier::Regional,
|
||
morphology_zone: MorphologyZone::AlluvialPlain,
|
||
trait_selection: vec![],
|
||
dominant_bulk_class: BulkClass::NonPhysical,
|
||
dominant_production_ubiquity: ProductionUbiquity::Common,
|
||
geographic_sector: None,
|
||
body_district_type_mix: vec![],
|
||
settlement_district_pos: (0, 0),
|
||
district_dominant_by_type: Default::default(),
|
||
swerve_rates_bps: (0, 0),
|
||
swerve_foreign_pool: vec![],
|
||
swerve_heritage_pool: vec![],
|
||
}),
|
||
quarter_id: city_id * 10,
|
||
chain: SeedChain::root(42 + city_id),
|
||
economic_role: "service_mixed".to_string(),
|
||
population: 500_000,
|
||
founding_age_years: 200,
|
||
exterior_catalog: ExteriorCatalog::default(),
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn submit_and_drain() {
|
||
let q = make_queue();
|
||
q.submit(analyze("TestBody"), GenPriority::Medium);
|
||
// Give Rayon time to load the heightmap and run the cascade.
|
||
std::thread::sleep(Duration::from_millis(100));
|
||
let completions = q.drain_completions();
|
||
assert_eq!(completions.len(), 1);
|
||
// The work item ran the real cascade and produced a populated state.
|
||
assert!(matches!(
|
||
&completions[0],
|
||
GenCompletion::BodyAnalyzed { body_id, state }
|
||
if body_id == "TestBody"
|
||
&& state.heightmap_width == 32
|
||
&& state.heightmap_height == 16
|
||
));
|
||
}
|
||
|
||
#[test]
|
||
fn dedup_analyze_body() {
|
||
let q = make_queue();
|
||
// Submit the same body twice before it can complete.
|
||
q.submit(analyze("Dup"), GenPriority::Low);
|
||
q.submit(analyze("Dup"), GenPriority::Low);
|
||
std::thread::sleep(Duration::from_millis(50));
|
||
let completions = q.drain_completions();
|
||
// Should have completed exactly once.
|
||
assert_eq!(completions.len(), 1);
|
||
}
|
||
|
||
#[test]
|
||
fn priority_ordering() {
|
||
// Submit three items rapidly; Immediate should be dispatched first.
|
||
// 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(gen_skeleton(1), GenPriority::Low);
|
||
q.submit(gen_skeleton(2), GenPriority::Immediate);
|
||
q.submit(gen_skeleton(3), GenPriority::Medium);
|
||
std::thread::sleep(Duration::from_millis(100));
|
||
let completions = q.drain_completions();
|
||
assert_eq!(completions.len(), 3);
|
||
}
|
||
|
||
#[test]
|
||
fn priority_ordering_respected_under_saturation() {
|
||
// 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. 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(analyze("BodyA"), GenPriority::Low);
|
||
q.submit(analyze("BodyB"), GenPriority::Immediate);
|
||
// 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]
|
||
fn drain_empty_returns_empty() {
|
||
let q = make_queue();
|
||
let result = q.drain_completions();
|
||
assert!(result.is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn pending_count_decreases_after_completion() {
|
||
let q = make_queue();
|
||
q.submit(
|
||
GenWorkItem::FillChunk {
|
||
quarter_id: 99,
|
||
block_pos: (0, 0),
|
||
sub_chunk: (0, 0),
|
||
block_tags: vec![],
|
||
},
|
||
GenPriority::High,
|
||
);
|
||
std::thread::sleep(Duration::from_millis(50));
|
||
let completions = q.drain_completions();
|
||
assert!(!completions.is_empty() || q.pending_count() == 0);
|
||
}
|
||
|
||
/// A `QuarterWorldState` with one building in block (0,0), used to exercise the
|
||
/// full FillChunk path (build item from cached state → Rayon → completion).
|
||
fn quarter_with_one_building() -> QuarterWorldState {
|
||
use crate::atlas::tile_condition::TileCondition;
|
||
use crate::simulation::generator::{
|
||
ArchitectureFlavorRef, BuildingEntryClass, BuildingExteriorTag, BuildingPropertyTag,
|
||
ConstructionEra, EraCause, FacadeRhythm, FloorExtent, FloorHeightProfile, HsvColor,
|
||
QuarterSkeleton, RoofForm, SetbackTier, StreetSurface, TileRect, WallMaterial,
|
||
ZoneTypeId,
|
||
};
|
||
use std::collections::BTreeMap;
|
||
|
||
let mut block_tags: BTreeMap<(u8, u8), Vec<BuildingPropertyTag>> = BTreeMap::new();
|
||
block_tags.insert(
|
||
(0, 0),
|
||
vec![BuildingPropertyTag {
|
||
zone_type_id: ZoneTypeId::new("residential_low"),
|
||
footprint: TileRect::new(4, 4, 6, 6),
|
||
extent: FloorExtent {
|
||
base_floor: 0,
|
||
floor_count: 2,
|
||
heights: FloorHeightProfile::Uniform(3),
|
||
},
|
||
entry_class: BuildingEntryClass::Public,
|
||
flavor_ref: ArchitectureFlavorRef::InVocabulary(0),
|
||
era: ConstructionEra::Founding,
|
||
era_cause: EraCause::Original,
|
||
initial_condition: TileCondition::Intact,
|
||
exterior: BuildingExteriorTag {
|
||
wall_material: WallMaterial::Generic,
|
||
roof_form: RoofForm::Generic,
|
||
facade_rhythm: FacadeRhythm::Generic,
|
||
setback_tier: SetbackTier::Standard,
|
||
color: HsvColor {
|
||
hue: 0,
|
||
sat: 0,
|
||
val: 5_000,
|
||
},
|
||
street_surface: StreetSurface::Generic,
|
||
},
|
||
doors: Vec::new(),
|
||
}],
|
||
);
|
||
QuarterWorldState {
|
||
skeleton: QuarterSkeleton {
|
||
quarter_id: 4242,
|
||
..Default::default()
|
||
},
|
||
block_tags,
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn build_fill_chunk_item_pulls_block_tags() {
|
||
let quarter = quarter_with_one_building();
|
||
let item = build_fill_chunk_item(&quarter, (0, 0), (0, 0));
|
||
let GenWorkItem::FillChunk {
|
||
quarter_id,
|
||
block_pos,
|
||
sub_chunk,
|
||
block_tags,
|
||
} = item
|
||
else {
|
||
panic!("expected FillChunk");
|
||
};
|
||
assert_eq!(quarter_id, 4242);
|
||
assert_eq!(block_pos, (0, 0));
|
||
assert_eq!(sub_chunk, (0, 0));
|
||
assert_eq!(
|
||
block_tags.len(),
|
||
1,
|
||
"covering block's tags must be resolved"
|
||
);
|
||
|
||
// A block with no buildings is a valid empty fill, not a missing-skeleton error.
|
||
let empty = build_fill_chunk_item(&quarter, (3, 3), (0, 0));
|
||
let GenWorkItem::FillChunk { block_tags, .. } = empty else {
|
||
panic!("expected FillChunk");
|
||
};
|
||
assert!(block_tags.is_empty(), "empty block → empty tags");
|
||
}
|
||
|
||
#[test]
|
||
fn fill_chunk_round_trip_produces_populated_shell() {
|
||
let q = make_queue();
|
||
let quarter = quarter_with_one_building();
|
||
q.submit(
|
||
build_fill_chunk_item(&quarter, (0, 0), (0, 0)),
|
||
GenPriority::High,
|
||
);
|
||
std::thread::sleep(Duration::from_millis(50));
|
||
let completions = q.drain_completions();
|
||
assert_eq!(completions.len(), 1);
|
||
let GenCompletion::ChunkFilled { filled } = &completions[0] else {
|
||
panic!("expected ChunkFilled, got {:?}", completions[0]);
|
||
};
|
||
assert_eq!(filled.quarter_id, 4242);
|
||
assert_eq!(filled.chunk_in_quarter(), (0, 0));
|
||
assert!(
|
||
filled.voxel_count() > 0,
|
||
"a chunk containing a building must derive shell voxels"
|
||
);
|
||
}
|
||
|
||
// -------------------------------------------------------------------
|
||
// DeriveWindow / submit_window coalescing (D-226 T-1124 amendment, T-1137)
|
||
// -------------------------------------------------------------------
|
||
|
||
/// Build a `DeriveWindow` work item pointing at a tiny test heightmap,
|
||
/// mirroring `analyze()`'s fixture shape.
|
||
fn derive_window(body_id: &str, conn_id: ConnectionId, center: DistrictPos) -> GenWorkItem {
|
||
GenWorkItem::DeriveWindow {
|
||
body_id: body_id.to_string(),
|
||
conn_id,
|
||
heightmap_path: test_heightmap_path(),
|
||
sea_level: 0.3,
|
||
body_seed: SeedChain::for_body(42, body_id),
|
||
body_params: Box::new(BodyParams {
|
||
hydrosphere: Some("ocean".into()),
|
||
atmosphere: Some("breathable".into()),
|
||
planet_class: Some("temperate".into()),
|
||
body_radius_km: Some(6371.0),
|
||
..Default::default()
|
||
}),
|
||
center,
|
||
n: 4,
|
||
granularity: crate::atlas::layer_proxy::WINDOW_GRANULARITY_DISTRICT,
|
||
min_wl_m: 0,
|
||
}
|
||
}
|
||
|
||
/// The full `DeriveWindow` → Rayon → `WindowDerived` round trip: the
|
||
/// real re-derive-via-`run_layer1` path (T-1137 binding decision) runs
|
||
/// end to end and produces a populated `DistrictWindowLayer`.
|
||
#[test]
|
||
fn derive_window_round_trip_produces_populated_layer() {
|
||
let q = make_queue();
|
||
q.submit(
|
||
derive_window("TestBody", ConnectionId(0), (2, -1)),
|
||
GenPriority::Immediate,
|
||
);
|
||
std::thread::sleep(Duration::from_millis(150));
|
||
let completions = q.drain_completions();
|
||
assert_eq!(completions.len(), 1);
|
||
let GenCompletion::WindowDerived { body_id, layer } = &completions[0] else {
|
||
panic!("expected WindowDerived, got {:?}", completions[0]);
|
||
};
|
||
assert_eq!(body_id, "TestBody");
|
||
assert_eq!(layer.center, (2, -1));
|
||
assert_eq!(layer.n, 4);
|
||
assert_eq!(layer.morphology.len(), 16);
|
||
assert_eq!(layer.elev_q.len(), 16);
|
||
assert_eq!(layer.temp_dc.len(), 16);
|
||
assert_eq!(layer.moisture_q.len(), 16);
|
||
assert_eq!(layer.vegetation.len(), 16);
|
||
assert_eq!(layer.glaciation.len(), 16);
|
||
}
|
||
|
||
/// `submit_window` coalescing (D-226 T-1124 amendment §1 "recommended"):
|
||
/// two `DeriveWindow` items for the SAME `(connection, body)` queued
|
||
/// while the pool is saturated collapse to ONE pending entry — the
|
||
/// second submission replaces the first rather than queuing alongside it.
|
||
#[test]
|
||
fn submit_window_coalesces_same_connection_and_body() {
|
||
// Single-thread pool: the first item occupies the only worker, so
|
||
// subsequent DeriveWindow submissions stay in `pending` long enough
|
||
// to inspect (mirrors `priority_ordering_respected_under_saturation`'s
|
||
// saturation trick). Uses `analyze()` (real cascade work — measurable
|
||
// latency), NOT `FillChunk` (documented "trivially fast" — it can
|
||
// complete before the next `submit_window` call even runs, which
|
||
// would make `pending_count()` observe 0 instead of 1, a real race
|
||
// this test hit before switching occupiers).
|
||
let q = GenerationQueue::with_threads(1);
|
||
q.submit(analyze("Occupier"), GenPriority::Low);
|
||
|
||
let conn = ConnectionId(7);
|
||
q.submit_window(derive_window("Coal", conn, (0, 0)), GenPriority::Immediate);
|
||
assert_eq!(
|
||
q.pending_count(),
|
||
1,
|
||
"one DeriveWindow queued behind the saturating item"
|
||
);
|
||
|
||
// A second DeriveWindow for the SAME (connection, body) supersedes
|
||
// the first — pending count stays at 1, not 2.
|
||
q.submit_window(derive_window("Coal", conn, (5, 5)), GenPriority::Immediate);
|
||
assert_eq!(
|
||
q.pending_count(),
|
||
1,
|
||
"same (connection, body) DeriveWindow must supersede, not queue alongside"
|
||
);
|
||
|
||
// Drain everything and confirm exactly one WindowDerived for "Coal",
|
||
// carrying the SECOND (superseding) center — not the first.
|
||
std::thread::sleep(Duration::from_millis(150));
|
||
let mut completions = q.drain_completions();
|
||
std::thread::sleep(Duration::from_millis(150));
|
||
completions.extend(q.drain_completions());
|
||
|
||
let window_completions: Vec<_> = completions
|
||
.iter()
|
||
.filter_map(|c| {
|
||
if let GenCompletion::WindowDerived { body_id, layer } = c {
|
||
if body_id == "Coal" {
|
||
return Some(layer);
|
||
}
|
||
}
|
||
None
|
||
})
|
||
.collect();
|
||
assert_eq!(
|
||
window_completions.len(),
|
||
1,
|
||
"exactly one WindowDerived for the coalesced body, not two"
|
||
);
|
||
assert_eq!(
|
||
window_completions[0].center,
|
||
(5, 5),
|
||
"the surviving item must be the SECOND (superseding) submission"
|
||
);
|
||
}
|
||
|
||
/// `submit_window` does NOT coalesce across different connections or
|
||
/// different bodies — only an exact `(connection, body)` match supersedes.
|
||
#[test]
|
||
fn submit_window_does_not_coalesce_different_keys() {
|
||
let q = GenerationQueue::with_threads(1);
|
||
// See `submit_window_coalesces_same_connection_and_body`'s comment on
|
||
// why the occupier must be `analyze()`, not `FillChunk`.
|
||
q.submit(analyze("Occupier2"), GenPriority::Low);
|
||
|
||
// Different connections, same body — must NOT coalesce.
|
||
q.submit_window(
|
||
derive_window("Shared", ConnectionId(1), (0, 0)),
|
||
GenPriority::Immediate,
|
||
);
|
||
q.submit_window(
|
||
derive_window("Shared", ConnectionId(2), (1, 1)),
|
||
GenPriority::Immediate,
|
||
);
|
||
assert_eq!(
|
||
q.pending_count(),
|
||
2,
|
||
"different connections requesting the same body must NOT coalesce"
|
||
);
|
||
}
|
||
|
||
// -------------------------------------------------------------------
|
||
// TerrainAnalysisCache (T-1137, PR #187 review — Tyre C1)
|
||
// -------------------------------------------------------------------
|
||
|
||
fn window_test_hm() -> crate::atlas::heightmap::BodyHeightmap {
|
||
use crate::atlas::heightmap::BodyHeightmap;
|
||
let (w, h) = (32u32, 16u32);
|
||
let n = (w * h) as usize;
|
||
let data = (0..n)
|
||
.map(|i| {
|
||
let r = (i / w as usize) as f32 / h as f32;
|
||
let c = (i % w as usize) as f32 / w as f32;
|
||
(r * 0.6 + c * 0.4).min(1.0)
|
||
})
|
||
.collect();
|
||
BodyHeightmap {
|
||
body_id: "test".into(),
|
||
width: w,
|
||
height: h,
|
||
data,
|
||
sea_level: 0.3,
|
||
}
|
||
}
|
||
|
||
/// A miss re-derives and populates the entry; a subsequent hit for the
|
||
/// SAME body returns an equal `TerrainAnalysis` (D-227: the same
|
||
/// heightmap always derives to the same analysis) WITHOUT growing the
|
||
/// cache — `len()` stays at 1, proving the second call short-circuited
|
||
/// past `run_layer1` rather than deriving-then-overwriting.
|
||
#[test]
|
||
fn terrain_analysis_cache_hit_reuses_entry() {
|
||
let mut cache = TerrainAnalysisCache::new(8);
|
||
let hm = window_test_hm();
|
||
|
||
assert!(!cache.contains("BodyA"));
|
||
let first = cache.get_or_derive("BodyA", &hm);
|
||
assert_eq!(cache.len(), 1);
|
||
assert!(cache.contains("BodyA"));
|
||
|
||
let second = cache.get_or_derive("BodyA", &hm);
|
||
assert_eq!(
|
||
cache.len(),
|
||
1,
|
||
"a hit must not insert a second entry for the same body"
|
||
);
|
||
assert_eq!(
|
||
first.ocean_mask, second.ocean_mask,
|
||
"same heightmap → identical re-derived analysis (D-227)"
|
||
);
|
||
assert_eq!(first.slope_deg, second.slope_deg);
|
||
assert_eq!(first.elev_pct, second.elev_pct);
|
||
}
|
||
|
||
/// Different bodies get independent entries, and a capacity-2 cache
|
||
/// evicts the LEAST-RECENTLY-USED entry — not insertion order — when a
|
||
/// third body is derived. Mirrors `BodyWorldStateCache`'s own
|
||
/// `update_last_accessed_on_get` precedent: touching "BodyA" again before
|
||
/// the third insert must save it from eviction.
|
||
#[test]
|
||
fn terrain_analysis_cache_evicts_lru_not_fifo() {
|
||
let mut cache = TerrainAnalysisCache::new(2);
|
||
let hm = window_test_hm();
|
||
|
||
cache.get_or_derive("BodyA", &hm);
|
||
cache.get_or_derive("BodyB", &hm);
|
||
assert_eq!(cache.len(), 2);
|
||
|
||
// Touch BodyA again — it is now the MOST recently used, so BodyB
|
||
// (untouched since its own insert) is the true LRU victim.
|
||
cache.get_or_derive("BodyA", &hm);
|
||
|
||
// Insert a third body — capacity 2 forces an eviction.
|
||
cache.get_or_derive("BodyC", &hm);
|
||
assert_eq!(cache.len(), 2);
|
||
assert!(
|
||
cache.contains("BodyA"),
|
||
"recently re-touched BodyA must survive eviction"
|
||
);
|
||
assert!(
|
||
!cache.contains("BodyB"),
|
||
"BodyB (true LRU — untouched since its own insert) must be evicted"
|
||
);
|
||
assert!(cache.contains("BodyC"));
|
||
}
|
||
|
||
/// End-to-end: two `DeriveWindow` work items for the SAME body, submitted
|
||
/// through the real `GenerationQueue` (not the bare `TerrainAnalysisCache`
|
||
/// unit above), share ONE `TerrainAnalysisCache` entry — the fix for the
|
||
/// PR #187 review C1 finding (the original landing called `run_layer1`
|
||
/// unconditionally on every `DeriveWindow`, so panning around one body
|
||
/// paid the ~45 ms re-derive on every window instead of just the first).
|
||
/// Both windows must still complete correctly (content assertions, not
|
||
/// timing — a wall-clock assertion would be flaky); the cache-population
|
||
/// count is the load-bearing proof of reuse.
|
||
#[test]
|
||
fn two_windows_on_same_body_share_one_terrain_analysis_entry() {
|
||
let q = GenerationQueue::with_threads(2);
|
||
let conn_a = ConnectionId(1);
|
||
let conn_b = ConnectionId(2);
|
||
// Two DIFFERENT connections so submit_window's coalescing (which
|
||
// supersedes same-connection/same-body pending items) doesn't collapse
|
||
// these into one work item — the point here is two DISTINCT completed
|
||
// derives sharing the cache, not coalescing (already covered above).
|
||
q.submit_window(
|
||
derive_window("SharedBody", conn_a, (0, 0)),
|
||
GenPriority::Immediate,
|
||
);
|
||
q.submit_window(
|
||
derive_window("SharedBody", conn_b, (10, 10)),
|
||
GenPriority::Immediate,
|
||
);
|
||
|
||
std::thread::sleep(Duration::from_millis(200));
|
||
let completions = q.drain_completions();
|
||
let windows: Vec<_> = completions
|
||
.iter()
|
||
.filter_map(|c| {
|
||
if let GenCompletion::WindowDerived { body_id, layer } = c {
|
||
if body_id == "SharedBody" {
|
||
return Some(layer);
|
||
}
|
||
}
|
||
None
|
||
})
|
||
.collect();
|
||
assert_eq!(windows.len(), 2, "both windows must complete");
|
||
let centers: std::collections::BTreeSet<_> = windows.iter().map(|l| l.center).collect();
|
||
assert_eq!(
|
||
centers,
|
||
std::collections::BTreeSet::from([(0, 0), (10, 10)]),
|
||
"both distinct windows survived, not coalesced"
|
||
);
|
||
|
||
// The queue's own TerrainAnalysisCache must hold exactly ONE entry
|
||
// for "SharedBody" — both derives shared it rather than each
|
||
// re-deriving independently.
|
||
let cache = q.terrain_cache.lock().unwrap();
|
||
assert_eq!(
|
||
cache.len(),
|
||
1,
|
||
"two DeriveWindow items for the same body must share one TerrainAnalysis entry"
|
||
);
|
||
assert!(cache.contains("SharedBody"));
|
||
}
|
||
}
|