chore: chunk gen should be better now

Cargo.lock

@@ -2765,6 +2765,7 @@ dependencies = [
  "hmac 0.13.0",
  "libloading",
  "num-bigint",
+ "num_cpus",
  "ordered-float",
  "pkcs8 0.11.0-rc.11",
  "postcard",
@@ -2776,7 +2777,8 @@ dependencies = [
  "pumpkin-protocol",
  "pumpkin-util",
  "pumpkin-world",
- "rand 0.10.0",
+ "rand 0.10.1",
+ "rayon",
  "rsa",
  "rustc-hash",
  "rustc_version_runtime",
@@ -2841,7 +2843,7 @@ dependencies = [
  "phf",
  "pumpkin-nbt",
  "pumpkin-util",
- "rand 0.10.0",
+ "rand 0.10.1",
  "serde",
 ]
 
@@ -2960,7 +2962,8 @@ dependencies = [
  "pumpkin-data",
  "pumpkin-nbt",
  "pumpkin-util",
- "rand 0.10.0",
+ "rand 0.10.1",
+ "rayon",
  "rustc-hash",
  "ruzstd",
  "serde",
@@ -3019,9 +3022,9 @@ dependencies = [
 
 [[package]]
 name = "rand"
-version = "0.10.0"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc266eb313df6c5c09c1c7b1fbe2510961e5bcd3add930c1e31f7ed9da0feff8"
+checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207"
 dependencies = [
  "chacha20",
  "getrandom 0.4.2",
@@ -3726,7 +3729,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.4.2",
+ "getrandom 0.3.4",
  "once_cell",
  "rustix 1.1.4",
  "windows-sys 0.61.2",

Cargo.toml

@@ -153,7 +153,7 @@ pumpkin-protocol = { path = "pumpkin-protocol" }
 pumpkin-util = { path = "pumpkin-util" }
 pumpkin-world = { path = "pumpkin-world" }
 quote = "1.0"
-rand = "0.10.0"
+rand = "0.10.1"
 rsa = "=0.10.0-rc.17"
 cipher = "0.5.1"
 rustc-hash = "2.1.2"

pumpkin-world/Cargo.toml

@@ -44,6 +44,7 @@ ruzstd.workspace = true
 
 itertools.workspace = true
 enum_dispatch.workspace = true
+rayon.workspace = true
 
 # thread_local.workspace = true
 # lru.workspace = true

pumpkin-world/src/chunk_system/schedule.rs

@@ -12,6 +12,7 @@ use crate::chunk::io::Dirtiable;
 use crate::level::{Level, SyncChunk};
 use dashmap::DashMap;
 use pumpkin_config::lighting::LightingEngineConfig;
+use pumpkin_data::chunk_gen_settings::GenerationSettings;
 use pumpkin_util::math::vector2::Vector2;
 use slotmap::Key;
 use std::cmp::{Ordering, max};
@@ -66,10 +67,14 @@ pub struct GenerationSchedule {
 
     io_lock: IOLock,
     running_task_count: u16,
+    max_in_flight: u16,
+    queue_dirty: bool,
     recv_chunk: crossfire::compat::MRx<(ChunkPos, RecvChunk)>,
-    io_read: crossfire::compat::MTx<ChunkPos>,
+    io_read: crossfire::compat::MTx<Vec<ChunkPos>>,
     io_write: crossfire::compat::Tx<Vec<(ChunkPos, Chunk)>>,
     generate: crossfire::compat::MTx<(ChunkPos, Cache, StagedChunkEnum)>,
+    send_chunk: crossfire::compat::MTx<(ChunkPos, RecvChunk)>,
+    gen_pool: Option<Arc<rayon::ThreadPool>>,
     listener: Arc<ChunkListener>,
     lighting_config: LightingEngineConfig,
 }
@@ -82,6 +87,7 @@ impl GenerationSchedule {
         level_channel: Arc<LevelChannel>,
         listener: Arc<ChunkListener>,
         thread_tracker: &mut Vec<thread::JoinHandle<()>>,
+        gen_pool: Option<Arc<rayon::ThreadPool>>,
     ) {
         let (send_chunk, recv_chunk) = crossfire::compat::mpmc::unbounded_blocking();
 
@@ -110,6 +116,7 @@ impl GenerationSchedule {
             io_lock.clone(),
         ));
 
+        if gen_pool.is_none() {
             for i in 0..gen_thread_count {
                 let recv_gen = recv_gen.clone();
                 let send_chunk = send_chunk.clone();
@@ -124,6 +131,13 @@ impl GenerationSchedule {
 
                 thread_tracker.push(handle);
             }
+        }
+
+        let max_in_flight = if gen_pool.is_some() {
+            (num_cpus::get() * 4) as u16
+        } else {
+            gen_thread_count as u16
+        };
 
         let level_sched = level;
         let lighting_config = level_sched.lighting_config;
@@ -141,10 +155,14 @@ impl GenerationSchedule {
                     waiting_for_chunks: HashSetType::default(),
                     io_lock,
                     running_task_count: 0,
+                    max_in_flight,
+                    queue_dirty: false,
                     recv_chunk,
                     io_read: send_read_io,
                     io_write: send_write_io,
                     generate: send_gen,
+                    send_chunk,
+                    gen_pool,
                     listener,
                     chunk_map: Default::default(),
                     lighting_config,
@@ -204,21 +222,21 @@ impl GenerationSchedule {
     }
 
     fn sort_queue(&mut self) {
-        let mut new_queue = BinaryHeap::with_capacity(self.queue.len());
+        if self.queue.is_empty() {
-        for i in &self.queue {
+            return;
+        }
+        let mut tasks: Vec<_> = self.queue.drain().collect();
+        for i in &mut tasks {
             if let Some(node) = self.graph.nodes.get(i.1) {
-                new_queue.push(TaskHeapNode(
+                i.0 = Self::calc_priority(
-                    Self::calc_priority(
                     &self.last_level,
                     &self.last_high_priority,
                     node.pos,
                     node.stage,
-                    ),
+                );
-                    i.1,
-                ));
             }
         }
-        self.queue = new_queue;
+        self.queue = BinaryHeap::from(tasks);
     }
 
     /// Ensure that the dependency chain for `req_stage` exists on `holder` (for chunk at
@@ -395,9 +413,9 @@ impl GenerationSchedule {
         }
         if let Some(high_priority) = new_data.1 {
             self.last_high_priority = high_priority;
+            self.queue_dirty = true;
         }
         let Some(new_level) = new_data.0 else {
-            self.sort_queue();
             return true;
         };
         for (pos, (old_stage, new_stage)) in new_level.0 {
@@ -503,7 +521,7 @@ impl GenerationSchedule {
             self.chunk_map.insert(pos, holder);
         }
         self.last_level = new_level.1;
-        self.sort_queue();
+        self.queue_dirty = true;
         true
     }
 
@@ -915,6 +933,22 @@ impl GenerationSchedule {
                 break;
             }
 
+            // 1. Get latest world state (player moves, etc)
+            self.resort_work(self.send_level.get());
+
+            // 2. Process all pending chunk results from workers
+            while let Ok((pos, data)) = self.recv_chunk.try_recv() {
+                self.receive_chunk(pos, data);
+            }
+
+            // 3. Re-sort if world state changed or new tasks added
+            if self.queue_dirty {
+                self.sort_queue();
+                self.queue_dirty = false;
+            }
+
+            // 4. Process ready tasks in the queue (up to max_in_flight)
+            let mut io_batch = Vec::with_capacity(16);
             'out2: while let Some(task) = self.queue.pop() {
                 if level.shut_down_chunk_system.load(Relaxed) {
                     self.queue.push(task);
@@ -923,13 +957,22 @@ impl GenerationSchedule {
                     break 'out2;
                 }
 
-                if self.resort_work(self.send_level.get()) {
+                if self.running_task_count >= self.max_in_flight {
                     self.queue.push(task);
                     break 'out2;
                 }
+
+                // Briefly check for high-priority results or world changes to avoid stalling
                 while let Ok((pos, data)) = self.recv_chunk.try_recv() {
                     self.receive_chunk(pos, data);
+                    if self.resort_work(self.send_level.get()) {
+                        // If world state changed, we MUST re-sort before continuing
+                        self.queue.push(task);
+                        self.queue_dirty = true;
+                        break 'out2;
                     }
+                }
+
                 if let Some(node) = self.graph.nodes.get_mut(task.1) {
                     if node.in_degree != 0 {
                         node.in_queue = false;
@@ -951,12 +994,24 @@ impl GenerationSchedule {
                         }
                         holder.occupied = occupy;
 
-                        if self.io_read.send(node.pos).is_err() {
+                        io_batch.push(node.pos);
+                        if io_batch.len() >= 16
+                            && self.io_read.send(std::mem::take(&mut io_batch)).is_err()
+                        {
                             info!("IO read thread closed, saving remaining chunks...");
                             self.save_all_chunk(true);
                             break 'out2;
                         }
                     } else {
+                        // Send any pending IO batch before starting generation
+                        if !io_batch.is_empty()
+                            && self.io_read.send(std::mem::take(&mut io_batch)).is_err()
+                        {
+                            info!("IO read thread closed, saving remaining chunks...");
+                            self.save_all_chunk(true);
+                            break 'out2;
+                        }
+
                         let write_radius = node.stage.get_write_radius();
 
                         // Pre-validate that every chunk in the write area (including the
@@ -1058,7 +1113,21 @@ impl GenerationSchedule {
                         }
 
                         self.running_task_count += 1;
-                        if self.generate.send((node.pos, cache, node.stage)).is_err() {
+                        if let Some(pool) = &self.gen_pool {
+                            let pos = node.pos;
+                            let stage = node.stage;
+                            let send_chunk = self.send_chunk.clone();
+                            let level = level.clone();
+                            let settings =
+                                GenerationSettings::from_dimension(&level.world_gen.dimension);
+
+                            pool.spawn(move || {
+                                let result = crate::chunk_system::worker_logic::run_generation(
+                                    pos, cache, stage, &level, settings,
+                                );
+                                let _ = send_chunk.send((pos, result));
+                            });
+                        } else if self.generate.send((node.pos, cache, node.stage)).is_err() {
                             self.running_task_count = self.running_task_count.saturating_sub(1);
                             info!("Generation thread closed, saving remaining chunks...");
                             self.save_all_chunk(true);
@@ -1068,25 +1137,40 @@ impl GenerationSchedule {
                 }
             }
 
+            // Flush any remaining IO batch
+            if !io_batch.is_empty() && self.io_read.send(std::mem::take(&mut io_batch)).is_err() {
+                info!("IO read thread closed, saving remaining chunks...");
+                self.save_all_chunk(true);
+            }
+
+            // 3. If queue is empty, wait for work or results
             if self.queue.is_empty() {
-                // Wait while there are in-flight tasks OR tasks parked waiting for chunk data.
+                // If we have tasks in flight, wait for them with timeout
-                while (self.running_task_count > 0 || !self.waiting_for_chunks.is_empty())
+                if self.running_task_count > 0 || !self.waiting_for_chunks.is_empty() {
-                    && self.queue.is_empty()
+                    match self.recv_chunk.recv_timeout(Duration::from_millis(5)) {
-                {
+                        Ok((pos, data)) => {
-                    if let Ok((pos, data)) = self.recv_chunk.try_recv() {
                             self.receive_chunk(pos, data);
                             self.resort_work(self.send_level.get());
+                        }
+                        Err(crossfire::compat::RecvTimeoutError::Timeout) => {
+                            // Periodically check LevelChannel for new requests
+                            self.resort_work(self.send_level.get());
+                        }
+                        Err(crossfire::compat::RecvTimeoutError::Disconnected) => break,
+                    }
+                    if self.queue_dirty {
+                        self.sort_queue();
+                        self.queue_dirty = false;
+                    }
                 } else {
-                        if level.shut_down_chunk_system.load(Relaxed) {
+                    // No tasks in flight, wait indefinitely for LevelChannel changes
-                            break;
-                        }
-                        thread::sleep(Duration::from_millis(50));
-                    }
-                }
-                if self.queue.is_empty() && self.waiting_for_chunks.is_empty() {
                     debug_assert!(self.debug_check());
                     debug_assert_eq!(self.running_task_count, 0);
                     self.resort_work(self.send_level.wait_and_get(&level));
+                    if self.queue_dirty {
+                        self.sort_queue();
+                        self.queue_dirty = false;
+                    }
                 }
             }
         }

pumpkin-world/src/chunk_system/worker_logic.rs

@@ -55,7 +55,7 @@ fn needs_relighting(chunk: &crate::chunk::ChunkData, config: &LightingEngineConf
 }
 
 pub async fn io_read_work(
-    recv: crossfire::compat::MAsyncRx<ChunkPos>,
+    recv: crossfire::compat::MAsyncRx<Vec<ChunkPos>>,
     send: crossfire::compat::MTx<(ChunkPos, RecvChunk)>,
     level: Arc<Level>,
     lock: IOLock,
@@ -64,23 +64,26 @@ pub async fn io_read_work(
     debug!("io read thread start");
     let biome_mixer_seed = hash_seed(level.world_gen.random_config.seed);
     let dimension = &level.world_gen.dimension;
-    let (t_send, mut t_recv) = tokio::sync::mpsc::channel(1);
+    let (t_send, mut t_recv) = tokio::sync::mpsc::channel(16);
 
     // Cleaner loop and async recv
-    while let Ok(pos) = recv.recv().await {
+    while let Ok(batch) = recv.recv().await {
+        for pos in &batch {
             // Lock handling
             tokio::task::block_in_place(|| {
                 let mut data = lock.0.lock().unwrap();
-            while data.contains_key(&pos) {
+                while data.contains_key(pos) {
                     data = lock.1.wait(data).unwrap();
                 }
             });
+        }
 
         level
             .chunk_saver
-            .fetch_chunks(&level.level_folder, &[pos], t_send.clone())
+            .fetch_chunks(&level.level_folder, &batch, t_send.clone())
             .await;
 
+        for _ in 0..batch.len() {
             let data = match t_recv.recv().await {
                 Some(res) => res,
                 None => break,
@@ -88,6 +91,7 @@ pub async fn io_read_work(
 
             match data {
                 Loaded(chunk) => {
+                    let pos = ChunkPos::new(chunk.x, chunk.z);
                     if chunk.status == ChunkStatus::Full {
                         // Relighting check
                         let needs_relight = needs_relighting(&chunk, &level.lighting_config);
@@ -108,7 +112,7 @@ pub async fn io_read_work(
                             // Clear all lighting data
                             let section_count = proto.light.sky_light.len();
                             proto.light.sky_light = (0..section_count)
-                            .map(|_| LightContainer::new_empty(0))
+                                .map(|_| LightContainer::new_empty(15))
                                 .collect();
                             proto.light.block_light = (0..section_count)
                                 .map(|_| LightContainer::new_empty(0))
@@ -134,7 +138,8 @@ pub async fn io_read_work(
                         }
                     } else {
                         // Standard ProtoChunk handling for non-full chunks
-                    let val = RecvChunk::IO(Chunk::Proto(Box::new(ProtoChunk::from_chunk_data(
+                        let val =
+                            RecvChunk::IO(Chunk::Proto(Box::new(ProtoChunk::from_chunk_data(
                                 &chunk,
                                 dimension,
                                 level.world_gen.default_block,
@@ -144,15 +149,8 @@ pub async fn io_read_work(
                             break;
                         }
                     }
-                continue;
                 }
-            LoadedData::Missing(_) => {}
+                LoadedData::Missing(pos) | LoadedData::Error((pos, _)) => {
-            LoadedData::Error(_) => {
-                warn!("chunk data read error pos: {pos:?}. regenerating");
-            }
-        }
-
-        // Final send for missing/error cases (regenerate)
                     if send
                         .send((
                             pos,
@@ -169,6 +167,9 @@ pub async fn io_read_work(
                         break;
                     }
                 }
+            }
+        }
+    }
     debug!("io read thread stop");
 }
 
@@ -226,21 +227,13 @@ pub async fn io_write_work(recv: AsyncRx<Vec<(ChunkPos, Chunk)>>, level: Arc<Lev
     }
 }
 
-pub fn generation_work(
+pub fn run_generation(
-    recv: crossfire::compat::MRx<(ChunkPos, Cache, StagedChunkEnum)>,
+    pos: ChunkPos,
-    send: crossfire::compat::MTx<(ChunkPos, RecvChunk)>,
+    mut cache: Cache,
-    level: Arc<Level>,
+    stage: StagedChunkEnum,
-) {
+    level: &Level,
-    let settings = GenerationSettings::from_dimension(&level.world_gen.dimension);
+    settings: &GenerationSettings,
-
+) -> RecvChunk {
-    loop {
-        let (pos, mut cache, stage) = if let Ok(data) = recv.recv() {
-            data
-        } else {
-            debug!("generation channel closed, exiting");
-            break;
-        };
-
     // Run generation with panic catching
     let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
         cache.advance(
@@ -258,11 +251,7 @@ pub fn generation_work(
     }));
 
     match result {
-            Ok(cache) => {
+        Ok(cache) => RecvChunk::Generation(cache),
-                if send.send((pos, RecvChunk::Generation(cache))).is_err() {
-                    break;
-                }
-            }
         Err(payload) => {
             let msg = payload
                 .downcast_ref::<&str>()
@@ -276,16 +265,33 @@ pub fn generation_work(
 
             error!("Chunk generation FAILED at {pos:?} ({stage:?}): {msg}");
 
-                // Send failure notification
-                let _ = send.send((
-                    pos,
             RecvChunk::GenerationFailure {
                 pos,
                 stage,
                 error: msg.to_string(),
-                    },
-                ));
             }
         }
     }
 }
+
+pub fn generation_work(
+    recv: crossfire::compat::MRx<(ChunkPos, Cache, StagedChunkEnum)>,
+    send: crossfire::compat::MTx<(ChunkPos, RecvChunk)>,
+    level: Arc<Level>,
+) {
+    let settings = GenerationSettings::from_dimension(&level.world_gen.dimension);
+
+    loop {
+        let (pos, cache, stage) = if let Ok(data) = recv.recv() {
+            data
+        } else {
+            debug!("generation channel closed, exiting");
+            break;
+        };
+
+        let result = run_generation(pos, cache, stage, &level, settings);
+        if send.send((pos, result)).is_err() {
+            break;
+        }
+    }
+}

pumpkin-world/src/dimension.rs

@@ -11,6 +11,7 @@ pub fn into_level(
     mut base_directory: PathBuf,
     block_registry: Arc<dyn BlockRegistryExt>,
     seed: i64,
+    gen_pool: Option<Arc<rayon::ThreadPool>>,
 ) -> Arc<Level> {
     if dimension == Dimension::OVERWORLD {
     } else if dimension == Dimension::THE_NETHER {
@@ -24,5 +25,6 @@ pub fn into_level(
         block_registry,
         seed,
         dimension,
+        gen_pool,
     )
 }

pumpkin-world/src/level.rs

@@ -14,7 +14,6 @@ use crate::{
     tick::{OrderedTick, ScheduledTick, TickPriority},
     world::BlockRegistryExt,
 };
-use crossbeam::channel::Sender;
 use dashmap::{DashMap, Entry};
 use pumpkin_config::{chunk::ChunkConfig, lighting::LightingEngineConfig, world::LevelConfig};
 use pumpkin_data::biome::Biome;
@@ -96,12 +95,12 @@ pub struct Level {
     /// Number of ticks between autosave checks. If 0, autosave is disabled.
     pub autosave_ticks: u64,
 
-    gen_entity_request_tx: Sender<Vector2<i32>>,
     pending_entity_generations: Arc<DashMap<Vector2<i32>, Vec<oneshot::Sender<SyncEntityChunk>>>>,
 
     pub level_channel: Arc<LevelChannel>,
     pub thread_tracker: Mutex<Vec<thread::JoinHandle<()>>>,
     pub chunk_listener: Arc<ChunkListener>,
+    pub gen_pool: Option<Arc<rayon::ThreadPool>>,
 }
 
 pub struct TickData {
@@ -145,6 +144,7 @@ impl Level {
         block_registry: Arc<dyn BlockRegistryExt>,
         seed: i64,
         dimension: Dimension,
+        gen_pool: Option<Arc<rayon::ThreadPool>>,
     ) -> Arc<Self> {
         let region_folder = root_folder.join("region");
         let entities_folder = root_folder.join("entities");
@@ -178,7 +178,6 @@ impl Level {
             >::new(config.clone())),
         };
 
-        let (gen_entity_request_tx, gen_entity_request_rx) = crossbeam::channel::unbounded();
         let pending_entity_generations = Arc::new(DashMap::new());
         let level_channel = Arc::new(LevelChannel::new());
         let thread_tracker = Mutex::new(Vec::new());
@@ -205,44 +204,55 @@ impl Level {
             should_save: AtomicBool::new(false),
             should_unload: AtomicBool::new(false),
             autosave_ticks: level_config.autosave_ticks,
-            gen_entity_request_tx,
             pending_entity_generations: pending_entity_generations.clone(),
             level_channel: level_channel.clone(),
             thread_tracker,
             chunk_listener: listener.clone(),
+            gen_pool: gen_pool.clone(),
         });
 
         // TODO
         let total_cores = num_cpus::get().saturating_sub(2).max(1);
         let threads_per_dimension = (total_cores / 2).max(1);
-        let entity_threads = 1;
 
         GenerationSchedule::create(
-            2,
+            4,
             threads_per_dimension,
             level_ref.clone(),
             level_channel,
             listener,
             level_ref.thread_tracker.lock().unwrap().as_mut(),
+            gen_pool,
         );
 
-        let mut tracker_lock = level_ref.thread_tracker.lock().unwrap();
+        level_ref
-
-        for thread_id in 0..entity_threads {
-            let level_clone = level_ref.clone();
-            let pending_clone = pending_entity_generations.clone();
-            let rx = gen_entity_request_rx.clone();
-
-            let handle = thread::Builder::new()
-                .name(format!("Entity Chunk Generation Thread {thread_id}"))
-                .spawn(move || {
-                    loop {
-                        if level_clone.cancel_token.is_cancelled() {
-                            break;
     }
 
-                        match rx.recv_timeout(std::time::Duration::from_millis(500)) {
+    pub fn spawn_entity_generation(self: &Arc<Self>, pos: Vector2<i32>) {
-                            Ok(pos) => {
+        let level = self.clone();
+        if let Some(pool) = &self.gen_pool {
+            pool.spawn(move || {
+                let arc_chunk = Arc::new(ChunkEntityData {
+                    x: pos.x,
+                    z: pos.y,
+                    data: tokio::sync::Mutex::new(FxHashMap::default()),
+                    dirty: AtomicBool::new(true),
+                });
+
+                level.loaded_entity_chunks.insert(pos, arc_chunk.clone());
+
+                if let Some((_, waiters)) = level.pending_entity_generations.remove(&pos) {
+                    for tx in waiters {
+                        let _ = tx.send(arc_chunk.clone());
+                    }
+                }
+            });
+        } else {
+            // Fallback to spawning a new thread if no pool is available (should not happen in production)
+            let level_clone = level.clone();
+            thread::Builder::new()
+                .name(format!("Entity Gen {:?}", pos))
+                .spawn(move || {
                     let arc_chunk = Arc::new(ChunkEntityData {
                         x: pos.x,
                         z: pos.y,
@@ -254,24 +264,15 @@ impl Level {
                         .loaded_entity_chunks
                         .insert(pos, arc_chunk.clone());
 
-                                if let Some((_, waiters)) = pending_clone.remove(&pos) {
+                    if let Some((_, waiters)) = level_clone.pending_entity_generations.remove(&pos)
+                    {
                         for tx in waiters {
                             let _ = tx.send(arc_chunk.clone());
                         }
                     }
-                            }
-                            Err(crossbeam::channel::RecvTimeoutError::Timeout) => continue,
-                            Err(crossbeam::channel::RecvTimeoutError::Disconnected) => break,
-                        }
-                    }
                 })
-                .expect("Failed to spawn Entity Generation Thread");
+                .expect("Failed to spawn entity generation thread");
-
-            tracker_lock.push(handle);
         }
-
-        drop(tracker_lock);
-        level_ref
     }
 
     /// Spawns a task associated with this world. All tasks spawned with this method are awaited
@@ -645,7 +646,7 @@ impl Level {
                                         }
                                         dashmap::mapref::entry::Entry::Vacant(entry) => {
                                             entry.insert(vec![tx]);
-                                            let _ = level_clone.gen_entity_request_tx.send(pos);
+                                            level_clone.spawn_entity_generation(pos);
                                         }
                                     }
                                     if let Ok(chunk) = rx.await {
@@ -683,7 +684,7 @@ impl Level {
                 }
                 dashmap::mapref::entry::Entry::Vacant(entry) => {
                     entry.insert(vec![tx]);
-                    let _ = self.gen_entity_request_tx.send(pos);
+                    self.spawn_entity_generation(pos);
                 }
             }
             rx.await.expect("Entity generation worker dropped")

pumpkin/Cargo.toml

@@ -19,6 +19,8 @@ doctest = false
 [dependencies]
 # pumpkin
 arc-swap.workspace = true
+rayon.workspace = true
+num_cpus.workspace = true
 pumpkin-util.workspace = true
 pumpkin-nbt.workspace = true
 pumpkin-config.workspace = true

pumpkin/src/entity/player.rs

@@ -109,8 +109,6 @@ const MAX_PREVIOUS_MESSAGES: u8 = 20; // Vanilla: 20
 pub const DATA_VERSION: i32 = 4786; // 26.1
 
 enum BatchState {
-    Initial,
-    Waiting,
     Count(u8),
 }
 
@@ -168,7 +166,7 @@ impl ChunkManager {
             chunk_sent: HashMap::new(),
             chunk_queue: BinaryHeap::new(),
             entity_chunk_queue: VecDeque::new(),
-            batches_sent_since_ack: BatchState::Initial,
+            batches_sent_since_ack: BatchState::Count(0),
             last_chunk_batch_sent_at: Instant::now(),
             world,
         }
@@ -191,8 +189,6 @@ impl ChunkManager {
     const fn ack_window_open(&self) -> bool {
         match self.batches_sent_since_ack {
             BatchState::Count(count) => count < Self::NOTCHIAN_BATCHES_WITHOUT_ACK_UNTIL_PAUSE,
-            BatchState::Initial => true,
-            BatchState::Waiting => false,
         }
     }
 
@@ -203,7 +199,7 @@ impl ChunkManager {
     }
 
     pub fn pull_new_chunks(&mut self) {
-        // log::debug!("pull new chunks");
+        let mut new_nodes = Vec::new();
         while let Ok((pos, chunk)) = self.chunk_listener.try_recv() {
             let dst = (pos.x - self.center.x)
                 .abs()
@@ -212,12 +208,19 @@ impl ChunkManager {
                 continue;
             }
             if self.should_enqueue_chunk(pos, &chunk) {
-                // log::debug!("receive new chunk {pos:?}");
+                new_nodes.push(HeapNode(dst, pos, chunk));
-                self.chunk_queue.push(HeapNode(dst, pos, chunk));
+            }
+        }
+
+        if !new_nodes.is_empty() {
+            if self.chunk_queue.is_empty() {
+                self.chunk_queue = BinaryHeap::from(new_nodes);
+            } else {
+                for node in new_nodes {
+                    self.chunk_queue.push(node);
+                }
             }
         }
-        // log::debug!("chunk_queue size {}", self.chunk_queue.len());
-        // log::debug!("chunk_sent size {}", self.chunk_sent.len());
     }
 
     pub fn update_center_and_view_distance(
@@ -257,23 +260,28 @@ impl ChunkManager {
                 && !unloading_chunks.contains(pos)
         });
 
-        let mut new_queue = BinaryHeap::with_capacity(self.chunk_queue.len());
+        let mut tasks: Vec<_> = self.chunk_queue.drain().collect();
-        for node in self.chunk_queue.drain() {
+        tasks.retain(|node| {
             let dst = (node.1.x - center.x).abs().max((node.1.y - center.y).abs());
-            if dst <= view_distance_i32 && !unloading_chunks.contains(&node.1) {
+            dst <= view_distance_i32 && !unloading_chunks.contains(&node.1)
-                new_queue.push(HeapNode(dst, node.1, node.2));
+        });
+        for node in &mut tasks {
+            node.0 = (node.1.x - center.x).abs().max((node.1.y - center.y).abs());
         }
-        }
-        self.chunk_queue = new_queue;
 
         for pos in loading_chunks {
             if !self.chunk_sent.contains_key(pos)
                 && let Some(chunk) = level.loaded_chunks.get(pos)
             {
-                self.push_chunk(*pos, chunk.value().clone());
+                let chunk = chunk.value().clone();
+                if self.should_enqueue_chunk(*pos, &chunk) {
+                    let dst = (pos.x - center.x).abs().max((pos.y - center.y).abs());
+                    tasks.push(HeapNode(dst, *pos, chunk));
                 }
             }
         }
+        self.chunk_queue = BinaryHeap::from(tasks);
+    }
 
     pub fn clean_up(&mut self, level: &Arc<Level>) {
         let mut lock = level.chunk_loading.lock().unwrap();
@@ -290,7 +298,7 @@ impl ChunkManager {
         self.chunk_sent.clear();
         self.chunk_queue.clear();
         self.entity_chunk_queue.clear();
-        self.batches_sent_since_ack = BatchState::Initial;
+        self.batches_sent_since_ack = BatchState::Count(0);
         self.last_chunk_batch_sent_at = Instant::now();
     }
 
@@ -307,7 +315,7 @@ impl ChunkManager {
         self.chunk_queue.clear();
         self.world = new_world;
         // Reset batch state so chunks can be sent immediately in the new dimension
-        self.batches_sent_since_ack = BatchState::Initial;
+        self.batches_sent_since_ack = BatchState::Count(0);
         self.last_chunk_batch_sent_at = Instant::now();
     }
 
@@ -347,8 +355,6 @@ impl ChunkManager {
             BatchState::Count(count) => {
                 *count = count.saturating_add(1);
             }
-            state @ BatchState::Initial => *state = BatchState::Waiting,
-            BatchState::Waiting => (),
         }
         self.last_chunk_batch_sent_at = Instant::now();
 
@@ -371,8 +377,6 @@ impl ChunkManager {
             BatchState::Count(count) => {
                 *count = count.saturating_add(1);
             }
-            state @ BatchState::Initial => *state = BatchState::Waiting,
-            BatchState::Waiting => (),
         }
         self.last_chunk_batch_sent_at = Instant::now();
 

pumpkin/src/server/mod.rs

@@ -254,6 +254,15 @@ impl Server {
         };
         let server = Arc::new(server);
 
+        let total_cores = num_cpus::get().saturating_sub(2).max(1);
+        let gen_pool = Arc::new(
+            rayon::ThreadPoolBuilder::new()
+                .num_threads(total_cores)
+                .thread_name(|i| format!("Gen-Pool-{i}"))
+                .build()
+                .expect("Failed to build generation thread pool"),
+        );
+
         let server_clone = server.clone();
         tokio::spawn(async move {
             server_clone
@@ -268,6 +277,7 @@ impl Server {
             let l_info = server.level_info.clone(); // Access from struct
             let weak = Arc::downgrade(&server);
             let config = Arc::new(server.advanced_config.world.clone());
+            let pool = gen_pool.clone();
 
             tokio::task::spawn_blocking(move || {
                 info!(
@@ -277,7 +287,7 @@ impl Server {
                         .to_pretty_console()
                 );
                 World::load(
-                    into_level(dim, &config, path, registry.clone(), seed),
+                    into_level(dim, &config, path, registry.clone(), seed, Some(pool)),
                     l_info,
                     dim,
                     registry,