diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index e87a33a..626b242 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -73,8 +73,11 @@ fn main() -> anyhow::Result<()> { let generator = VoxelGenerator::new(worldgen_config, seed); + // The region directory holds the `.region` save files for this world; a later slice will resolve it per named world under the platform user-data directory. + let region_dir = std::path::PathBuf::from("saves/default/region"); + let mut world = World::new(); - world.insert_resource(ServerWorld::new(generator)); + world.insert_resource(ServerWorld::new(generator, region_dir)); // Spawn a single dummy player anchor at the world origin. world.spawn(( diff --git a/crates/server/src/save.rs b/crates/server/src/save.rs index 92ce02d..a217fa4 100644 --- a/crates/server/src/save.rs +++ b/crates/server/src/save.rs @@ -8,6 +8,8 @@ //! (`.tmp` + fsync + rename); the on-disk format is unchanged, so a later slice can switch to an //! append-in-place strategy without a format change. +mod region_actor; mod region_file; +pub use region_actor::{SaveActor, SaveRequest}; pub use region_file::{REGION_SIZE, RegionFile, region_coords, region_path}; diff --git a/crates/server/src/save/region_actor.rs b/crates/server/src/save/region_actor.rs new file mode 100644 index 0000000..c17a287 --- /dev/null +++ b/crates/server/src/save/region_actor.rs @@ -0,0 +1,83 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! A dedicated thread that owns every open region file and answers load requests over a channel. + +use std::collections::HashMap; +use std::collections::hash_map::Entry; +use std::path::{Path, PathBuf}; +use std::thread::{self, JoinHandle}; + +use crossbeam_channel::{Receiver, Sender}; +use shared::save::SaveError; +use shared::world::{ChunkData, ChunkPos}; + +use super::region_file::{RegionFile, region_coords, region_path}; + +/// A request sent to the save actor. Each variant carries its own one-shot reply channel. +pub enum SaveRequest { + /// Reads the stored chunk at a position, replying with the saved modification if one exists. + Read { + /// The chunk position to look up. + pos: ChunkPos, + /// The one-shot channel the actor replies on: `Ok(Some(data))` for a saved modification, `Ok(None)` when the chunk was never modified, or `Err` on a save-layer failure. + reply: Sender, SaveError>>, + }, +} + +/// A handle to the running save actor: the request sender plus the owning thread's join handle. +pub struct SaveActor { + /// The sending end of the request channel; cloned into every worker so it can issue reads. + request_tx: Sender, + /// The actor thread handle, retained so it can be joined on shutdown. + // * Retained ahead of a dedicated shutdown path; not yet read because the server has no graceful-stop sequence. + #[expect(dead_code)] + handle: JoinHandle<()>, +} + +impl SaveActor { + /// Spawns the actor thread, which owns the region files beneath `region_dir` for its lifetime. + #[must_use] + pub fn spawn(region_dir: PathBuf) -> Self { + let (request_tx, request_rx) = crossbeam_channel::unbounded::(); + let handle = thread::spawn(move || actor_loop(®ion_dir, &request_rx)); + Self { request_tx, handle } + } + + /// Returns a fresh sender for a worker to issue requests through. + #[must_use] + pub fn sender(&self) -> Sender { + self.request_tx.clone() + } +} + +/// The actor's run loop: it owns the region-file map and answers requests until every sender is dropped. +fn actor_loop(region_dir: &Path, request_rx: &Receiver) { + // The actor is the sole owner of this map, so region files need no lock of their own. + let mut regions: HashMap<(i32, i32), RegionFile> = HashMap::new(); + while let Ok(request) = request_rx.recv() { + match request { + SaveRequest::Read { pos, reply } => { + let result = read_chunk(&mut regions, region_dir, pos); + // A send error means the requesting worker has gone away; the reply is simply dropped. + let _ = reply.send(result); + } + } + } +} + +/// Reads the stored chunk at `pos`, opening and caching its region file on first access. +fn read_chunk( + regions: &mut HashMap<(i32, i32), RegionFile>, + region_dir: &Path, + pos: ChunkPos, +) -> Result, SaveError> { + let key = region_coords(pos.x, pos.z); + // The region file is opened once on first touch; every later read of a chunk in it hits the in-memory copy. + let region = match regions.entry(key) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => { + entry.insert(RegionFile::open(region_path(region_dir, pos.x, pos.z))?) + } + }; + region.read_chunk(pos) +} diff --git a/crates/server/src/world_server.rs b/crates/server/src/world_server.rs index 57946bf..deb2b6b 100644 --- a/crates/server/src/world_server.rs +++ b/crates/server/src/world_server.rs @@ -6,15 +6,20 @@ use bevy_ecs::prelude::Resource; use crossbeam_channel::{Receiver, Sender}; use shared::{ generator::VoxelGenerator, - world::{Chunk, ChunkPos}, + save::SaveError, + world::{Chunk, ChunkData, ChunkPos}, }; use std::{ collections::{HashMap, HashSet}, + path::PathBuf, sync::Arc, thread::JoinHandle, }; +use tracing::warn; -/// Outcome of a single streaming reconcile pass, for logging and tests. +use crate::save::{SaveActor, SaveRequest}; + +/// Outcome of a single streaming reconcile pass, surfaced for logging and tests. #[derive(Copy, Clone, Debug, PartialEq, Eq)] pub struct StreamStats { /// Number of chunks generated and inserted this pass. @@ -32,12 +37,16 @@ pub struct StreamStats { pub struct ServerWorld { /// Currently resident chunks, keyed by chunk-space position. chunks: HashMap, - /// Sending end of the job channel; the main thread pushes positions that require generation. + /// Sending end of the job channel; the main thread pushes positions to load. job_tx: Sender, /// Receiving end of the result channel; the main thread drains finished chunks returned by workers. result_rx: Receiver<(ChunkPos, Chunk)>, - /// Positions dispatched to a worker but not yet returned, preventing the same chunk from being re-dispatched on subsequent passes. + /// Positions dispatched to a worker but not yet returned, preventing the same chunk being re-dispatched on subsequent passes. in_flight: HashSet, + /// The dedicated thread owning all region files, kept alive for the world's lifetime. + // Retained so its request channel stays open for the workers; not read again after construction. + #[expect(dead_code)] + save_actor: SaveActor, /// Handles to the generation worker threads, retained so they can be joined on shutdown. // Retained ahead of a dedicated shutdown path; not yet read because the server has no graceful-stop sequence. #[expect(dead_code)] @@ -45,28 +54,31 @@ pub struct ServerWorld { } impl ServerWorld { - /// Initializes a new authoritative server world with the provided generator. + /// Initializes a new authoritative server world with the provided generator, resolving chunk loads against region files under `region_dir`. #[must_use] - pub fn new(generator: VoxelGenerator) -> Self { + pub fn new(generator: VoxelGenerator, region_dir: PathBuf) -> Self { let generator = Arc::new(generator); let (job_tx, job_rx) = crossbeam_channel::unbounded::(); let (result_tx, result_rx) = crossbeam_channel::unbounded::<(ChunkPos, Chunk)>(); - // One worker per logical core, falling back to a small fixed pool if the platform cannot report its parallelism. + // The actor owns every region file; workers reach it only through cloned request senders. + let save_actor = SaveActor::spawn(region_dir); + let worker_count = std::thread::available_parallelism().map_or(4, std::num::NonZero::get); let workers = (0..worker_count) .map(|_| { - // Each worker owns its own clones: a shared handle to the read-only generator, its own view of the shared job queue, and its own sender back into the result channel. + // Each worker shares a handle to the read-only generator, its own view of the shared job queue, its own sender back into the result channel, and its own request sender to the save actor. let generator = Arc::clone(&generator); let job_rx = job_rx.clone(); let result_tx = result_tx.clone(); + let save_tx = save_actor.sender(); std::thread::spawn(move || { // Block until a job arrives. while let Ok(pos) = job_rx.recv() { - let chunk = generator.generate_chunk(pos); - // A send error means the main thread has gone away; nothing left to do but let the worker wind down. + let chunk = load_chunk(&generator, &save_tx, pos); + // A send error means the main thread has gone away; nothing is left to do but let the worker wind down. if result_tx.send((pos, chunk)).is_err() { break; } @@ -75,7 +87,7 @@ impl ServerWorld { }) .collect(); - // Drop the template ends left over after cloning. + // Drop the template ends left over after cloning so the channels close once the real holders are gone. drop(job_rx); drop(result_tx); @@ -84,6 +96,7 @@ impl ServerWorld { job_tx, result_rx, in_flight: HashSet::new(), + save_actor, workers, } } @@ -94,33 +107,38 @@ impl ServerWorld { self.chunks.len() } + /// Returns the resident chunk at `pos`, or `None` if it is not currently loaded. + #[must_use] + pub fn chunk(&self, pos: ChunkPos) -> Option<&Chunk> { + self.chunks.get(&pos) + } + /// Number of chunks dispatched to the worker pool but not yet returned. #[must_use] pub fn in_flight_count(&self) -> usize { self.in_flight.len() } - /// Returns `true` when the worker pool has no outstanding generation work, i.e. every dispatched chunk has been returned. A loading gate can poll this to decide when an initial region has finished streaming. + /// Returns `true` when the worker pool has no outstanding work, i.e. every dispatched chunk has been returned. The loading gate polls this to decide when the initial region has finished streaming. #[must_use] pub fn streaming_idle(&self) -> bool { self.in_flight.is_empty() } - /// Reconciles resident chunks against a desired set without blocking the caller: finished chunks are drained from the worker pool, resident chunks absent from `desired` are evicted, and still-missing chunks are dispatched to the pool. + /// Reconciles resident chunks against the desired set without blocking the caller: finished chunks are drained from the worker pool, resident chunks absent from `desired` are evicted, and still-missing chunks are dispatched to the pool. pub fn reconcile(&mut self, desired: &HashSet) -> StreamStats { - // Drain: absorb every chunk the workers have finished since the last pass. + // Drain: absorb every chunk the workers finished since the last pass. let mut loaded = 0; while let Ok((pos, chunk)) = self.result_rx.try_recv() { - // The position is no longer dispatched now that its chunk has returned. self.in_flight.remove(&pos); - // Guard against the eviction race: the anchor may have moved away while this chunk was generating, so a returned chunk is only kept if it is still wanted. + // A finished chunk is only kept if it is still wanted. if desired.contains(&pos) { self.chunks.insert(pos, chunk); loaded += 1; } } - // Evict: drop resident chunks that no anchor wants any more. + // Evict: drop resident chunks no anchor wants any more. let stale: Vec = self .chunks .keys() @@ -131,11 +149,11 @@ impl ServerWorld { self.chunks.remove(pos); } - // Dispatch: request generation for every wanted position that is neither resident nor already in flight. + // Dispatch: request a load for every wanted position that is neither resident nor already in flight. for &pos in desired { if !self.chunks.contains_key(&pos) && !self.in_flight.contains(&pos) { self.in_flight.insert(pos); - // A send error means every worker has shut down; there is nothing useful to do with the position, so the failure is ignored. + // A send error means the workers shut down; nothing useful can be done with the position, so the failure is ignored. let _ = self.job_tx.send(pos); } } @@ -149,6 +167,43 @@ impl ServerWorld { } } +/// Resolves a chunk position to dense voxel data: a saved modification is applied over its baseline, otherwise the deterministic baseline is regenerated directly. +fn load_chunk(generator: &VoxelGenerator, save_tx: &Sender, pos: ChunkPos) -> Chunk { + match request_saved_chunk(save_tx, pos) { + Ok(Some(data)) => { + // A saved modification stores only edits, so the baseline is regenerated and the edits are layered on top. + // TODO: once worldgen versioning exists, the baseline must be regenerated at `data.worldgen_version()` rather than the current version; today there is a single version, so the current baseline matches. + data.materialize(&generator.generate_chunk(pos)) + } + // The chunk was never modified, so its content is exactly the deterministic baseline. + Ok(None) => generator.generate_chunk(pos), + Err(error) => { + // A save-layer failure must not wedge streaming; the chunk falls back to a fresh baseline and the error is logged. + warn!(?error, ?pos, "chunk load failed; regenerating baseline"); + generator.generate_chunk(pos) + } + } +} + +/// Sends a read request to the save actor and blocks for its reply, mapping a departed actor to an absent record so generation can still proceed. +fn request_saved_chunk( + save_tx: &Sender, + pos: ChunkPos, +) -> Result, SaveError> { + let (reply_tx, reply_rx) = crossbeam_channel::bounded(1); + if save_tx + .send(SaveRequest::Read { + pos, + reply: reply_tx, + }) + .is_err() + { + return Ok(None); + } + // A receive error means the actor dropped the reply end, treated the same as no saved data. + reply_rx.recv().unwrap_or(Ok(None)) +} + /// Inserts every chunk position inside the streaming cylinder around `center` into `out`. pub fn cylinder_chunks( center: ChunkPos, @@ -172,14 +227,17 @@ pub fn cylinder_chunks( #[cfg(test)] mod tests { - use super::{ChunkPos, ServerWorld, StreamStats, cylinder_chunks}; + use super::{ServerWorld, StreamStats, cylinder_chunks}; use shared::generator::{VoxelGenerator, WorldGenConfig}; - use shared::world::BlockId; + use shared::save::SaveError; + use shared::world::{BlockId, ChunkData, ChunkPos}; use std::collections::HashSet; use std::time::{Duration, Instant}; - /// Builds a server world backed by a real worker pool for streaming tests. - fn test_world() -> ServerWorld { + use crate::save::{RegionFile, region_path}; + + /// Builds a generator with a small, cheap terrain configuration for streaming tests. + fn test_generator() -> VoxelGenerator { let config = WorldGenConfig { base_height: 8, noise_scale: 0.05, @@ -187,10 +245,15 @@ mod tests { subsurface_block: BlockId(2), stone_block: BlockId(3), }; - ServerWorld::new(VoxelGenerator::new(config, 42)) + VoxelGenerator::new(config, 42) } - /// Repeatedly reconciles against `desired` until the worker pool reports no outstanding work, returning the final pass's stats. Fails the test if the pool does not drain within a fixed timeout. + /// Builds a server world whose saves resolve against `region_dir`. + fn test_world(region_dir: std::path::PathBuf) -> ServerWorld { + ServerWorld::new(test_generator(), region_dir) + } + + /// Repeatedly reconciles `desired` until the worker pool reports no outstanding work, returning the final pass's stats. Fails the test if the pool does not drain within a fixed timeout. fn drain_to_idle(world: &mut ServerWorld, desired: &HashSet) -> StreamStats { let deadline = Instant::now() + Duration::from_secs(5); loop { @@ -207,12 +270,14 @@ mod tests { } #[test] - fn reconcile_converges_over_multiple_passes() { - let mut world = test_world(); + fn reconcile_converges_over_multiple_passes() -> Result<(), SaveError> { + // A fresh empty directory means every load is a miss and resolves to the baseline. + let dir = tempfile::tempdir()?; + let mut world = test_world(dir.path().to_path_buf()); let mut desired = HashSet::new(); cylinder_chunks(ChunkPos::new(0, 0, 0), 2, &mut desired); - // The first pass only dispatches work; because generation is off-thread, nothing is resident yet and every position is in flight. + // The first pass only dispatches work; because loading is off-thread, nothing is resident yet and every position is in flight. let first = world.reconcile(&desired); assert_eq!(first.loaded, 0); assert_eq!(first.resident, 0); @@ -222,11 +287,13 @@ mod tests { let final_stats = drain_to_idle(&mut world, &desired); assert_eq!(final_stats.in_flight, 0); assert_eq!(final_stats.resident, desired.len()); + Ok(()) } #[test] - fn evicted_chunk_is_not_repopulated_on_arrival() { - let mut world = test_world(); + fn evicted_chunk_is_not_repopulated_on_arrival() -> Result<(), SaveError> { + let dir = tempfile::tempdir()?; + let mut world = test_world(dir.path().to_path_buf()); let target = ChunkPos::new(0, 0, 0); let mut desired = HashSet::new(); desired.insert(target); @@ -234,28 +301,54 @@ mod tests { // Dispatch the chunk, then immediately stop wanting it. world.reconcile(&desired); - // Every subsequent pass reconciles against an empty desired set, so whenever the worker returns the chunk the drain guard discards it rather than resurrecting an unwanted chunk. + // Every subsequent pass reconciles against an empty desired set, so the finished chunk is discarded on arrival rather than inserted. let empty = HashSet::new(); let final_stats = drain_to_idle(&mut world, &empty); assert_eq!(final_stats.in_flight, 0); - assert_eq!( - final_stats.resident, 0, - "a chunk that is no longer wanted must not become resident when it arrives" - ); + assert_eq!(final_stats.resident, 0); + Ok(()) } #[test] - fn cylinder_is_symmetric_and_bounded() { + fn saved_modification_is_applied_over_baseline() -> Result<(), SaveError> { + // A modified chunk is written to disk, then streamed back; the resident chunk must show the edit rather than the bare baseline. + let dir = tempfile::tempdir()?; + let pos = ChunkPos::new(0, 0, 0); + let edited_index = 100u32; + let edited_block = BlockId(999); + + let mut data = ChunkData::new(pos, 0); + data.set(edited_index, edited_block); + + let mut region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?; + region.write_chunk(pos, &data, 0)?; + region.save()?; + + let mut world = test_world(dir.path().to_path_buf()); + let mut desired = HashSet::new(); + desired.insert(pos); + drain_to_idle(&mut world, &desired); + + // The resident chunk must carry the stored edit layered over its regenerated baseline. + assert!( + world + .chunk(pos) + .is_some_and(|chunk| chunk.blocks[edited_index as usize] == edited_block) + ); + Ok(()) + } + + #[test] + fn cylinder_contains_expected_columns() { let mut set = HashSet::new(); cylinder_chunks(ChunkPos::new(0, 0, 0), 2, &mut set); - // Center column is always included. assert!(set.contains(&ChunkPos::new(0, 0, 0))); - // A corner outside the disc (dx=2, dz=2 -> 8 > 4) is excluded. + // A corner cell is outside the disc (dx=2, dz=2 -> 8 > 4). assert!(!set.contains(&ChunkPos::new(2, 0, 2))); // An axis cell at exactly the radius is included (dx=2, dz=0 -> 4 == 4). assert!(set.contains(&ChunkPos::new(2, 0, 0))); - // Vertical extent is radius/2 = 1, so y=2 is out of range. + // The vertical extent is radius/2 = 1, so y=2 is out of range. assert!(!set.contains(&ChunkPos::new(0, 2, 0))); assert!(set.contains(&ChunkPos::new(0, 1, 0))); } @@ -268,7 +361,7 @@ mod tests { let mut shifted = HashSet::new(); cylinder_chunks(ChunkPos::new(10, 0, -5), 3, &mut shifted); - // Shape is translation-invariant: same count regardless of center. + // The shape is translation-invariant: the same count regardless of center. assert_eq!(origin.len(), shifted.len()); } }