diff --git a/Cargo.lock b/Cargo.lock index 5691250b8..6693fe38b 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index 1738cb113..9155441aa 100644 --- a/Cargo.toml +++ b/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" diff --git a/pumpkin-world/Cargo.toml b/pumpkin-world/Cargo.toml index ad2433b57..462488d83 100644 --- a/pumpkin-world/Cargo.toml +++ b/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 diff --git a/pumpkin-world/src/chunk_system/schedule.rs b/pumpkin-world/src/chunk_system/schedule.rs index a83a401d2..8114e00f7 100644 --- a/pumpkin-world/src/chunk_system/schedule.rs +++ b/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, + io_read: crossfire::compat::MTx>, io_write: crossfire::compat::Tx>, generate: crossfire::compat::MTx<(ChunkPos, Cache, StagedChunkEnum)>, + send_chunk: crossfire::compat::MTx<(ChunkPos, RecvChunk)>, + gen_pool: Option>, listener: Arc, lighting_config: LightingEngineConfig, } @@ -82,6 +87,7 @@ impl GenerationSchedule { level_channel: Arc, listener: Arc, thread_tracker: &mut Vec>, + gen_pool: Option>, ) { let (send_chunk, recv_chunk) = crossfire::compat::mpmc::unbounded_blocking(); @@ -110,21 +116,29 @@ impl GenerationSchedule { io_lock.clone(), )); - for i in 0..gen_thread_count { - let recv_gen = recv_gen.clone(); - let send_chunk = send_chunk.clone(); - let level_clone = level.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(); + let level_clone = level.clone(); - let handle = thread::Builder::new() - .name(format!("Gen-{i}")) - .spawn(move || { - generation_work(recv_gen, send_chunk, level_clone); - }) - .expect("Failed to spawn Generation Thread"); + let handle = thread::Builder::new() + .name(format!("Gen-{i}")) + .spawn(move || { + generation_work(recv_gen, send_chunk, level_clone); + }) + .expect("Failed to spawn Generation Thread"); - thread_tracker.push(handle); + 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; let handle = thread::Builder::new() @@ -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()); - for i in &self.queue { + if self.queue.is_empty() { + 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( - Self::calc_priority( - &self.last_level, - &self.last_high_priority, - node.pos, - node.stage, - ), - i.1, - )); + i.0 = Self::calc_priority( + &self.last_level, + &self.last_high_priority, + node.pos, + node.stage, + ); } } - 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. - while (self.running_task_count > 0 || !self.waiting_for_chunks.is_empty()) - && self.queue.is_empty() - { - if let Ok((pos, data)) = self.recv_chunk.try_recv() { - self.receive_chunk(pos, data); - self.resort_work(self.send_level.get()); - } else { - if level.shut_down_chunk_system.load(Relaxed) { - break; + // If we have tasks in flight, wait for them with timeout + if self.running_task_count > 0 || !self.waiting_for_chunks.is_empty() { + match self.recv_chunk.recv_timeout(Duration::from_millis(5)) { + Ok((pos, data)) => { + self.receive_chunk(pos, data); + self.resort_work(self.send_level.get()); } - thread::sleep(Duration::from_millis(50)); + 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.is_empty() && self.waiting_for_chunks.is_empty() { + if self.queue_dirty { + self.sort_queue(); + self.queue_dirty = false; + } + } else { + // No tasks in flight, wait indefinitely for LevelChannel changes 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; + } } } } diff --git a/pumpkin-world/src/chunk_system/worker_logic.rs b/pumpkin-world/src/chunk_system/worker_logic.rs index 60e4fea60..31f629fc5 100644 --- a/pumpkin-world/src/chunk_system/worker_logic.rs +++ b/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, + recv: crossfire::compat::MAsyncRx>, send: crossfire::compat::MTx<(ChunkPos, RecvChunk)>, level: Arc, lock: IOLock, @@ -64,109 +64,110 @@ 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 { - // Lock handling - tokio::task::block_in_place(|| { - let mut data = lock.0.lock().unwrap(); - while data.contains_key(&pos) { - data = lock.1.wait(data).unwrap(); - } - }); + 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) { + 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; - let data = match t_recv.recv().await { - Some(res) => res, - None => break, - }; + for _ in 0..batch.len() { + let data = match t_recv.recv().await { + Some(res) => res, + None => break, + }; - match data { - Loaded(chunk) => { - if chunk.status == ChunkStatus::Full { - // Relighting check - let needs_relight = needs_relighting(&chunk, &level.lighting_config); + 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); - if needs_relight { - debug!( - "Chunk {pos:?} has uniform lighting, downgrading to Features stage for relighting" - ); + if needs_relight { + debug!( + "Chunk {pos:?} has uniform lighting, downgrading to Features stage for relighting" + ); - // Create ProtoChunk using the async method - let mut proto = ProtoChunk::from_chunk_data( - &chunk, - dimension, - level.world_gen.default_block, - biome_mixer_seed, - ); + // Create ProtoChunk using the async method + let mut proto = ProtoChunk::from_chunk_data( + &chunk, + dimension, + level.world_gen.default_block, + biome_mixer_seed, + ); - // Clear all lighting data - let section_count = proto.light.sky_light.len(); - proto.light.sky_light = (0..section_count) - .map(|_| LightContainer::new_empty(0)) - .collect(); - proto.light.block_light = (0..section_count) - .map(|_| LightContainer::new_empty(0)) - .collect(); + // Clear all lighting data + let section_count = proto.light.sky_light.len(); + proto.light.sky_light = (0..section_count) + .map(|_| LightContainer::new_empty(15)) + .collect(); + proto.light.block_light = (0..section_count) + .map(|_| LightContainer::new_empty(0)) + .collect(); - // Set stage to Features - proto.stage = StagedChunkEnum::Features; + // Set stage to Features + proto.stage = StagedChunkEnum::Features; - if send - .send((pos, RecvChunk::IO(Chunk::Proto(Box::new(proto))))) - .is_err() - { - break; + if send + .send((pos, RecvChunk::IO(Chunk::Proto(Box::new(proto))))) + .is_err() + { + break; + } + } else { + // Send fully valid chunk + if send + .send((pos, RecvChunk::IO(Chunk::Level(chunk)))) + .is_err() + { + break; + } } } else { - // Send fully valid chunk - if send - .send((pos, RecvChunk::IO(Chunk::Level(chunk)))) - .is_err() - { + // Standard ProtoChunk handling for non-full chunks + let val = + RecvChunk::IO(Chunk::Proto(Box::new(ProtoChunk::from_chunk_data( + &chunk, + dimension, + level.world_gen.default_block, + biome_mixer_seed, + )))); + if send.send((pos, val)).is_err() { break; } } - } else { - // Standard ProtoChunk handling for non-full chunks - let val = RecvChunk::IO(Chunk::Proto(Box::new(ProtoChunk::from_chunk_data( - &chunk, - dimension, - level.world_gen.default_block, - biome_mixer_seed, - )))); - if send.send((pos, val)).is_err() { + } + LoadedData::Missing(pos) | LoadedData::Error((pos, _)) => { + if send + .send(( + pos, + RecvChunk::IO(Chunk::Proto(Box::new(ProtoChunk::new( + pos.x, + pos.y, + dimension, + level.world_gen.default_block, + biome_mixer_seed, + )))), + )) + .is_err() + { break; } } - continue; } - LoadedData::Missing(_) => {} - LoadedData::Error(_) => { - warn!("chunk data read error pos: {pos:?}. regenerating"); - } - } - - // Final send for missing/error cases (regenerate) - if send - .send(( - pos, - RecvChunk::IO(Chunk::Proto(Box::new(ProtoChunk::new( - pos.x, - pos.y, - dimension, - level.world_gen.default_block, - biome_mixer_seed, - )))), - )) - .is_err() - { - break; } } debug!("io read thread stop"); @@ -226,6 +227,53 @@ pub async fn io_write_work(recv: AsyncRx>, level: Arc RecvChunk { + // Run generation with panic catching + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + cache.advance( + stage, + &level.lighting_config, + level.block_registry.as_ref(), + settings, + &level.world_gen.random_config, + &level.world_gen.terrain_cache, + &level.world_gen.base_router, + level.world_gen.dimension, + &level.world_gen.global_structure_cache, + ); + cache // Return cache on success + })); + + match result { + Ok(cache) => RecvChunk::Generation(cache), + Err(payload) => { + let msg = payload + .downcast_ref::<&str>() + .copied() + .or_else(|| { + payload + .downcast_ref::() + .map(std::string::String::as_str) + }) + .unwrap_or("Unknown panic payload"); + + error!("Chunk generation FAILED at {pos:?} ({stage:?}): {msg}"); + + 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)>, @@ -234,58 +282,16 @@ pub fn generation_work( let settings = GenerationSettings::from_dimension(&level.world_gen.dimension); loop { - let (pos, mut cache, stage) = if let Ok(data) = recv.recv() { + let (pos, 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( - stage, - &level.lighting_config, - level.block_registry.as_ref(), - settings, - &level.world_gen.random_config, - &level.world_gen.terrain_cache, - &level.world_gen.base_router, - level.world_gen.dimension, - &level.world_gen.global_structure_cache, - ); - cache // Return cache on success - })); - - match result { - Ok(cache) => { - if send.send((pos, RecvChunk::Generation(cache))).is_err() { - break; - } - } - Err(payload) => { - let msg = payload - .downcast_ref::<&str>() - .copied() - .or_else(|| { - payload - .downcast_ref::() - .map(std::string::String::as_str) - }) - .unwrap_or("Unknown panic payload"); - - error!("Chunk generation FAILED at {pos:?} ({stage:?}): {msg}"); - - // Send failure notification - let _ = send.send(( - pos, - RecvChunk::GenerationFailure { - pos, - stage, - error: msg.to_string(), - }, - )); - } + let result = run_generation(pos, cache, stage, &level, settings); + if send.send((pos, result)).is_err() { + break; } } } diff --git a/pumpkin-world/src/dimension.rs b/pumpkin-world/src/dimension.rs index 5d5c786ef..99f5bd8aa 100644 --- a/pumpkin-world/src/dimension.rs +++ b/pumpkin-world/src/dimension.rs @@ -11,6 +11,7 @@ pub fn into_level( mut base_directory: PathBuf, block_registry: Arc, seed: i64, + gen_pool: Option>, ) -> Arc { if dimension == Dimension::OVERWORLD { } else if dimension == Dimension::THE_NETHER { @@ -24,5 +25,6 @@ pub fn into_level( block_registry, seed, dimension, + gen_pool, ) } diff --git a/pumpkin-world/src/level.rs b/pumpkin-world/src/level.rs index e85d2bb5a..764c8bd05 100644 --- a/pumpkin-world/src/level.rs +++ b/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>, pending_entity_generations: Arc, Vec>>>, pub level_channel: Arc, pub thread_tracker: Mutex>>, pub chunk_listener: Arc, + pub gen_pool: Option>, } pub struct TickData { @@ -145,6 +144,7 @@ impl Level { block_registry: Arc, seed: i64, dimension: Dimension, + gen_pool: Option>, ) -> Arc { 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,73 +204,75 @@ 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(); + pub fn spawn_entity_generation(self: &Arc, pos: Vector2) { + 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), + }); - let handle = thread::Builder::new() - .name(format!("Entity Chunk Generation Thread {thread_id}")) + 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 || { - loop { - if level_clone.cancel_token.is_cancelled() { - break; - } + let arc_chunk = Arc::new(ChunkEntityData { + x: pos.x, + z: pos.y, + data: tokio::sync::Mutex::new(FxHashMap::default()), + dirty: AtomicBool::new(true), + }); - match rx.recv_timeout(std::time::Duration::from_millis(500)) { - Ok(pos) => { - let arc_chunk = Arc::new(ChunkEntityData { - x: pos.x, - z: pos.y, - data: tokio::sync::Mutex::new(FxHashMap::default()), - dirty: AtomicBool::new(true), - }); + level_clone + .loaded_entity_chunks + .insert(pos, arc_chunk.clone()); - level_clone - .loaded_entity_chunks - .insert(pos, arc_chunk.clone()); - - if let Some((_, waiters)) = pending_clone.remove(&pos) { - for tx in waiters { - let _ = tx.send(arc_chunk.clone()); - } - } - } - Err(crossbeam::channel::RecvTimeoutError::Timeout) => continue, - Err(crossbeam::channel::RecvTimeoutError::Disconnected) => break, + if let Some((_, waiters)) = level_clone.pending_entity_generations.remove(&pos) + { + for tx in waiters { + let _ = tx.send(arc_chunk.clone()); } } }) - .expect("Failed to spawn Entity Generation Thread"); - - tracker_lock.push(handle); + .expect("Failed to spawn entity generation thread"); } - - 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") diff --git a/pumpkin/Cargo.toml b/pumpkin/Cargo.toml index 70f07cdf3..071e35c97 100644 --- a/pumpkin/Cargo.toml +++ b/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 diff --git a/pumpkin/src/entity/player.rs b/pumpkin/src/entity/player.rs index ce9dcf656..bc15f6e27 100644 --- a/pumpkin/src/entity/player.rs +++ b/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:?}"); - self.chunk_queue.push(HeapNode(dst, pos, chunk)); + new_nodes.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,22 +260,27 @@ impl ChunkManager { && !unloading_chunks.contains(pos) }); - let mut new_queue = BinaryHeap::with_capacity(self.chunk_queue.len()); - for node in self.chunk_queue.drain() { + let mut tasks: Vec<_> = self.chunk_queue.drain().collect(); + 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) { - new_queue.push(HeapNode(dst, node.1, node.2)); - } + dst <= view_distance_i32 && !unloading_chunks.contains(&node.1) + }); + 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) { @@ -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(); diff --git a/pumpkin/src/server/mod.rs b/pumpkin/src/server/mod.rs index f98d9e64b..c808a10d3 100644 --- a/pumpkin/src/server/mod.rs +++ b/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,