diff --git a/Cargo.lock b/Cargo.lock index e0b0a95..020db95 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -560,20 +560,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" [[package]] -name = "crossbeam-deque" -version = "0.8.7" +name = "crossbeam-channel" +version = "0.5.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" -dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", -] - -[[package]] -name = "crossbeam-epoch" -version = "0.9.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" dependencies = [ "crossbeam-utils", ] @@ -661,12 +651,6 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d8b14ccef22fc6f5a8f4d7d768562a182c04ce9a3b3157b91390b52ddfdf1a76" -[[package]] -name = "either" -version = "1.16.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" - [[package]] name = "equivalent" version = "1.0.2" @@ -1623,26 +1607,6 @@ dependencies = [ "raw-window-handle", ] -[[package]] -name = "rayon" -version = "1.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" -dependencies = [ - "either", - "rayon-core", -] - -[[package]] -name = "rayon-core" -version = "1.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" -dependencies = [ - "crossbeam-deque", - "crossbeam-utils", -] - [[package]] name = "redox_syscall" version = "0.4.1" @@ -1819,8 +1783,8 @@ name = "server" version = "0.1.0" dependencies = [ "bevy_ecs", + "crossbeam-channel", "glam 0.27.0", - "rayon", "serde_json", "shared", "tracing", diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index ae4aad7..87da38d 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -10,8 +10,8 @@ workspace = true [dependencies] bevy_ecs = "0.19" +crossbeam-channel = "0.5.16" glam = "0.27" -rayon = "1.12.0" serde_json = "1.0.149" shared = { version = "0.1.0", path = "../shared" } tracing = "0.1.44" diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index aa7c210..6e8d1e4 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -38,6 +38,7 @@ fn stream_chunks( loaded = stats.loaded, unloaded = stats.unloaded, resident = stats.resident, + in_flight = stats.in_flight, "streaming reconcile" ); } diff --git a/crates/server/src/world_server.rs b/crates/server/src/world_server.rs index 6b6e1f5..bed997b 100644 --- a/crates/server/src/world_server.rs +++ b/crates/server/src/world_server.rs @@ -3,12 +3,16 @@ //! Authoritative chunk storage and generation logic for the server. use bevy_ecs::prelude::Resource; -use rayon::prelude::*; +use crossbeam_channel::{Receiver, Sender}; use shared::{ generator::VoxelGenerator, world::{Chunk, ChunkPos}, }; -use std::collections::{HashMap, HashSet}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + thread::JoinHandle, +}; /// Outcome of a single streaming reconcile pass, for logging and tests. #[derive(Copy, Clone, Debug, PartialEq, Eq)] @@ -19,24 +23,71 @@ pub struct StreamStats { pub unloaded: usize, /// Number of chunks resident after the pass. pub resident: usize, + /// Number of chunks in flight. + pub in_flight: usize, } /// The server's authoritative representation of the world. #[derive(Resource)] pub struct ServerWorld { /// Deterministic voxel source used to generate missing chunks on demand. - generator: VoxelGenerator, + generator: Arc, /// Currently resident chunks, keyed by chunk-space position. chunks: HashMap, + /// Sending end of the job channel; the main thread pushes positions that require generation. + 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. + in_flight: HashSet, + /// 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)] + workers: Vec>, } impl ServerWorld { /// Initializes a new authoritative server world with the provided generator. #[must_use] pub fn new(generator: VoxelGenerator) -> 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. + 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. + let generator = Arc::clone(&generator); + let job_rx = job_rx.clone(); + let result_tx = result_tx.clone(); + + 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. + if result_tx.send((pos, chunk)).is_err() { + break; + } + } + }) + }) + .collect(); + + // Drop the template ends left over after cloning. + drop(job_rx); + drop(result_tx); + Self { generator, chunks: HashMap::new(), + job_tx, + result_rx, + in_flight: HashSet::new(), + workers, } } @@ -53,9 +104,21 @@ impl ServerWorld { self.chunks.len() } - /// Reconciles resident chunks against a desired set: generates the chunks present in `desired` but not resident, and evicts resident chunks absent from `desired`. + /// 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. pub fn reconcile(&mut self, desired: &HashSet) -> StreamStats { - // Evict resident chunks that no anchor wants any more. + // Drain: absorb every chunk the workers have 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. + if desired.contains(&pos) { + self.chunks.insert(pos, chunk); + loaded += 1; + } + } + + // Evict: drop resident chunks that no anchor wants any more. let stale: Vec = self .chunks .keys() @@ -66,29 +129,20 @@ impl ServerWorld { self.chunks.remove(pos); } - // Collect the desired positions that are not yet resident. - let pending: Vec = desired - .iter() - .filter(|pos| !self.chunks.contains_key(pos)) - .copied() - .collect(); - - // Generate the missing chunks in parallel - let generator = &self.generator; - let generated: Vec<(ChunkPos, Chunk)> = pending - .par_iter() - .map(|&pos| (pos, generator.generate_chunk(pos))) - .collect(); - - let loaded = generated.len(); - for (pos, chunk) in generated { - self.chunks.insert(pos, chunk); + // Dispatch: request generation 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. + let _ = self.job_tx.send(pos); + } } StreamStats { loaded, unloaded: stale.len(), resident: self.chunks.len(), + in_flight: self.in_flight.len(), } } }