feat(server): generate chunks on worker pool

Cargo.lock

@@ -560,20 +560,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b"
 
 [[package]]
-name = "crossbeam-deque"
+name = "crossbeam-channel"
-version = "0.8.7"
+version = "0.5.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb"
+checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e"
-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"
 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",

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"

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"
     );
 }

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<VoxelGenerator>,
     /// Currently resident chunks, keyed by chunk-space position.
     chunks: HashMap<ChunkPos, Chunk>,
+    /// Sending end of the job channel; the main thread pushes positions that require generation.
+    job_tx: Sender<ChunkPos>,
+    /// 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<ChunkPos>,
+    /// 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<JoinHandle<()>>,
 }
 
 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::<ChunkPos>();
+        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<ChunkPos>) -> 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<ChunkPos> = self
             .chunks
             .keys()
@@ -66,29 +129,20 @@ impl ServerWorld {
             self.chunks.remove(pos);
         }
 
-        // Collect the desired positions that are not yet resident.
+        // Dispatch: request generation for every wanted position that is neither resident nor already in flight.
-        let pending: Vec<ChunkPos> = desired
+        for &pos in desired {
-            .iter()
+            if !self.chunks.contains_key(&pos) && !self.in_flight.contains(&pos) {
-            .filter(|pos| !self.chunks.contains_key(pos))
+                self.in_flight.insert(pos);
-            .copied()
+                // A send error means every worker has shut down; there is nothing useful to do with the position, so the failure is ignored.
-            .collect();
+                let _ = self.job_tx.send(pos);
-
+            }
-        // 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);
         }
 
         StreamStats {
             loaded,
             unloaded: stale.len(),
             resident: self.chunks.len(),
+            in_flight: self.in_flight.len(),
         }
     }
 }