diff --git a/server/src/lib.rs b/server/src/lib.rs index 1daca7b8d..454a502d4 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -11,6 +11,7 @@ pub mod simulation; pub mod storyteller; pub mod tick_phases; pub mod voice; +pub mod workers; // test_world::reset is always compiled (used by simulation::input). // Room definitions, constants, and setup_gauntlet are gated behind // the "gauntlet" feature (default-on) to allow stripping from release builds. diff --git a/server/src/simulation/mod.rs b/server/src/simulation/mod.rs index 6dc975f13..0956ccd74 100644 --- a/server/src/simulation/mod.rs +++ b/server/src/simulation/mod.rs @@ -70,6 +70,12 @@ impl Plugin for SimulationPlugin { .init_resource::() .init_resource::(); + // Background worker pools (#843 Part C) — off-tick computation on separate cores. + // Stub handlers for now; real implementations plug in at Phase 5+. + app.insert_resource(crate::workers::stubs::ChunkGenWorker::new(2)); + app.insert_resource(crate::workers::stubs::NpcPrepWorker::new(1)); + app.insert_resource(crate::workers::stubs::OffscreenTickWorker::new(1)); + // Phase sub-plugins (#843) — each owns its domain app.add_plugins(time_plugin::TimePlugin); app.add_plugins(input_plugin::InputPlugin); @@ -77,6 +83,18 @@ impl Plugin for SimulationPlugin { app.add_plugins(social_plugin::SocialPlugin); app.add_plugins(economy_plugin::EconomyPlugin); + // Background worker tick integration (#843 Part C) + app.add_systems( + Update, + crate::workers::systems::poll_worker_results + .in_set(crate::tick_phases::TickPhase::PreInput), + ); + app.add_systems( + Update, + crate::workers::systems::push_worker_requests + .in_set(crate::tick_phases::TickPhase::PostSnapshot), + ); + // save_load is an exclusive system (takes &mut World). // Exclusive systems in Bevy 0.18 cannot use .in_set() — use .before()/.after() // to position it within the Snapshot phase window. diff --git a/server/src/simulation/monologue.rs b/server/src/simulation/monologue.rs index 9c423fca2..b2ef8dc52 100644 --- a/server/src/simulation/monologue.rs +++ b/server/src/simulation/monologue.rs @@ -696,10 +696,15 @@ pub fn process_contradiction_monologue( #[cfg(test)] mod tests { use super::*; - use crate::simulation::rng::SimRng; + use crate::simulation::rng::{EntityRng, SimRng}; use crate::simulation::time::SimulationTime; use bevy_ecs::world::World; + /// Test helper: create an EntityRng for the player entity in tests. + fn test_entity_rng() -> EntityRng { + EntityRng::from_seed_and_id(42, 0) + } + // ----------------------------------------------------------------------- // SprintAnomalyQueue unit tests (#428, D-055) // ----------------------------------------------------------------------- @@ -795,6 +800,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), MonologueBuffer::default(), @@ -823,6 +829,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), MonologueBuffer::default(), @@ -857,6 +864,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), buffer, @@ -887,6 +895,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), MonologueBuffer::default(), @@ -912,6 +921,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), MonologueBuffer::default(), @@ -935,6 +945,7 @@ mod tests { let mut world = setup_anomaly_world(); world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1000,6 +1011,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1042,6 +1054,7 @@ mod tests { let player = world .spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1106,6 +1119,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), buffer, @@ -1139,6 +1153,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1186,6 +1201,7 @@ mod tests { let player = world .spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1228,6 +1244,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1268,6 +1285,7 @@ mod tests { world.spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(5, 5, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1344,6 +1362,7 @@ mod tests { world .spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(10, 10, 0), MonologueState::default(), MonologueBuffer::default(), @@ -1912,6 +1931,7 @@ mod tests { world .spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(0, 0, 0), MonologueState::default(), MonologueBuffer::default(), @@ -2120,6 +2140,7 @@ mod tests { let player = world .spawn(( PlayerCharacter, + test_entity_rng(), TilePosition::new(0, 0, 0), MonologueState::default(), MonologueBuffer::default(), diff --git a/server/src/workers/mod.rs b/server/src/workers/mod.rs new file mode 100644 index 000000000..a6ddc53cf --- /dev/null +++ b/server/src/workers/mod.rs @@ -0,0 +1,28 @@ +//! Background worker infrastructure (#843 Part C). +//! +//! Generic thread pool for off-tick computation. Workers run on dedicated +//! threads, separate from the Bevy tick loop. The tick loop pushes requests +//! (non-blocking) and polls for results (non-blocking). +//! +//! Each worker type is a Bevy `Resource` wrapping a typed `BackgroundWorkerPool`. +//! Systems in `TickPhase::PreInput` poll results; systems in `TickPhase::PostSnapshot` +//! push new requests. +//! +//! ## Determinism (D-010) +//! +//! Workers preserve determinism through request-at-tick, deliver-complete: +//! - Inputs come from the deterministic tick loop (same seed → same requests) +//! - Outputs are deterministic per input (seeded RNG, no external state) +//! - The tick loop never incorporates partial results +//! - The only non-deterministic part is latency (when results arrive) +//! +//! ## Delivery strategies +//! +//! Each request declares how the tick loop handles a not-yet-ready result: +//! - `Fallback`: use default/cached value, player never waits +//! - `GracefulDegrade`: area loads partially, elements pop in when ready +//! - `ModalLock`: tick loop pauses until result arrives (rare, diegetic UI) + +pub mod pool; +pub mod stubs; +pub mod systems; diff --git a/server/src/workers/pool.rs b/server/src/workers/pool.rs new file mode 100644 index 000000000..daf9f62ef --- /dev/null +++ b/server/src/workers/pool.rs @@ -0,0 +1,158 @@ +//! Generic background worker pool. +//! +//! Wraps crossbeam channels + spawned threads. The handler closure captures +//! any context it needs (world seed, config, etc.) at construction time. +//! +//! ## Worker crash behavior +//! +//! TODO (#843): If a worker thread panics, in-flight requests on that thread +//! are lost. The pool does NOT currently re-enqueue them. For Fallback-strategy +//! workers (voice, LLM) this is acceptable — the tick loop uses the default. +//! For ModalLock-strategy workers (chunk gen on gate transit), a lost request +//! means the player waits forever. Future: add heartbeat monitoring and +//! automatic re-enqueue on worker death. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::thread::{self, JoinHandle}; + +use crossbeam_channel::{Receiver, Sender, TryRecvError}; + +/// How the tick loop handles a not-yet-ready result. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DeliveryStrategy { + /// Use default/cached value. Player never waits. Seamless. + Fallback, + /// Area loads partially, elements pop in when ready. Visible but non-blocking. + GracefulDegrade, + /// Tick loop pauses until result arrives. Diegetic buffer UI shown to player. + /// Use sparingly — only for gate transit, initial world load, etc. + ModalLock, +} + +/// A request with its delivery strategy attached. +#[derive(Debug)] +pub struct WorkRequest { + pub payload: Req, + pub strategy: DeliveryStrategy, +} + +/// A generic background worker pool. +/// +/// `Req` and `Resp` are the request/response types. The handler closure +/// runs on worker threads and transforms requests into responses. +/// +/// ```ignore +/// let pool = BackgroundWorkerPool::spawn(2, |req: ChunkGenRequest| { +/// // runs on worker thread — has access to captured world_seed +/// generate_chunk(req, world_seed) +/// }); +/// pool.submit(WorkRequest { payload: req, strategy: DeliveryStrategy::GracefulDegrade }); +/// for result in pool.drain() { /* incorporate */ } +/// ``` +pub struct BackgroundWorkerPool { + tx: Sender>, + rx: Receiver<(Resp, DeliveryStrategy)>, + shutdown: Arc, + _workers: Vec>, +} + +impl BackgroundWorkerPool +where + Req: Send + 'static, + Resp: Send + 'static, +{ + /// Spawn `count` worker threads with the given handler closure. + /// + /// The handler is cloned to each thread. It should capture any context + /// it needs (world seed, config registries, etc.) via `Arc` or `Clone`. + pub fn spawn(count: usize, handler: H) -> Self + where + H: Fn(Req) -> Resp + Send + Sync + Clone + 'static, + { + let (req_tx, req_rx) = crossbeam_channel::unbounded::>(); + let (resp_tx, resp_rx) = crossbeam_channel::unbounded::<(Resp, DeliveryStrategy)>(); + let shutdown = Arc::new(AtomicBool::new(false)); + + let mut workers = Vec::with_capacity(count); + for id in 0..count { + let rx = req_rx.clone(); + let tx = resp_tx.clone(); + let shutdown = shutdown.clone(); + let handler = handler.clone(); + + let handle = thread::Builder::new() + .name(format!("worker-{id}")) + .spawn(move || { + while !shutdown.load(Ordering::Relaxed) { + match rx.recv_timeout(std::time::Duration::from_millis(100)) { + Ok(work) => { + let strategy = work.strategy; + let result = handler(work.payload); + if tx.send((result, strategy)).is_err() { + break; // response channel closed + } + } + Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break, + } + } + }) + .expect("failed to spawn worker thread"); + + workers.push(handle); + } + + Self { + tx: req_tx, + rx: resp_rx, + shutdown, + _workers: workers, + } + } + + /// Submit a request. Non-blocking — returns immediately. + pub fn submit(&self, request: WorkRequest) { + let _ = self.tx.send(request); + } + + /// Poll for a single completed result. Non-blocking. + pub fn poll(&self) -> Option<(Resp, DeliveryStrategy)> { + match self.rx.try_recv() { + Ok(result) => Some(result), + Err(TryRecvError::Empty) => None, + Err(TryRecvError::Disconnected) => None, + } + } + + /// Drain all completed results. Non-blocking. + pub fn drain(&self) -> Vec<(Resp, DeliveryStrategy)> { + let mut results = Vec::new(); + while let Ok(result) = self.rx.try_recv() { + results.push(result); + } + results + } + + /// Check if any results are ready without consuming them. + pub fn has_results(&self) -> bool { + !self.rx.is_empty() + } + + /// Number of pending requests in the work queue. + pub fn pending_count(&self) -> usize { + self.tx.len() + } + + /// Signal all workers to shut down gracefully. + pub fn shutdown(&self) { + self.shutdown.store(true, Ordering::Relaxed); + } +} + +impl Drop for BackgroundWorkerPool { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::Relaxed); + // Workers will exit on next recv_timeout cycle (100ms max) + } +} diff --git a/server/src/workers/stubs.rs b/server/src/workers/stubs.rs new file mode 100644 index 000000000..9acd006b8 --- /dev/null +++ b/server/src/workers/stubs.rs @@ -0,0 +1,143 @@ +//! Stub worker resources (#843 Part C). +//! +//! Each worker type is a Bevy Resource wrapping a typed `BackgroundWorkerPool`. +//! The actual computation logic is a no-op placeholder — implementations +//! plug in when the phases that need them arrive (Phase 5+). +//! +//! The infrastructure (channels, threads, push/poll) is real and tested. + +use bevy_ecs::prelude::*; + +use super::pool::{BackgroundWorkerPool, DeliveryStrategy, WorkRequest}; + +// --------------------------------------------------------------------------- +// ChunkGen — terrain, buildings, props, navmesh for a world chunk +// --------------------------------------------------------------------------- + +/// Request to generate a chunk at the given coordinates. +#[derive(Debug, Clone)] +pub struct ChunkGenRequest { + pub chunk_x: i32, + pub chunk_y: i32, + pub chunk_z: i32, + pub world_seed: u64, +} + +/// Generated chunk data (placeholder — real struct will hold terrain, props, navmesh). +#[derive(Debug)] +pub struct ChunkGenResult { + pub chunk_x: i32, + pub chunk_y: i32, + pub chunk_z: i32, + /// Placeholder: real result holds terrain heightfield, building list, prop list, navmesh. + pub generated: bool, +} + +/// Bevy resource for the chunk generation worker pool. +#[derive(Resource)] +pub struct ChunkGenWorker { + pub pool: BackgroundWorkerPool, +} + +impl ChunkGenWorker { + pub fn new(thread_count: usize) -> Self { + Self { + pool: BackgroundWorkerPool::spawn(thread_count, |req: ChunkGenRequest| { + // Stub: real implementation generates terrain from seed + coordinates + ChunkGenResult { + chunk_x: req.chunk_x, + chunk_y: req.chunk_y, + chunk_z: req.chunk_z, + generated: true, + } + }), + } + } + + 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() + } +} + +// --------------------------------------------------------------------------- +// NpcPrep — pre-compute NPC state for areas about to become visible +// --------------------------------------------------------------------------- + +/// Request to prepare NPCs for an area. +#[derive(Debug, Clone)] +pub struct NpcPrepRequest { + pub area_id: String, + pub world_seed: u64, +} + +/// Prepared NPC state (placeholder). +#[derive(Debug)] +pub struct NpcPrepResult { + pub area_id: String, + pub npc_count: usize, +} + +/// Bevy resource for the NPC preparation worker pool. +#[derive(Resource)] +pub struct NpcPrepWorker { + pub pool: BackgroundWorkerPool, +} + +impl NpcPrepWorker { + pub fn new(thread_count: usize) -> Self { + Self { + pool: BackgroundWorkerPool::spawn(thread_count, |req: NpcPrepRequest| { + // Stub: real implementation pre-generates NPC appearance, inventory, mood + NpcPrepResult { + area_id: req.area_id, + npc_count: 0, + } + }), + } + } +} + +// --------------------------------------------------------------------------- +// OffscreenTick — advance NPC state for entities outside the active tier +// --------------------------------------------------------------------------- + +/// Request to advance off-screen NPCs by a specific number of ticks. +#[derive(Debug, Clone)] +pub struct OffscreenTickRequest { + pub area_id: String, + /// Exact number of simulation ticks to advance. Deterministic. + pub ticks_to_advance: u64, + pub world_seed: u64, +} + +/// Updated NPC state after off-screen ticking (placeholder). +#[derive(Debug)] +pub struct OffscreenTickResult { + pub area_id: String, + pub ticks_advanced: u64, +} + +/// Bevy resource for the off-screen tick worker pool. +#[derive(Resource)] +pub struct OffscreenTickWorker { + pub pool: BackgroundWorkerPool, +} + +impl OffscreenTickWorker { + pub fn new(thread_count: usize) -> Self { + Self { + pool: BackgroundWorkerPool::spawn(thread_count, |req: OffscreenTickRequest| { + // Stub: real implementation snapshots NPC state, advances N ticks + OffscreenTickResult { + area_id: req.area_id, + ticks_advanced: req.ticks_to_advance, + } + }), + } + } +} diff --git a/server/src/workers/systems.rs b/server/src/workers/systems.rs new file mode 100644 index 000000000..e877f00c7 --- /dev/null +++ b/server/src/workers/systems.rs @@ -0,0 +1,59 @@ +//! Tick-loop integration systems for background workers (#843 Part C). +//! +//! - `poll_worker_results`: runs in PreInput, drains completed results +//! - `push_worker_requests`: runs in PostSnapshot, queues new work +//! +//! Both are no-ops when no results/requests are pending. The worker pools +//! exist as Bevy Resources — systems access them via `Res` etc. + +use bevy_ecs::prelude::*; + +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+. +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() { + tracing::debug!( + chunk_x = result.chunk_x, + chunk_y = result.chunk_y, + "ChunkGen result ready" + ); + } + } + + if let Some(worker) = npc_worker { + for (result, _strategy) in worker.pool.drain() { + tracing::debug!(area = %result.area_id, "NpcPrep result ready"); + } + } + + if let Some(worker) = offscreen_worker { + for (result, _strategy) in worker.pool.drain() { + tracing::debug!( + area = %result.area_id, + ticks = result.ticks_advanced, + "OffscreenTick result ready" + ); + } + } +} + +/// PostSnapshot: push new work requests based on player state. +/// +/// Placeholder — real logic (predict player movement direction, queue +/// chunks ahead, snapshot departing NPCs) plugs in at Phase 5+. +pub fn push_worker_requests() { + // No-op until chunk streaming and NPC tier transitions are implemented. + // The worker pools are ready to receive requests via: + // chunk_worker.submit(req, DeliveryStrategy::GracefulDegrade); + // npc_worker.pool.submit(WorkRequest { ... }); +}