From 300ccde10841443bee60a2855dddb50b072ea372 Mon Sep 17 00:00:00 2001 From: kralverde <80051564+kralverde@users.noreply.github.com> Date: Wed, 12 Mar 2025 08:40:53 -1000 Subject: [PATCH] ChunkIO api and world saving optimizations (#624) --- pumpkin-config/Cargo.toml | 4 + pumpkin-config/src/chunk.rs | 5 +- pumpkin-config/src/lib.rs | 47 +- pumpkin-world/Cargo.toml | 11 +- pumpkin-world/benches/chunk_noise_populate.rs | 22 +- pumpkin-world/src/chunk/format/anvil.rs | 789 +++++++++++++++--- pumpkin-world/src/chunk/format/linear.rs | 86 +- pumpkin-world/src/chunk/format/mod.rs | 2 + .../src/chunk/io/chunk_file_manager.rs | 125 ++- pumpkin-world/src/chunk/io/mod.rs | 29 +- pumpkin-world/src/chunk/mod.rs | 1 + .../src/generation/generic_generator.rs | 2 + .../src/generation/implementation/test.rs | 2 + pumpkin-world/src/level.rs | 37 +- pumpkin/src/entity/living.rs | 4 +- pumpkin/src/entity/player.rs | 6 +- pumpkin/src/lib.rs | 26 +- pumpkin/src/net/authentication.rs | 8 +- pumpkin/src/net/lan_broadcast.rs | 6 +- pumpkin/src/net/packet/config.rs | 4 +- pumpkin/src/net/packet/login.rs | 42 +- pumpkin/src/net/packet/play.rs | 6 +- pumpkin/src/net/query.rs | 4 +- pumpkin/src/net/rcon/mod.rs | 4 +- pumpkin/src/server/mod.rs | 6 +- pumpkin/src/world/mod.rs | 57 +- 26 files changed, 997 insertions(+), 338 deletions(-) diff --git a/pumpkin-config/Cargo.toml b/pumpkin-config/Cargo.toml index 53471a761..65fe6c6fc 100644 --- a/pumpkin-config/Cargo.toml +++ b/pumpkin-config/Cargo.toml @@ -10,3 +10,7 @@ log.workspace = true uuid.workspace = true toml = "0.8" + +[features] +# Adds helper to change the config at runtime +test_helper = [] diff --git a/pumpkin-config/src/chunk.rs b/pumpkin-config/src/chunk.rs index eb000c91e..9927c4929 100644 --- a/pumpkin-config/src/chunk.rs +++ b/pumpkin-config/src/chunk.rs @@ -2,14 +2,15 @@ use std::str; use serde::{Deserialize, Serialize}; -#[derive(Deserialize, Serialize, Default)] +#[derive(Deserialize, Serialize, Default, Clone)] #[serde(default)] pub struct ChunkConfig { pub compression: ChunkCompression, pub format: ChunkFormat, + pub write_in_place: bool, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] pub struct ChunkCompression { pub algorithm: Compression, pub level: u32, diff --git a/pumpkin-config/src/lib.rs b/pumpkin-config/src/lib.rs index 486a1f9c0..5b5bbcfc8 100644 --- a/pumpkin-config/src/lib.rs +++ b/pumpkin-config/src/lib.rs @@ -36,10 +36,46 @@ use resource_pack::ResourcePackConfig; const CONFIG_ROOT_FOLDER: &str = "config/"; -pub static ADVANCED_CONFIG: LazyLock = - LazyLock::new(AdvancedConfiguration::load); +pub static BASIC_CONFIG: LazyLock = LazyLock::new(|| { + let exec_dir = env::current_dir().unwrap(); + BasicConfiguration::load(&exec_dir) +}); -pub static BASIC_CONFIG: LazyLock = LazyLock::new(BasicConfiguration::load); +#[cfg(not(feature = "test_helper"))] +static ADVANCED_CONFIG: LazyLock = LazyLock::new(|| { + let exec_dir = env::current_dir().unwrap(); + AdvancedConfiguration::load(&exec_dir) +}); + +#[cfg(not(feature = "test_helper"))] +pub fn advanced_config() -> &'static AdvancedConfiguration { + &ADVANCED_CONFIG +} + +// This is pretty jank but it works :( +// TODO: Can we refactor this better? +#[cfg(feature = "test_helper")] +use std::cell::RefCell; + +// Yes, we are leaking memory here, but it is only for tests. Need to maintain pairity with the +// non-test code +#[cfg(feature = "test_helper")] +thread_local! { + // Needs to be thread local so we don't override the config while another test is running + static ADVANCED_CONFIG: RefCell<&'static AdvancedConfiguration> = RefCell::new(Box::leak(Box::new(AdvancedConfiguration::default()))); +} + +#[cfg(feature = "test_helper")] +pub fn override_config_for_testing(config: AdvancedConfiguration) { + ADVANCED_CONFIG.with_borrow_mut(|ref_config| { + *ref_config = Box::leak(Box::new(config)); + }); +} + +#[cfg(feature = "test_helper")] +pub fn advanced_config() -> &'static AdvancedConfiguration { + ADVANCED_CONFIG.with_borrow(|config| *config) +} /// The idea is that Pumpkin should very customizable. /// You can Enable or Disable Features depending on your needs. @@ -125,12 +161,11 @@ impl Default for BasicConfiguration { } trait LoadConfiguration { - fn load() -> Self + fn load(exec_dir: &Path) -> Self where Self: Sized + Default + Serialize + DeserializeOwned, { - let exe_dir = env::current_dir().unwrap(); - let config_dir = exe_dir.join(CONFIG_ROOT_FOLDER); + let config_dir = exec_dir.join(CONFIG_ROOT_FOLDER); if !config_dir.exists() { log::debug!("creating new config root folder"); fs::create_dir(&config_dir).expect("Failed to create Config root folder"); diff --git a/pumpkin-world/Cargo.toml b/pumpkin-world/Cargo.toml index 1cbe12027..a074f8df3 100644 --- a/pumpkin-world/Cargo.toml +++ b/pumpkin-world/Cargo.toml @@ -36,20 +36,21 @@ flate2 = "1.1" lz4 = "1.28" zstd = "0.13.3" - +itertools = "0.14.0" file-guard = "0.2" indexmap = "2.7" - enum_dispatch = "0.3" - noise = "0.9" - serde_json5 = "0.2.0" - derive-getters = "0.5.0" + [dev-dependencies] criterion = { version = "0.5", features = ["html_reports", "async_tokio"] } temp-dir = "0.1.14" +# Print log info inside tests when needed +env_logger = "0.11.7" +# Allows us to modify the config +pumpkin-config = { path = "../pumpkin-config", features = ["test_helper"] } [[bench]] name = "chunk_noise_populate" diff --git a/pumpkin-world/benches/chunk_noise_populate.rs b/pumpkin-world/benches/chunk_noise_populate.rs index 8751933ee..0aa8bf951 100644 --- a/pumpkin-world/benches/chunk_noise_populate.rs +++ b/pumpkin-world/benches/chunk_noise_populate.rs @@ -6,7 +6,7 @@ use pumpkin_world::{ GlobalProtoNoiseRouter, GlobalRandomConfig, NOISE_ROUTER_ASTS, bench_create_and_populate_noise, chunk::ChunkData, global_path, level::Level, }; -use tokio::{runtime::Runtime, sync::RwLock, task::JoinSet}; +use tokio::{runtime::Runtime, sync::RwLock}; fn bench_populate_noise(c: &mut Criterion) { let seed = 0; @@ -29,6 +29,8 @@ async fn test_reads(level: &Arc, positions: Vec>) { let _ = x; } } + +/* async fn test_reads_parallel(level: &Arc, positions: Vec>, threads: usize) { let mut tasks = JoinSet::new(); @@ -45,11 +47,13 @@ async fn test_reads_parallel(level: &Arc, positions: Vec>, t let _ = tasks.join_all().await; } +*/ async fn test_writes(level: &Arc, chunks: Vec<(Vector2, Arc>)>) { level.write_chunks(chunks).await; } +/* async fn test_writes_parallel( level: &Arc, chunks: Vec<(Vector2, Arc>)>, @@ -70,13 +74,14 @@ async fn test_writes_parallel( let _ = tasks.join_all().await; } +*/ // -16..16 == 32 chunks, 32*32 == 1024 chunks const MIN_CHUNK: i32 = -16; const MAX_CHUNK: i32 = 16; -/// How many chunks to use on parallel tests -const CHUNKS_ON_PARALLEL: usize = 32; +// How many chunks to use on parallel tests +//const CHUNKS_ON_PARALLEL: usize = 32; fn initialize_level( async_handler: &Runtime, @@ -114,6 +119,8 @@ fn initialize_level( } // Depends on config options from `./config` +/* +// This doesn't really test anything... fn bench_chunk_io_parallel(c: &mut Criterion) { // System temp dirs are in-memory, so we cant use temp_dir let root_dir = global_path!("./bench_root_tmp"); @@ -164,7 +171,9 @@ fn bench_chunk_io_parallel(c: &mut Criterion) { read_group.finish(); fs::remove_dir_all(&root_dir).unwrap(); // cleanup + } +*/ // Depends on config options from `./config` fn bench_chunk_io(c: &mut Criterion) { @@ -235,10 +244,5 @@ fn bench_chunk_io(c: &mut Criterion) { fs::remove_dir_all(&root_dir).unwrap(); // cleanup } -criterion_group!( - benches, - bench_populate_noise, - bench_chunk_io_parallel, - bench_chunk_io -); +criterion_group!(benches, bench_populate_noise, bench_chunk_io); criterion_main!(benches); diff --git a/pumpkin-world/src/chunk/format/anvil.rs b/pumpkin-world/src/chunk/format/anvil.rs index 9b911dd9f..861d596f7 100644 --- a/pumpkin-world/src/chunk/format/anvil.rs +++ b/pumpkin-world/src/chunk/format/anvil.rs @@ -2,28 +2,26 @@ use async_trait::async_trait; use bytes::*; use flate2::read::{GzDecoder, GzEncoder, ZlibDecoder, ZlibEncoder}; use indexmap::IndexMap; -use pumpkin_config::ADVANCED_CONFIG; +use itertools::Itertools; +use pumpkin_config::advanced_config; use pumpkin_data::{block::Block, chunk::ChunkStatus}; use pumpkin_nbt::serializer::to_bytes; use pumpkin_util::math::ceil_log2; use pumpkin_util::math::vector2::Vector2; use std::{ collections::{HashMap, HashSet}, - io::{Read, Write}, - sync::Arc, + io::{Read, SeekFrom, Write}, + path::{Path, PathBuf}, time::{SystemTime, UNIX_EPOCH}, }; use tokio::{ - io::{AsyncWrite, AsyncWriteExt}, - sync::RwLock, + io::{AsyncSeekExt, AsyncWrite, AsyncWriteExt, BufWriter}, + sync::Mutex, }; -use crate::{ - chunk::{ - ChunkData, ChunkReadingError, ChunkSerializingError, ChunkWritingError, CompressionError, - io::{ChunkSerializer, LoadedData}, - }, - level::SyncChunk, +use crate::chunk::{ + ChunkData, ChunkReadingError, ChunkSerializingError, ChunkWritingError, CompressionError, + io::{ChunkSerializer, LoadedData}, }; use super::{ChunkNbt, ChunkSection, ChunkSectionBlockStates, PaletteEntry}; @@ -84,10 +82,42 @@ pub struct AnvilChunkData { compressed_data: Bytes, } +enum WriteAction { + // Don't write anything + Pass, + // Write the entire file + All, + // Only write certain indices + Parts(HashSet), +} + +impl WriteAction { + /// If we are currently not writing, sets to new Parts enum, + /// If we have parts enum, add to it, + /// If we have All enum, do nothing + fn maybe_update_chunk_index(&mut self, index: usize) { + match self { + Self::Pass => *self = Self::Parts(HashSet::from_iter([index])), + Self::Parts(parts) => { + let _ = parts.insert(index); + } + Self::All => {} + } + } +} + +struct AnvilChunkMetadata { + serialized_data: AnvilChunkData, + timestamp: u32, + + // NOTE: This is only valid if our WriteAction is `Parts` + file_sector_offset: u32, +} + pub struct AnvilChunkFile { - timestamp_table: [u32; CHUNK_COUNT], - // TODO: Only save mutated chunks (chunks that are unchanged do not need to be re-written) - chunks_data: [Option; CHUNK_COUNT], + chunks_data: [Option; CHUNK_COUNT], + end_sector: u32, + write_action: Mutex, } impl Compression { @@ -214,11 +244,16 @@ impl AnvilChunkData { /// Size of serialized chunk with padding #[inline] fn padded_size(&self) -> usize { - let total_size = self.raw_write_size(); - let sector_count = total_size.div_ceil(SECTOR_BYTES); + let sector_count = self.sector_count() as usize; sector_count * SECTOR_BYTES } + #[inline] + fn sector_count(&self) -> u32 { + let total_size = self.raw_write_size(); + total_size.div_ceil(SECTOR_BYTES) as u32 + } + fn from_bytes(bytes: Bytes) -> Result { let mut bytes = bytes; // Minus one for the compression byte @@ -269,14 +304,19 @@ impl AnvilChunkData { Ok(chunk) } - fn from_chunk(chunk: &ChunkData) -> Result { + fn from_chunk( + chunk: &ChunkData, + compression: Option, + ) -> Result { let raw_bytes = chunk_to_bytes(chunk) .map_err(|err| ChunkWritingError::ChunkSerializingError(err.to_string()))?; - let compression: Compression = ADVANCED_CONFIG.chunk.compression.algorithm.clone().into(); + let compression = compression + .unwrap_or_else(|| advanced_config().chunk.compression.algorithm.clone().into()); + // We need to buffer here anyway so theres no use in making an impl Write for this let compressed_data = compression - .compress_data(&raw_bytes, ADVANCED_CONFIG.chunk.compression.level) + .compress_data(&raw_bytes, advanced_config().chunk.compression.level) .map_err(ChunkWritingError::Compression)?; Ok(AnvilChunkData { @@ -298,36 +338,126 @@ impl AnvilChunkFile { let index = (local_z << SUBREGION_BITS) + local_x; index as usize } -} -impl Default for AnvilChunkFile { - fn default() -> Self { - Self { - timestamp_table: [0; CHUNK_COUNT], - chunks_data: [const { None }; CHUNK_COUNT], + async fn write_indices(&self, path: &Path, indices: &[usize]) -> Result<(), std::io::Error> { + log::trace!("Writing in place: {:?}", path); + + let file = tokio::fs::OpenOptions::new() + .read(false) + .write(true) + .create(true) + .truncate(false) + .append(false) + .open(path) + .await?; + + let mut write = BufWriter::new(file); + // The first two sectors are reserved for the location table + for (index, metadata) in self.chunks_data.iter().enumerate() { + if let Some(chunk) = metadata { + let chunk_data = &chunk.serialized_data; + let sector_count = chunk_data.sector_count(); + log::trace!( + "Writing position for chunk {} - {}:{}", + index, + chunk.file_sector_offset, + sector_count + ); + write + .write_u32((chunk.file_sector_offset << 8) | sector_count) + .await?; + } else { + // If the chunk is not present, we write 0 to the location and timestamp tables + write.write_u32(0).await?; + }; } - } -} -#[async_trait] -impl ChunkSerializer for AnvilChunkFile { - type Data = SyncChunk; + for metadata in &self.chunks_data { + if let Some(chunk) = metadata { + write.write_u32(chunk.timestamp).await?; + } else { + // If the chunk is not present, we write 0 to the location and timestamp tables + write.write_u32(0).await?; + } + } - fn get_chunk_key(chunk: &Vector2) -> String { - let (region_x, region_z) = Self::get_region_coords(chunk); - format!("./r.{}.{}.mca", region_x, region_z) + let mut chunks = indices + .iter() + .map(|index| { + ( + index, + self.chunks_data[*index] + .as_ref() + .expect("We are trying to write a chunk, but it does not exist!"), + ) + }) + .collect::>(); + + // Sort such that writes are in order + chunks.sort_by_key(|chunk| chunk.1.file_sector_offset); + + #[cfg(debug_assertions)] + { + // Verify we are actually two sectors into the file + let current_pos = write.stream_position().await?; + assert!(current_pos as usize == 2 * SECTOR_BYTES); + } + + let mut current_sector = 2; + for (index, chunk) in chunks { + debug_assert!( + current_sector <= chunk.file_sector_offset, + "Current sector is {} but we want to write to {}!", + current_sector, + chunk.file_sector_offset + ); + + // Seek only if we need to + if chunk.file_sector_offset != current_sector { + log::trace!("Seeking to sector {}", chunk.file_sector_offset); + let _ = write + .seek(SeekFrom::Start( + chunk.file_sector_offset as u64 * SECTOR_BYTES as u64, + )) + .await?; + current_sector = chunk.file_sector_offset; + } + log::trace!( + "Writing chunk {} - {}:{}", + index, + current_sector, + chunk.serialized_data.sector_count() + ); + + current_sector += chunk.serialized_data.sector_count(); + + chunk.serialized_data.write(&mut write).await?; + } + + write.flush().await } - async fn write( - &self, - write: &mut (impl AsyncWrite + Unpin + Send), - ) -> Result<(), std::io::Error> { + /// Write entire file, disregarding saved offsets + async fn write_all(&self, path: &Path) -> Result<(), std::io::Error> { + let temp_path = path.with_extension("tmp"); + log::trace!("Writing tmp file to disk: {:?}", temp_path); + + let file = tokio::fs::OpenOptions::new() + .read(false) + .write(true) + .create(true) + .truncate(true) + .open(&temp_path) + .await?; + + let mut write = BufWriter::new(file); + // The first two sectors are reserved for the location table let mut current_sector: u32 = 2; - for i in 0..CHUNK_COUNT { - if let Some(chunk) = &self.chunks_data[i] { - let chunk_bytes = chunk.padded_size(); - let sector_count = (chunk_bytes / SECTOR_BYTES) as u32; + for metadata in &self.chunks_data { + if let Some(chunk) = metadata { + let chunk = &chunk.serialized_data; + let sector_count = chunk.sector_count(); write .write_u32((current_sector << 8) | sector_count) .await?; @@ -338,14 +468,73 @@ impl ChunkSerializer for AnvilChunkFile { }; } - for timestamp in self.timestamp_table { - write.write_u32(timestamp).await?; + for metadata in &self.chunks_data { + if let Some(chunk) = metadata { + write.write_u32(chunk.timestamp).await?; + } else { + // If the chunk is not present, we write 0 to the location and timestamp tables + write.write_u32(0).await?; + } } for chunk in self.chunks_data.iter().flatten() { - chunk.write(write).await?; + chunk.serialized_data.write(&mut write).await?; } + write.flush().await?; + // The rename of the file works like an atomic operation ensuring + // that the data is not corrupted before the rename is completed + tokio::fs::rename(temp_path, path).await?; + + log::trace!("Wrote file to Disk: {:?}", path); + Ok(()) + } +} + +impl Default for AnvilChunkFile { + fn default() -> Self { + Self { + chunks_data: [const { None }; CHUNK_COUNT], + write_action: Mutex::new(WriteAction::Pass), + // Two sectors for offset + timestamp + end_sector: 2, + } + } +} + +#[async_trait] +impl ChunkSerializer for AnvilChunkFile { + type Data = ChunkData; + type WriteBackend = PathBuf; + + fn should_write(&self, is_watched: bool) -> bool { + !is_watched + } + + fn get_chunk_key(chunk: &Vector2) -> String { + let (region_x, region_z) = Self::get_region_coords(chunk); + format!("./r.{}.{}.mca", region_x, region_z) + } + + async fn write(&self, path: PathBuf) -> Result<(), std::io::Error> { + let mut write_action = self.write_action.lock().await; + match &*write_action { + WriteAction::Pass => { + log::debug!( + "Skipping write for {:?} as there were no dirty chunks", + path + ); + Ok(()) + } + WriteAction::All => self.write_all(&path).await, + WriteAction::Parts(parts) => { + self.write_indices(&path, Vec::from_iter(parts.iter().cloned()).as_slice()) + .await + } + }?; + + // If we still are in memory after this, we don't need to write again! + *write_action = WriteAction::Pass; Ok(()) } @@ -361,42 +550,214 @@ impl ChunkSerializer for AnvilChunkFile { let mut chunk_file = AnvilChunkFile::default(); + let mut last_offset = 2; for i in 0..CHUNK_COUNT { - chunk_file.timestamp_table[i] = timestamp_bytes.get_u32(); + let timestamp = timestamp_bytes.get_u32(); let location = location_bytes.get_u32(); let sector_count = (location & 0xFF) as usize; let sector_offset = (location >> 8) as usize; + let end_offset = sector_offset + sector_count; // If the sector offset or count is 0, the chunk is not present (we should not parse empty chunks) if sector_offset == 0 || sector_count == 0 { continue; } + if end_offset > last_offset { + last_offset = end_offset; + } + // We always subtract 2 for the first two sectors for the timestamp and location tables // that we walked earlier let bytes_offset = (sector_offset - 2) * SECTOR_BYTES; let bytes_count = sector_count * SECTOR_BYTES; - chunk_file.chunks_data[i] = Some(AnvilChunkData::from_bytes( + let serialized_data = AnvilChunkData::from_bytes( raw_file_bytes.slice(bytes_offset..bytes_offset + bytes_count), - )?); + )?; + + chunk_file.chunks_data[i] = Some(AnvilChunkMetadata { + serialized_data, + timestamp, + file_sector_offset: sector_offset as u32, + }); } + chunk_file.end_sector = last_offset as u32; Ok(chunk_file) } - async fn update_chunks(&mut self, chunks_data: &[Self::Data]) -> Result<(), ChunkWritingError> { + async fn update_chunk(&mut self, chunk: &ChunkData) -> Result<(), ChunkWritingError> { let epoch = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_secs() as u32; - for chunk in chunks_data { - let chunk = chunk.read().await; - let index = AnvilChunkFile::get_chunk_index(&chunk.position); - self.chunks_data[index] = Some(AnvilChunkData::from_chunk(&chunk)?); - self.timestamp_table[index] = epoch; + let index = AnvilChunkFile::get_chunk_index(&chunk.position); + // Default to the compression type read from the file + let compression_type = self.chunks_data[index] + .as_ref() + .and_then(|chunk_data| chunk_data.serialized_data.compression); + let new_chunk_data = AnvilChunkData::from_chunk(chunk, compression_type)?; + + let mut write_action = self.write_action.lock().await; + if !advanced_config().chunk.write_in_place { + *write_action = WriteAction::All; + } + + match &*write_action { + WriteAction::All => { + log::trace!("Write action is all: setting chunk in place"); + // Doesn't matter, just add the data + self.chunks_data[index] = Some(AnvilChunkMetadata { + serialized_data: new_chunk_data, + timestamp: epoch, + file_sector_offset: 0, + }); + } + _ => { + match self.chunks_data[index].as_ref() { + None => { + log::trace!( + "Chunk {} does not exist, appending to EOF: {}:{}", + index, + self.end_sector, + new_chunk_data.sector_count() + ); + // This chunk didn't exist before; append to EOF + let new_eof = self.end_sector + new_chunk_data.sector_count(); + self.chunks_data[index] = Some(AnvilChunkMetadata { + serialized_data: new_chunk_data, + timestamp: epoch, + file_sector_offset: self.end_sector, + }); + self.end_sector = new_eof; + write_action.maybe_update_chunk_index(index); + } + Some(old_chunk) => { + if old_chunk.serialized_data.sector_count() == new_chunk_data.sector_count() + { + log::trace!( + "Chunk {} exists, writing in place: {}:{}", + index, + old_chunk.file_sector_offset, + new_chunk_data.sector_count() + ); + // We can just add it + self.chunks_data[index] = Some(AnvilChunkMetadata { + serialized_data: new_chunk_data, + timestamp: epoch, + file_sector_offset: old_chunk.file_sector_offset, + }); + write_action.maybe_update_chunk_index(index); + } else { + // Walk back the end of the list; seeing if theres something that can fit + // in our spot. Here we play a game between is it worth it to do all + // this swapping. I figure if we don't find it after 64 chunks, just + // re-write the whole file instead + // The number is a guestimation and no rigorious thought when into it. + // The more we leapfrog like this, there is a higher + // (abiet still small) of these chunks being corrupted if we are doing a + // write operation when there is an un-clean shutdown + // + // Writing all is "safer" in the sense that no chunks will corrupt, + // but will still roll back the entire region if + // there is an unclean shutdown + + let mut chunks = self + .chunks_data + .iter() + .enumerate() + .filter_map(|(index, chunk)| { + chunk.as_ref().map(|chunk| (index, chunk)) + }) + .collect::>(); + chunks.sort_by_key(|chunk| chunk.1.file_sector_offset); + + let mut chunks_to_shift = chunks + .into_iter() + .rev() + .take(64) + .take_while_inclusive(|chunk| { + chunk.1.serialized_data.sector_count() + != old_chunk.serialized_data.sector_count() + }) + .collect::>(); + + if chunks_to_shift.last().is_none_or(|chunk| chunk.0 == index) { + log::trace!( + "Unable to find a chunk to swap with; falling back to serialize all", + ); + + // give up... + *write_action = WriteAction::All; + self.chunks_data[index] = Some(AnvilChunkMetadata { + serialized_data: new_chunk_data, + timestamp: epoch, + file_sector_offset: 0, + }); + } else { + // swap last element of the chunks to shift (the first because we + // reversed it) and shift the rest down + let swap = chunks_to_shift + .pop() + .expect("We just checked that this exists"); + + let indices_to_shift = chunks_to_shift + .iter() + .map(|(index, _)| index) + .copied() + .collect::>(); + let swapped_sectors = swap.1.serialized_data.sector_count(); + let new_sectors = new_chunk_data.sector_count(); + let swapped_index = swap.0; + let old_offset = old_chunk.file_sector_offset; + self.chunks_data[index] = Some(AnvilChunkMetadata { + serialized_data: new_chunk_data, + timestamp: epoch, + file_sector_offset: swap.1.file_sector_offset, + }); + write_action.maybe_update_chunk_index(index); + + self.chunks_data[swapped_index] + .as_mut() + .expect("We checked if this was none") + .file_sector_offset = old_offset; + write_action.maybe_update_chunk_index(swapped_index); + + // Then offset everything else + + // If positive, now larger -> shift right, else shift left + let offset = new_sectors as i64 - swapped_sectors as i64; + + log::trace!( + "Swapping {} with {}, shifting all chunks {} and after by {}", + index, + swapped_index, + swapped_index, + offset + ); + + for shift_index in indices_to_shift { + let chunk_data = self.chunks_data[shift_index] + .as_mut() + .expect("We checked if this was none"); + let new_offset = chunk_data.file_sector_offset as i64 + offset; + chunk_data.file_sector_offset = new_offset as u32; + write_action.maybe_update_chunk_index(shift_index); + } + + // If the shift is negative then there will be trailing data, but i + // think thats fine + + let new_end = self.end_sector as i64 + offset; + self.end_sector = new_end as u32; + } + } + } + } + } } Ok(()) @@ -405,7 +766,7 @@ impl ChunkSerializer for AnvilChunkFile { async fn get_chunks( &self, chunks: &[Vector2], - stream: tokio::sync::mpsc::Sender>, + stream: tokio::sync::mpsc::Sender>, ) { // Create an unbounded buffer so we don't block the rayon thread pool let (bridge_send, mut bridge_recv) = tokio::sync::mpsc::unbounded_channel(); @@ -414,22 +775,25 @@ impl ChunkSerializer for AnvilChunkFile { // runtime for chunk in chunks.iter().cloned() { let index = AnvilChunkFile::get_chunk_index(&chunk); - let anvil_chunk = self.chunks_data[index].clone(); + match &self.chunks_data[index] { + None => stream + .send(LoadedData::Missing(chunk)) + .await + .expect("Failed to send chunk"), + Some(chunk_metadata) => { + let send = bridge_send.clone(); + let chunk_data = chunk_metadata.serialized_data.clone(); + rayon::spawn(move || { + let result = match chunk_data.to_chunk(chunk) { + Ok(chunk) => LoadedData::Loaded(chunk), + Err(err) => LoadedData::Error((chunk, err)), + }; - let send = bridge_send.clone(); - rayon::spawn(move || { - let result = if let Some(data) = anvil_chunk { - match data.to_chunk(chunk) { - Ok(chunk) => LoadedData::Loaded(Arc::new(RwLock::new(chunk))), - Err(err) => LoadedData::Error((chunk, err)), - } - } else { - LoadedData::Missing(chunk) - }; - - send.send(result) - .expect("Failed to send anvil chunks from rayon thread"); - }); + send.send(result) + .expect("Failed to send anvil chunks from rayon thread"); + }); + } + } } // Drop the original so streams clean-up drop(bridge_send); @@ -545,6 +909,7 @@ pub fn chunk_to_bytes(chunk_data: &ChunkData) -> Result, ChunkSerializin #[cfg(test)] mod tests { + use pumpkin_config::{AdvancedConfiguration, advanced_config, override_config_for_testing}; use pumpkin_util::math::vector2::Vector2; use std::fs; use std::path::PathBuf; @@ -555,8 +920,41 @@ mod tests { use crate::chunk::format::anvil::AnvilChunkFile; use crate::chunk::io::chunk_file_manager::ChunkFileManager; use crate::chunk::io::{ChunkIO, LoadedData}; + use crate::coordinates::ChunkRelativeBlockCoordinates; use crate::generation::{Seed, get_world_gen}; - use crate::level::LevelFolder; + use crate::level::{LevelFolder, SyncChunk}; + + async fn get_chunks( + saver: &ChunkFileManager, + folder: &LevelFolder, + chunks: &[(Vector2, SyncChunk)], + ) -> Box<[SyncChunk]> { + let mut read_chunks = Vec::new(); + let (send, mut recv) = tokio::sync::mpsc::channel(1); + + let chunk_pos = chunks.iter().map(|(at, _)| *at).collect::>(); + let spawn = saver.fetch_chunks(folder, &chunk_pos, send); + let collect = async { + while let Some(data) = recv.recv().await { + read_chunks.push(data); + } + }; + + tokio::join!(spawn, collect); + + let read_chunks = read_chunks + .into_iter() + .map(|chunk| match chunk { + LoadedData::Loaded(chunk) => chunk, + LoadedData::Missing(_) => panic!("Missing chunk"), + LoadedData::Error((position, error)) => { + panic!("Error reading chunk at {:?} | Error: {:?}", position, error) + } + }) + .collect::>(); + + read_chunks.into_boxed_slice() + } #[tokio::test(flavor = "multi_thread")] async fn not_existing() { @@ -585,7 +983,14 @@ mod tests { } #[tokio::test(flavor = "multi_thread")] - async fn test_writing() { + async fn test_write_in_place() { + let mut config = AdvancedConfiguration::default(); + config.chunk.write_in_place = true; + override_config_for_testing(config); + assert!(advanced_config().chunk.write_in_place); + + let _ = env_logger::try_init(); + let generator = get_world_gen(Seed(0)); let temp_dir = TempDir::new().unwrap(); @@ -606,36 +1011,226 @@ mod tests { } } - for i in 0..5 { - println!("Iteration {}", i + 1); + // TEST APPEND TO END + + chunk_saver + .save_chunks(&level_folder, chunks.clone()) + .await + .expect("Failed to write chunk"); + + // Create a new manager to ensure nothing is cached + let chunk_saver = ChunkFileManager::::default(); + let read_chunks = get_chunks(&chunk_saver, &level_folder, &chunks).await; + + for (_, chunk) in &chunks { + let chunk = chunk.read().await; + for read_chunk in read_chunks.iter() { + let read_chunk = read_chunk.read().await; + if read_chunk.position == chunk.position { + assert_eq!(chunk.subchunks, read_chunk.subchunks, "Chunks don't match"); + break; + } + } + } + + // TEST WRITE IN PLACE + + // Idk what blocks these are, they just have to be different + let mut chunk = chunks.first().unwrap().1.write().await; + chunk.subchunks.set_block( + ChunkRelativeBlockCoordinates { + x: 0u32.into(), + y: 0.into(), + z: 0u32.into(), + }, + 1000, + ); + // Mark dirty so we actually write it + chunk.dirty = true; + drop(chunk); + let mut chunk = chunks.last().unwrap().1.write().await; + chunk.subchunks.set_block( + ChunkRelativeBlockCoordinates { + x: 0u32.into(), + y: 0.into(), + z: 0u32.into(), + }, + 1000, + ); + // Mark dirty so we actually write it + chunk.dirty = true; + drop(chunk); + + chunk_saver + .save_chunks(&level_folder, chunks.clone()) + .await + .expect("Failed to write chunk"); + + // Create a new manager to ensure nothing is cached + let chunk_saver = ChunkFileManager::::default(); + let read_chunks = get_chunks(&chunk_saver, &level_folder, &chunks).await; + + for (_, chunk) in &chunks { + let chunk = chunk.read().await; + for read_chunk in read_chunks.iter() { + let read_chunk = read_chunk.read().await; + if read_chunk.position == chunk.position { + assert_eq!(chunk.subchunks, read_chunk.subchunks, "Chunks don't match"); + break; + } + } + } + + // TEST SWAP SHIFT + + // Make a big chunk + let mut chunk = chunks.first().unwrap().1.write().await; + for x in 0..16 { + for z in 0..16 { + for y in 0..4 { + let block_id = 16 * 16 * y + 16 * z + x; + chunk.subchunks.set_block( + ChunkRelativeBlockCoordinates { + x: x.into(), + y: (y as i32).into(), + z: z.into(), + }, + block_id, + ); + } + } + } + // Mark dirty so we actually write it + chunk.dirty = true; + drop(chunk); + let mut chunk = chunks[2].1.write().await; + for x in 0..16 { + for z in 0..16 { + for y in 0..4 { + let block_id = 16 * 16 * y + 16 * z + x; + chunk.subchunks.set_block( + ChunkRelativeBlockCoordinates { + x: x.into(), + y: (y as i32).into(), + z: z.into(), + }, + block_id, + ); + } + } + } + // Mark dirty so we actually write it + chunk.dirty = true; + drop(chunk); + + chunk_saver + .save_chunks(&level_folder, chunks.clone()) + .await + .expect("Failed to write chunk"); + + // Create a new manager to ensure nothing is cached + let chunk_saver = ChunkFileManager::::default(); + let read_chunks = get_chunks(&chunk_saver, &level_folder, &chunks).await; + + for (_, chunk) in &chunks { + let chunk = chunk.read().await; + for read_chunk in read_chunks.iter() { + let read_chunk = read_chunk.read().await; + if read_chunk.position == chunk.position { + assert_eq!(chunk.subchunks, read_chunk.subchunks, "Chunks don't match"); + break; + } + } + } + + // TEST DEFAULT TO WRITE ALL + + // Make an even bigger chunk + let mut chunk = chunks.last().unwrap().1.write().await; + for x in 0..16 { + for z in 0..16 { + for y in 0..16 { + let block_id = 16 * 16 * y + 16 * z + x; + chunk.subchunks.set_block( + ChunkRelativeBlockCoordinates { + x: x.into(), + y: (y as i32).into(), + z: z.into(), + }, + block_id, + ); + } + } + } + // Mark dirty so we actually write it + chunk.dirty = true; + drop(chunk); + + chunk_saver + .save_chunks(&level_folder, chunks.clone()) + .await + .expect("Failed to write chunk"); + + // Create a new manager to ensure nothing is cached + let chunk_saver = ChunkFileManager::::default(); + let read_chunks = get_chunks(&chunk_saver, &level_folder, &chunks).await; + + for (_, chunk) in &chunks { + let chunk = chunk.read().await; + for read_chunk in read_chunks.iter() { + let read_chunk = read_chunk.read().await; + if read_chunk.position == chunk.position { + assert_eq!(chunk.subchunks, read_chunk.subchunks, "Chunks don't match"); + break; + } + } + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_write_bulk() { + let mut config = AdvancedConfiguration::default(); + config.chunk.write_in_place = false; + override_config_for_testing(config); + assert!(!advanced_config().chunk.write_in_place); + + let _ = env_logger::try_init(); + + let generator = get_world_gen(Seed(0)); + + let temp_dir = TempDir::new().unwrap(); + let level_folder = LevelFolder { + root_folder: temp_dir.path().to_path_buf(), + region_folder: temp_dir.path().join("region"), + }; + fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder"); + let chunk_saver = ChunkFileManager::::default(); + + // Generate chunks + let mut chunks = vec![]; + for x in -5..5 { + for y in -5..5 { + let position = Vector2::new(x, y); + let chunk = generator.generate_chunk(position); + chunks.push((position, Arc::new(RwLock::new(chunk)))); + } + } + + for _ in 0..5 { + // Mark the chunks as dirty so we save them again + for (_, chunk) in &chunks { + let mut chunk = chunk.write().await; + chunk.dirty = true; + } + chunk_saver .save_chunks(&level_folder, chunks.clone()) .await .expect("Failed to write chunk"); - let mut read_chunks = Vec::new(); - let (send, mut recv) = tokio::sync::mpsc::channel(1); - - let chunk_pos = chunks.iter().map(|(at, _)| *at).collect::>(); - let spawn = chunk_saver.fetch_chunks(&level_folder, &chunk_pos, send); - let collect = async { - while let Some(data) = recv.recv().await { - read_chunks.push(data); - } - }; - - tokio::join!(spawn, collect); - - let read_chunks = read_chunks - .into_iter() - .map(|chunk| match chunk { - LoadedData::Loaded(chunk) => chunk, - LoadedData::Missing(_) => panic!("Missing chunk"), - LoadedData::Error((position, error)) => { - panic!("Error reading chunk at {:?} | Error: {:?}", position, error) - } - }) - .collect::>(); + // Create a new manager to ensure nothing is cached + let chunk_saver = ChunkFileManager::::default(); + let read_chunks = get_chunks(&chunk_saver, &level_folder, &chunks).await; for (_, chunk) in &chunks { let chunk = chunk.read().await; @@ -648,8 +1243,6 @@ mod tests { } } } - - println!("Checked chunks successfully"); } // TODO diff --git a/pumpkin-world/src/chunk/format/linear.rs b/pumpkin-world/src/chunk/format/linear.rs index 6f483bc33..61866314b 100644 --- a/pumpkin-world/src/chunk/format/linear.rs +++ b/pumpkin-world/src/chunk/format/linear.rs @@ -1,18 +1,16 @@ use std::io::ErrorKind; -use std::sync::Arc; +use std::path::PathBuf; use std::time::{SystemTime, UNIX_EPOCH}; use crate::chunk::format::anvil::AnvilChunkFile; use crate::chunk::io::{ChunkSerializer, LoadedData}; use crate::chunk::{ChunkData, ChunkReadingError, ChunkWritingError}; -use crate::level::SyncChunk; use async_trait::async_trait; use bytes::{Buf, BufMut, Bytes}; use log::error; -use pumpkin_config::ADVANCED_CONFIG; +use pumpkin_config::advanced_config; use pumpkin_util::math::vector2::Vector2; -use tokio::io::{AsyncWrite, AsyncWriteExt}; -use tokio::sync::RwLock; +use tokio::io::{AsyncWriteExt, BufWriter}; use super::anvil::{CHUNK_COUNT, chunk_to_bytes}; @@ -162,17 +160,32 @@ impl Default for LinearFile { #[async_trait] impl ChunkSerializer for LinearFile { - type Data = SyncChunk; + type Data = ChunkData; + type WriteBackend = PathBuf; + + fn should_write(&self, is_watched: bool) -> bool { + !is_watched + } fn get_chunk_key(chunk: &Vector2) -> String { let (region_x, region_z) = AnvilChunkFile::get_region_coords(chunk); format!("./r.{}.{}.linear", region_x, region_z) } - async fn write( - &self, - write: &mut (impl AsyncWrite + Unpin + Send), - ) -> Result<(), std::io::Error> { + async fn write(&self, path: PathBuf) -> Result<(), std::io::Error> { + let temp_path = path.with_extension("tmp"); + log::trace!("Writing tmp file to disk: {:?}", temp_path); + + let file = tokio::fs::OpenOptions::new() + .read(false) + .write(true) + .create(true) + .truncate(true) + .open(&temp_path) + .await?; + + let mut write = BufWriter::new(file); + // Parse the headers to a buffer let mut data_buffer: Vec = self .chunks_headers @@ -187,14 +200,14 @@ impl ChunkSerializer for LinearFile { // TODO: maybe zstd lib has memory leaks let compressed_buffer = zstd::bulk::compress( data_buffer.as_slice(), - ADVANCED_CONFIG.chunk.compression.level as i32, + advanced_config().chunk.compression.level as i32, ) .expect("Failed to compress the data buffer") .into_boxed_slice(); let file_header = LinearFileHeader { chunks_bytes: compressed_buffer.len(), - compression_level: ADVANCED_CONFIG.chunk.compression.level as u8, + compression_level: advanced_config().chunk.compression.level as u8, chunks_count: self .chunks_headers .iter() @@ -216,6 +229,13 @@ impl ChunkSerializer for LinearFile { write.write_all(&compressed_buffer).await?; write.write_all(&SIGNATURE).await?; + write.flush().await?; + + // The rename of the file works like an atomic operation ensuring + // that the data is not corrupted before the rename is completed + tokio::fs::rename(temp_path, &path).await?; + + log::trace!("Wrote file to Disk: {:?}", path); Ok(()) } @@ -286,25 +306,21 @@ impl ChunkSerializer for LinearFile { }) } - async fn update_chunks(&mut self, chunks_data: &[Self::Data]) -> Result<(), ChunkWritingError> { - for chunk_data in chunks_data { - let chunk_data = chunk_data.read().await; - let index = LinearFile::get_chunk_index(&chunk_data.position); - let chunk_raw: Bytes = chunk_to_bytes(&chunk_data) - .map_err(|err| ChunkWritingError::ChunkSerializingError(err.to_string()))? - .into(); - drop(chunk_data); + async fn update_chunk(&mut self, chunk: &ChunkData) -> Result<(), ChunkWritingError> { + let index = LinearFile::get_chunk_index(&chunk.position); + let chunk_raw: Bytes = chunk_to_bytes(chunk) + .map_err(|err| ChunkWritingError::ChunkSerializingError(err.to_string()))? + .into(); - let header = &mut self.chunks_headers[index]; - header.size = chunk_raw.len() as u32; - header.timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs() as u32; + let header = &mut self.chunks_headers[index]; + header.size = chunk_raw.len() as u32; + header.timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as u32; - // We update the data buffer - self.chunks_data[index] = Some(chunk_raw); - } + // We update the data buffer + self.chunks_data[index] = Some(chunk_raw); Ok(()) } @@ -312,7 +328,7 @@ impl ChunkSerializer for LinearFile { async fn get_chunks( &self, chunks: &[Vector2], - stream: tokio::sync::mpsc::Sender>, + stream: tokio::sync::mpsc::Sender>, ) { // Create an unbounded buffer so we don't block the rayon thread pool let (bridge_send, mut bridge_recv) = tokio::sync::mpsc::unbounded_channel(); @@ -329,7 +345,7 @@ impl ChunkSerializer for LinearFile { match ChunkData::from_bytes(&data, chunk) .map_err(ChunkReadingError::ParsingError) { - Ok(chunk) => LoadedData::Loaded(Arc::new(RwLock::new(chunk))), + Ok(chunk) => LoadedData::Loaded(chunk), Err(err) => LoadedData::Error((chunk, err)), } } else { @@ -399,6 +415,8 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn test_writing() { + let _ = env_logger::try_init(); + let generator = get_world_gen(Seed(0)); let temp_dir = TempDir::new().unwrap(); @@ -421,6 +439,12 @@ mod tests { for i in 0..5 { println!("Iteration {}", i + 1); + // Mark the chunks as dirty so we save them again + for (_, chunk) in &chunks { + let mut chunk = chunk.write().await; + chunk.dirty = true; + } + chunk_saver .save_chunks( &level_folder, diff --git a/pumpkin-world/src/chunk/format/mod.rs b/pumpkin-world/src/chunk/format/mod.rs index fed88dd7c..b089637e5 100644 --- a/pumpkin-world/src/chunk/format/mod.rs +++ b/pumpkin-world/src/chunk/format/mod.rs @@ -117,6 +117,8 @@ impl ChunkData { subchunks, heightmap: chunk_data.heightmaps, position, + // This chunk is read from disk, so it has not been modified + dirty: false, }) } } diff --git a/pumpkin-world/src/chunk/io/chunk_file_manager.rs b/pumpkin-world/src/chunk/io/chunk_file_manager.rs index d3340b330..bbc5d3759 100644 --- a/pumpkin-world/src/chunk/io/chunk_file_manager.rs +++ b/pumpkin-world/src/chunk/io/chunk_file_manager.rs @@ -1,7 +1,7 @@ use std::{ collections::BTreeMap, io::ErrorKind, - ops::{AddAssign, Deref, SubAssign}, + ops::{AddAssign, SubAssign}, path::{Path, PathBuf}, sync::Arc, }; @@ -12,13 +12,14 @@ use log::{error, trace}; use num_traits::Zero; use pumpkin_util::math::vector2::Vector2; use tokio::{ - io::{AsyncReadExt, AsyncWriteExt, BufWriter}, - sync::{OnceCell, RwLock}, + io::AsyncReadExt, + join, + sync::{OnceCell, RwLock, mpsc}, }; use crate::{ - chunk::{ChunkReadingError, ChunkWritingError}, - level::LevelFolder, + chunk::{ChunkData, ChunkReadingError, ChunkWritingError}, + level::{LevelFolder, SyncChunk}, }; use super::{ChunkIO, ChunkSerializer, LoadedData}; @@ -29,7 +30,7 @@ use super::{ChunkIO, ChunkSerializer, LoadedData}; /// /// It also avoid IO operations that could produce dataraces thanks to the /// custom *DashMap* like implementation. -pub struct ChunkFileManager { +pub struct ChunkFileManager> { // Dashmap has rw-locks on shards, but we want per-serializer file_locks: RwLock>>, watchers: RwLock>, @@ -37,7 +38,7 @@ pub struct ChunkFileManager { //to avoid clippy warnings we extract the type alias type SerializerCacheEntry = OnceCell>>; -impl Default for ChunkFileManager { +impl> Default for ChunkFileManager { fn default() -> Self { Self { file_locks: RwLock::new(BTreeMap::new()), @@ -46,12 +47,12 @@ impl Default for ChunkFileManager { } } -impl ChunkFileManager { +impl> ChunkFileManager { fn map_key(folder: &LevelFolder, file_name: &str) -> PathBuf { folder.region_folder.join(file_name) } - pub async fn read_file(&self, path: &Path) -> Result>, ChunkReadingError> { + async fn read_file(&self, path: &Path) -> Result>, ChunkReadingError> { // We get the entry from the DashMap and try to insert a new lock if it doesn't exist // using dead-lock safe methods like `or_try_insert_with` @@ -116,49 +117,15 @@ impl ChunkFileManager { Ok(serializer) } - - pub async fn write_file(path: &Path, serializer: &S) -> Result<(), ChunkWritingError> { - // We use tmp files to avoid corruption of the data if the process is abruptly interrupted. - let tmp_path = &path.with_extension("tmp"); - trace!("Writing tmp file to disk: {:?}", tmp_path); - - let file = tokio::fs::OpenOptions::new() - .read(false) - .write(true) - .create(true) - .truncate(true) - .open(tmp_path) - .await - .map_err(|err| ChunkWritingError::IoError(err.kind()))?; - let mut buf_writer = BufWriter::new(file); - - serializer - .write(&mut buf_writer) - .await - .map_err(|err| ChunkWritingError::IoError(err.kind()))?; - - buf_writer - .flush() - .await - .map_err(|err| ChunkWritingError::IoError(err.kind()))?; - - // The rename of the file works like an atomic operation ensuring - // that the data is not corrupted before the rename is completed - tokio::fs::rename(tmp_path, path) - .await - .map_err(|err| ChunkWritingError::IoError(err.kind()))?; - - trace!("Wrote file to Disk: {:?}", path); - Ok(()) - } } #[async_trait] -impl ChunkIO for ChunkFileManager +impl ChunkIO for ChunkFileManager where - D: 'static + Send + Sync + Sized, - S: ChunkSerializer, + S: ChunkSerializer, { + type Data = SyncChunk; + async fn watch_chunks(&self, folder: &LevelFolder, chunks: &[Vector2]) { // It is intentional that regions are watched multiple times (once per chunk) let mut watchers = self.watchers.write().await; @@ -201,7 +168,7 @@ where &self, folder: &LevelFolder, chunk_coords: &[Vector2], - stream: tokio::sync::mpsc::Sender>, + stream: tokio::sync::mpsc::Sender>, ) { let mut regions_chunks: BTreeMap>> = BTreeMap::new(); @@ -214,9 +181,9 @@ where .or_insert(vec![*at]); } - // we use a Sync Closure with an Async Block to execute the tasks in parallel - // with out waiting the future. Also it improve we File Cache utilizations. - let tasks = regions_chunks.into_iter().map(async |(file_name, chunks)| { + // we use a Sync Closure with an Async Block to execute the tasks concurrently + // Also improves File Cache utilizations. + let region_read_tasks = regions_chunks.into_iter().map(async |(file_name, chunks)| { let path = Self::map_key(folder, &file_name); let chunk_serializer = match self.read_file(&path).await { Ok(chunk_serializer) => chunk_serializer, @@ -231,20 +198,35 @@ where } }; + // Intermediate channel for wrapping the data with the Arc + let (send, mut recv) = mpsc::channel::>(1); + + let intermediary = async { + while let Some(data) = recv.recv().await { + let wrapped_data = data.map_loaded(|data| Arc::new(RwLock::new(data))); + stream + .send(wrapped_data) + .await + .expect("Failed chunk wrapper intermediary"); + } + }; + // We need to block the read to avoid other threads to write/modify the data let serializer = chunk_serializer.read().await; - serializer.get_chunks(&chunks, stream.clone()).await; + let reader = serializer.get_chunks(&chunks, send); + + join!(intermediary, reader); }); - let _ = join_all(tasks).await; + let _ = join_all(region_read_tasks).await; } async fn save_chunks( &self, folder: &LevelFolder, - chunks_data: Vec<(Vector2, D)>, + chunks_data: Vec<(Vector2, SyncChunk)>, ) -> Result<(), ChunkWritingError> { - let mut regions_chunks: BTreeMap> = BTreeMap::new(); + let mut regions_chunks: BTreeMap> = BTreeMap::new(); for (at, chunk) in chunks_data { let key = S::get_chunk_key(&at); @@ -283,23 +265,40 @@ where }?; let mut serializer = chunk_serializer.write().await; - serializer.update_chunks(&chunk_locks).await?; + for chunk_lock in chunk_locks { + let mut chunk = chunk_lock.write().await; + let chunk_is_dirty = chunk.dirty; + // Edge case: this chunk is loaded while we were saving, mark it as cleaned since we are + // updating what we will write here + chunk.dirty = false; + // It is important that we keep the lock after we mark the chunk as clean so no one else + // can modify it + let chunk = chunk.downgrade(); + + // We only need to update the chunk if it is dirty + if chunk_is_dirty { + serializer.update_chunk(&*chunk).await?; + } + } log::trace!("Updated data for file {:?}", path); - // Only write the file if no chunks are being used - if self + let is_watched = self .watchers .read() .await .get(&path) - .is_none_or(|count| count.is_zero()) - { + .is_some_and(|count| !count.is_zero()); + + if serializer.should_write(is_watched) { // With the modification done, we can drop the write lock but keep the read lock // to avoid other threads to write/modify the data, but allow other threads to read it let serializer = serializer.downgrade(); - let serializer_ref = serializer.deref(); - Self::write_file(&path, serializer_ref).await?; - log::trace!("Saved file {:?}", path); + + log::debug!("Writing file for {:?}", path); + serializer + .write(path.clone()) + .await + .map_err(|err| ChunkWritingError::IoError(err.kind()))?; drop(serializer); // If there are still no watchers, drop from the locks diff --git a/pumpkin-world/src/chunk/io/mod.rs b/pumpkin-world/src/chunk/io/mod.rs index 9dfa46436..8d872ee5a 100644 --- a/pumpkin-world/src/chunk/io/mod.rs +++ b/pumpkin-world/src/chunk/io/mod.rs @@ -3,7 +3,6 @@ use std::error; use async_trait::async_trait; use bytes::Bytes; use pumpkin_util::math::vector2::Vector2; -use tokio::io::AsyncWrite; use super::{ChunkReadingError, ChunkWritingError}; use crate::level::LevelFolder; @@ -28,6 +27,16 @@ where Error((Vector2, Err)), } +impl LoadedData { + pub fn map_loaded(self, map: impl FnOnce(D) -> D2) -> LoadedData { + match self { + Self::Loaded(data) => LoadedData::Loaded(map(data)), + Self::Missing(pos) => LoadedData::Missing(pos), + Self::Error(err) => LoadedData::Error(err), + } + } +} + /// Trait to handle the IO of chunks /// for loading and saving chunks data /// can be implemented for different types of IO @@ -36,24 +45,25 @@ where /// The `R` type is the type of the data that will be loaded/saved /// like ChunkData or EntityData #[async_trait] -pub trait ChunkIO +pub trait ChunkIO where Self: Send + Sync, - D: Send + Sized, { + type Data: Send + Sync + Sized; + /// Load the chunks data async fn fetch_chunks( &self, folder: &LevelFolder, chunk_coords: &[Vector2], - stream: tokio::sync::mpsc::Sender>, + stream: tokio::sync::mpsc::Sender>, ); /// Persist the chunks data async fn save_chunks( &self, folder: &LevelFolder, - chunks_data: Vec<(Vector2, D)>, + chunks_data: Vec<(Vector2, Self::Data)>, ) -> Result<(), ChunkWritingError>; /// Tells the `ChunkIO` that these chunks are currently loaded in memory @@ -78,18 +88,21 @@ where #[async_trait] pub trait ChunkSerializer: Send + Sync + Default { type Data: Send + Sync + Sized; + type WriteBackend; /// Get the key for the chunk (like the file name) fn get_chunk_key(chunk: &Vector2) -> String; + fn should_write(&self, is_watched: bool) -> bool; + /// Serialize the data to bytes. - async fn write(&self, w: &mut (impl AsyncWrite + Unpin + Send)) -> Result<(), std::io::Error>; + async fn write(&self, backend: Self::WriteBackend) -> Result<(), std::io::Error>; /// Create a new instance from bytes fn read(r: Bytes) -> Result; - /// Add the chunks data to the serializer - async fn update_chunks(&mut self, chunk_data: &[Self::Data]) -> Result<(), ChunkWritingError>; + /// Add the chunk data to the serializer + async fn update_chunk(&mut self, chunk_data: &Self::Data) -> Result<(), ChunkWritingError>; /// Get the chunks data from the serializer async fn get_chunks( diff --git a/pumpkin-world/src/chunk/mod.rs b/pumpkin-world/src/chunk/mod.rs index 582ee82e0..0102e3051 100644 --- a/pumpkin-world/src/chunk/mod.rs +++ b/pumpkin-world/src/chunk/mod.rs @@ -61,6 +61,7 @@ pub struct ChunkData { /// See `https://minecraft.wiki/w/Heightmap` for more info pub heightmap: ChunkHeightmaps, pub position: Vector2, + pub dirty: bool, } /// # Subchunks diff --git a/pumpkin-world/src/generation/generic_generator.rs b/pumpkin-world/src/generation/generic_generator.rs index 719276ef4..f2e3c4d0b 100644 --- a/pumpkin-world/src/generation/generic_generator.rs +++ b/pumpkin-world/src/generation/generic_generator.rs @@ -76,6 +76,8 @@ impl WorldGenerator for GenericGen subchunks, heightmap: Default::default(), position: at, + // We just generated this chunk! Mark it as dirty + dirty: true, } } } diff --git a/pumpkin-world/src/generation/implementation/test.rs b/pumpkin-world/src/generation/implementation/test.rs index 60a868e7a..153e5cd38 100644 --- a/pumpkin-world/src/generation/implementation/test.rs +++ b/pumpkin-world/src/generation/implementation/test.rs @@ -57,6 +57,8 @@ impl WorldGenerator for TestGenerator { subchunks, heightmap: Default::default(), position: at, + // This chunk was just created! We want to say its been changed + dirty: true, } } } diff --git a/pumpkin-world/src/level.rs b/pumpkin-world/src/level.rs index 17ed10ddf..7343661d1 100644 --- a/pumpkin-world/src/level.rs +++ b/pumpkin-world/src/level.rs @@ -3,7 +3,7 @@ use std::{fs, path::PathBuf, sync::Arc}; use dashmap::{DashMap, Entry}; use log::trace; use num_traits::Zero; -use pumpkin_config::{ADVANCED_CONFIG, chunk::ChunkFormat}; +use pumpkin_config::{advanced_config, chunk::ChunkFormat}; use pumpkin_util::math::vector2::Vector2; use tokio::{ sync::{RwLock, mpsc}, @@ -40,9 +40,16 @@ pub struct Level { pub level_info: LevelData, world_info_writer: Arc, level_folder: LevelFolder, + + // Holds this level's spawn chunks, which are always loaded + spawn_chunks: Arc, SyncChunk>>, + + // Chunks that are paired with chunk watchers. When a chunk is no longer watched, it is removed + // from the loaded chunks map and sent to the underlying ChunkIO loaded_chunks: Arc, SyncChunk>>, chunk_watchers: Arc, usize>>, - chunk_saver: Arc>, + + chunk_saver: Arc>, world_gen: Arc, // Gets unlocked when dropped // TODO: Make this a trait @@ -103,7 +110,7 @@ impl Level { let seed = Seed(level_info.world_gen_settings.seed as u64); let world_gen = get_world_gen(seed).into(); - let chunk_saver: Arc> = match ADVANCED_CONFIG.chunk.format { + let chunk_saver: Arc> = match advanced_config().chunk.format { //ChunkFormat::Anvil => (Arc::new(AnvilChunkFormat), Arc::new(AnvilChunkFormat)), ChunkFormat::Linear => Arc::new(ChunkFileManager::::default()), ChunkFormat::Anvil => Arc::new(ChunkFileManager::::default()), @@ -115,6 +122,7 @@ impl Level { world_info_writer: Arc::new(AnvilLevelInfo), level_folder, chunk_saver, + spawn_chunks: Arc::new(DashMap::new()), loaded_chunks: Arc::new(DashMap::new()), chunk_watchers: Arc::new(DashMap::new()), level_info, @@ -315,7 +323,7 @@ impl Level { let chunk_saver = self.chunk_saver.clone(); let level_folder = self.level_folder.clone(); - trace!("Writing chunks to disk {:}", chunks_to_write.len()); + trace!("Sending chunks to ChunkIO {:}", chunks_to_write.len()); if let Err(error) = chunk_saver .save_chunks(&level_folder, chunks_to_write) .await @@ -324,6 +332,22 @@ impl Level { } } + /// Initializes the spawn chunks to these chunks + pub async fn read_spawn_chunks(self: &Arc, chunks: &[Vector2]) { + let (send, mut recv) = mpsc::unbounded_channel(); + + let fetcher = self.fetch_chunks(chunks, send); + let handler = async { + while let Some((chunk, _)) = recv.recv().await { + let pos = chunk.read().await.position; + self.spawn_chunks.insert(pos, chunk); + } + }; + + let _ = tokio::join!(fetcher, handler); + log::debug!("Read {} chunks as spawn chunks", chunks.len()); + } + /// Reads/Generates many chunks in a world /// Note: The order of the output chunks will almost never be in the same order as the order of input chunks pub async fn fetch_chunks( @@ -350,6 +374,11 @@ impl Level { for chunk in chunks { if let Some(chunk) = self.loaded_chunks.get(chunk) { send_chunk(false, chunk.value().clone(), &channel); + } else if let Some(spawn_chunk) = self.spawn_chunks.get(chunk) { + // Also clone the arc into the loaded chunks + self.loaded_chunks + .insert(*chunk, spawn_chunk.value().clone()); + send_chunk(false, spawn_chunk.value().clone(), &channel); } else { remaining_chunks.push(*chunk); } diff --git a/pumpkin/src/entity/living.rs b/pumpkin/src/entity/living.rs index 4a1899524..b4db77ce8 100644 --- a/pumpkin/src/entity/living.rs +++ b/pumpkin/src/entity/living.rs @@ -4,7 +4,7 @@ use std::{collections::HashMap, sync::atomic::AtomicI32}; use crate::server::Server; use async_trait::async_trait; use crossbeam::atomic::AtomicCell; -use pumpkin_config::ADVANCED_CONFIG; +use pumpkin_config::advanced_config; use pumpkin_data::entity::{EffectType, EntityStatus}; use pumpkin_data::{damage::DamageType, sound::Sound}; use pumpkin_nbt::tag::NbtTag; @@ -274,7 +274,7 @@ impl EntityBase for LivingEntity { if !self.check_damage(amount) { return false; } - let config = &ADVANCED_CONFIG.pvp; + let config = &advanced_config().pvp; if !self .damage_with_context(amount, damage_type, None, None, None) diff --git a/pumpkin/src/entity/player.rs b/pumpkin/src/entity/player.rs index 01594a1cd..68b022667 100644 --- a/pumpkin/src/entity/player.rs +++ b/pumpkin/src/entity/player.rs @@ -11,7 +11,7 @@ use std::{ use async_trait::async_trait; use crossbeam::atomic::AtomicCell; -use pumpkin_config::{ADVANCED_CONFIG, BASIC_CONFIG}; +use pumpkin_config::{BASIC_CONFIG, advanced_config}; use pumpkin_data::{ block::BlockState, damage::DamageType, @@ -297,7 +297,7 @@ impl Player { .iter() .find(|op| op.uuid == gameprofile_clone.id) .map_or( - AtomicCell::new(ADVANCED_CONFIG.commands.default_op_level), + AtomicCell::new(advanced_config().commands.default_op_level), |op| AtomicCell::new(op.level), ), inventory: Mutex::new(PlayerInventory::new()), @@ -359,7 +359,7 @@ impl Player { let world = self.world().await; let victim_entity = victim.get_entity(); let attacker_entity = &self.living_entity.entity; - let config = &ADVANCED_CONFIG.pvp; + let config = &advanced_config().pvp; let inventory = self.inventory().lock().await; let item_slot = inventory.held_item(); diff --git a/pumpkin/src/lib.rs b/pumpkin/src/lib.rs index 0dd110556..60acdf3dc 100644 --- a/pumpkin/src/lib.rs +++ b/pumpkin/src/lib.rs @@ -7,7 +7,7 @@ use log::{Level, LevelFilter, Log}; use net::PacketHandlerState; use plugin::PluginManager; use plugin::server::server_command::ServerCommandEvent; -use pumpkin_config::{ADVANCED_CONFIG, BASIC_CONFIG}; +use pumpkin_config::{BASIC_CONFIG, advanced_config}; use pumpkin_macros::send_cancellable; use pumpkin_util::text::TextComponent; use rustyline_async::{Readline, ReadlineEvent}; @@ -100,10 +100,10 @@ impl Log for ReadlineLogWrapper { } pub static LOGGER_IMPL: LazyLock> = LazyLock::new(|| { - if ADVANCED_CONFIG.logging.enabled { + if advanced_config().logging.enabled { let mut config = simplelog::ConfigBuilder::new(); - if ADVANCED_CONFIG.logging.timestamp { + if advanced_config().logging.timestamp { config.set_time_format_custom(time::macros::format_description!( "[year]-[month]-[day] [hour]:[minute]:[second]" )); @@ -112,7 +112,7 @@ pub static LOGGER_IMPL: LazyLock> = La config.set_time_level(LevelFilter::Off); } - if !ADVANCED_CONFIG.logging.color { + if !advanced_config().logging.color { for level in Level::iter() { config.set_level_color(level, None); } @@ -121,7 +121,7 @@ pub static LOGGER_IMPL: LazyLock> = La config.set_write_log_enable_colors(true); } - if !ADVANCED_CONFIG.logging.threads { + if !advanced_config().logging.threads { config.set_thread_level(LevelFilter::Off); } else { config.set_thread_level(LevelFilter::Info); @@ -134,7 +134,7 @@ pub static LOGGER_IMPL: LazyLock> = La .and_then(Result::ok) .unwrap_or(LevelFilter::Info); - if ADVANCED_CONFIG.commands.use_console { + if advanced_config().commands.use_console { match Readline::new("$ ".to_owned()) { Ok((rl, stdout)) => { let logger = simplelog::WriteLogger::new(level, config.build(), stdout); @@ -187,12 +187,8 @@ impl PumpkinServer { pub async fn new() -> Self { let server = Arc::new(Server::new()); - // Spawn chunks are never unloaded - for world in server.worlds.read().await.iter() { - world - .level - .mark_chunks_as_newly_watched(&Server::spawn_chunks()) - .await; + for world in &*server.worlds.read().await { + world.level.read_spawn_chunks(&Server::spawn_chunks()).await; } // Setup the TCP server socket. @@ -204,7 +200,7 @@ impl PumpkinServer { .local_addr() .expect("Unable to get the address of server!"); - let rcon = ADVANCED_CONFIG.networking.rcon.clone(); + let rcon = advanced_config().networking.rcon.clone(); let mut ticker = Ticker::new(BASIC_CONFIG.tps); @@ -223,12 +219,12 @@ impl PumpkinServer { }); } - if ADVANCED_CONFIG.networking.query.enabled { + if advanced_config().networking.query.enabled { log::info!("Query protocol enabled. Starting..."); tokio::spawn(query::start_query_handler(server.clone(), addr)); } - if ADVANCED_CONFIG.networking.lan_broadcast.enabled { + if advanced_config().networking.lan_broadcast.enabled { log::info!("LAN broadcast enabled. Starting..."); tokio::spawn(lan_broadcast::start_lan_broadcast(addr)); } diff --git a/pumpkin/src/net/authentication.rs b/pumpkin/src/net/authentication.rs index 7623d2019..f56972dfe 100644 --- a/pumpkin/src/net/authentication.rs +++ b/pumpkin/src/net/authentication.rs @@ -1,7 +1,7 @@ use std::{collections::HashMap, net::IpAddr}; use base64::{Engine, engine::general_purpose}; -use pumpkin_config::{ADVANCED_CONFIG, networking::auth::TextureConfig}; +use pumpkin_config::{advanced_config, networking::auth::TextureConfig}; use pumpkin_protocol::Property; use reqwest::{StatusCode, Url}; use serde::Deserialize; @@ -50,12 +50,12 @@ pub async fn authenticate( ip: &IpAddr, auth_client: &reqwest::Client, ) -> Result { - let address = if ADVANCED_CONFIG + let address = if advanced_config() .networking .authentication .prevent_proxy_connections { - let auth_url = ADVANCED_CONFIG + let auth_url = advanced_config() .networking .authentication .prevent_proxy_connection_auth_url @@ -67,7 +67,7 @@ pub async fn authenticate( .replace("{server_hash}", server_hash) .replace("{ip}", &ip.to_string()) } else { - let auth_url = ADVANCED_CONFIG + let auth_url = advanced_config() .networking .authentication .url diff --git a/pumpkin/src/net/lan_broadcast.rs b/pumpkin/src/net/lan_broadcast.rs index 9f02635f8..ad231089d 100644 --- a/pumpkin/src/net/lan_broadcast.rs +++ b/pumpkin/src/net/lan_broadcast.rs @@ -1,4 +1,4 @@ -use pumpkin_config::{ADVANCED_CONFIG, BASIC_CONFIG}; +use pumpkin_config::{BASIC_CONFIG, advanced_config}; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::time::Duration; use tokio::net::UdpSocket; @@ -10,7 +10,7 @@ const BROADCAST_ADDRESS: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(224, 0, 2, 60)), 4445); pub async fn start_lan_broadcast(bound_addr: SocketAddr) { - let port = ADVANCED_CONFIG.networking.lan_broadcast.port.unwrap_or(0); + let port = advanced_config().networking.lan_broadcast.port.unwrap_or(0); let socket = UdpSocket::bind(format!("0.0.0.0:{port}")) .await @@ -21,7 +21,7 @@ pub async fn start_lan_broadcast(bound_addr: SocketAddr) { let mut interval = time::interval(Duration::from_millis(1500)); let motd: String; - let advanced_motd = &ADVANCED_CONFIG + let advanced_motd = &advanced_config() .networking .lan_broadcast .motd diff --git a/pumpkin/src/net/packet/config.rs b/pumpkin/src/net/packet/config.rs index ec50d1aa2..5f9495550 100644 --- a/pumpkin/src/net/packet/config.rs +++ b/pumpkin/src/net/packet/config.rs @@ -6,7 +6,7 @@ use crate::{ server::Server, }; use core::str; -use pumpkin_config::ADVANCED_CONFIG; +use pumpkin_config::advanced_config; use pumpkin_protocol::{ ConnectionState, client::config::{CFinishConfig, CRegistryData}, @@ -70,7 +70,7 @@ impl Client { } pub async fn handle_resource_pack_response(&self, packet: SConfigResourcePack) { - let resource_config = &ADVANCED_CONFIG.resource_pack; + let resource_config = &advanced_config().resource_pack; if resource_config.enabled { let expected_uuid = uuid::Uuid::new_v3(&uuid::Uuid::NAMESPACE_DNS, resource_config.url.as_bytes()); diff --git a/pumpkin/src/net/packet/login.rs b/pumpkin/src/net/packet/login.rs index 34535b3ba..3a66e85d8 100644 --- a/pumpkin/src/net/packet/login.rs +++ b/pumpkin/src/net/packet/login.rs @@ -1,6 +1,6 @@ use std::sync::LazyLock; -use pumpkin_config::{ADVANCED_CONFIG, BASIC_CONFIG}; +use pumpkin_config::{BASIC_CONFIG, advanced_config}; use pumpkin_protocol::{ ConnectionState, KnownPack, Label, Link, LinkType, client::{ @@ -27,47 +27,47 @@ use crate::{ static LINKS: LazyLock> = LazyLock::new(|| { let mut links: Vec = Vec::new(); - let bug_report = &ADVANCED_CONFIG.server_links.bug_report; + let bug_report = &advanced_config().server_links.bug_report; if !bug_report.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::BugReport), bug_report)); } - let support = &ADVANCED_CONFIG.server_links.support; + let support = &advanced_config().server_links.support; if !support.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::Support), support)); } - let status = &ADVANCED_CONFIG.server_links.status; + let status = &advanced_config().server_links.status; if !status.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::Status), status)); } - let feedback = &ADVANCED_CONFIG.server_links.feedback; + let feedback = &advanced_config().server_links.feedback; if !feedback.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::Feedback), feedback)); } - let community = &ADVANCED_CONFIG.server_links.community; + let community = &advanced_config().server_links.community; if !community.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::Community), community)); } - let website = &ADVANCED_CONFIG.server_links.website; + let website = &advanced_config().server_links.website; if !website.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::Website), website)); } - let forums = &ADVANCED_CONFIG.server_links.forums; + let forums = &advanced_config().server_links.forums; if !forums.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::Forums), forums)); } - let news = &ADVANCED_CONFIG.server_links.news; + let news = &advanced_config().server_links.news; if !news.is_empty() { links.push(Link::new(Label::BuiltIn(LinkType::News), news)); } - let announcements = &ADVANCED_CONFIG.server_links.announcements; + let announcements = &advanced_config().server_links.announcements; if !announcements.is_empty() { links.push(Link::new( Label::BuiltIn(LinkType::Announcements), @@ -75,7 +75,7 @@ static LINKS: LazyLock> = LazyLock::new(|| { )); } - for (key, value) in &ADVANCED_CONFIG.server_links.custom { + for (key, value) in &advanced_config().server_links.custom { links.push(Link::new( Label::TextComponent(TextComponent::text(key).into()), value, @@ -109,7 +109,7 @@ impl Client { // default game profile, when no online mode // TODO: make offline uuid let mut gameprofile = self.gameprofile.lock().await; - let proxy = &ADVANCED_CONFIG.networking.proxy; + let proxy = &advanced_config().networking.proxy; if proxy.enabled { if proxy.velocity.enabled { velocity::velocity_login(self).await; @@ -150,7 +150,7 @@ impl Client { ) .await; } else { - if ADVANCED_CONFIG.networking.packet_compression.enabled { + if advanced_config().networking.packet_compression.enabled { self.enable_compression().await; } self.finish_login(&profile).await; @@ -239,14 +239,14 @@ impl Client { return; } - if ADVANCED_CONFIG.networking.packet_compression.enabled { + if advanced_config().networking.packet_compression.enabled { self.enable_compression().await; } self.finish_login(profile).await; } async fn enable_compression(&self) { - let compression = ADVANCED_CONFIG.networking.packet_compression.info.clone(); + let compression = advanced_config().networking.packet_compression.info.clone(); self.send_packet(&CSetCompression::new(compression.threshold.into())) .await; self.set_compression(Some(compression)).await; @@ -270,13 +270,13 @@ impl Client { // Check if player should join if let Some(actions) = &profile.profile_actions { - if ADVANCED_CONFIG + if advanced_config() .networking .authentication .player_profile .allow_banned_players { - for allowed in &ADVANCED_CONFIG + for allowed in &advanced_config() .networking .authentication .player_profile @@ -297,7 +297,7 @@ impl Client { for property in &profile.properties { authentication::validate_textures( property, - &ADVANCED_CONFIG.networking.authentication.textures, + &advanced_config().networking.authentication.textures, ) .map_err(AuthError::TextureError)?; } @@ -317,7 +317,7 @@ impl Client { } pub async fn handle_plugin_response(&self, plugin_response: SLoginPluginResponse) { log::debug!("Handling plugin"); - let velocity_config = &ADVANCED_CONFIG.networking.proxy.velocity; + let velocity_config = &advanced_config().networking.proxy.velocity; if velocity_config.enabled { let mut address = self.address.lock().await; match velocity::receive_velocity_plugin_response( @@ -340,7 +340,7 @@ impl Client { self.connection_state.store(ConnectionState::Config); self.send_packet(&server.get_branding()).await; - if ADVANCED_CONFIG.server_links.enabled { + if advanced_config().server_links.enabled { self.send_packet(&CConfigServerLinks::new( &VarInt(LINKS.len() as i32), &LINKS, @@ -356,7 +356,7 @@ impl Client { ])) .await; - let resource_config = &ADVANCED_CONFIG.resource_pack; + let resource_config = &advanced_config().resource_pack; if resource_config.enabled { let uuid = Uuid::new_v3(&uuid::Uuid::NAMESPACE_DNS, resource_config.url.as_bytes()); let resource_pack = CConfigAddResourcePack::new( diff --git a/pumpkin/src/net/packet/play.rs b/pumpkin/src/net/packet/play.rs index 8e0c03030..83119efd1 100644 --- a/pumpkin/src/net/packet/play.rs +++ b/pumpkin/src/net/packet/play.rs @@ -15,7 +15,7 @@ use crate::{ server::Server, world::chunker, }; -use pumpkin_config::ADVANCED_CONFIG; +use pumpkin_config::advanced_config; use pumpkin_data::block::{Block, HorizontalFacing}; use pumpkin_data::entity::{EntityType, entity_from_egg}; use pumpkin_data::item::Item; @@ -455,7 +455,7 @@ impl Player { .await; }); - if ADVANCED_CONFIG.commands.log_console { + if advanced_config().commands.log_console { log::info!( "Player ({}): executed command /{}", self.gameprofile.name, @@ -854,7 +854,7 @@ impl Player { match action { ActionType::Attack => { let entity_id = interact.entity_id; - let config = &ADVANCED_CONFIG.pvp; + let config = &advanced_config().pvp; // TODO: do validation and stuff if !config.enabled { return; diff --git a/pumpkin/src/net/query.rs b/pumpkin/src/net/query.rs index f44a58570..868cfa70b 100644 --- a/pumpkin/src/net/query.rs +++ b/pumpkin/src/net/query.rs @@ -6,7 +6,7 @@ use std::{ time::Duration, }; -use pumpkin_config::{ADVANCED_CONFIG, BASIC_CONFIG}; +use pumpkin_config::{BASIC_CONFIG, advanced_config}; use pumpkin_protocol::query::{ CBasicStatus, CFullStatus, CHandshake, PacketType, RawQueryPacket, SHandshake, SStatusRequest, }; @@ -17,7 +17,7 @@ use crate::server::{CURRENT_MC_VERSION, Server}; pub async fn start_query_handler(server: Arc, bound_addr: SocketAddr) { let mut query_addr = bound_addr; - if let Some(port) = ADVANCED_CONFIG.networking.query.port { + if let Some(port) = advanced_config().networking.query.port { query_addr.set_port(port); } diff --git a/pumpkin/src/net/rcon/mod.rs b/pumpkin/src/net/rcon/mod.rs index c89d2bd8e..f235ae363 100644 --- a/pumpkin/src/net/rcon/mod.rs +++ b/pumpkin/src/net/rcon/mod.rs @@ -1,7 +1,7 @@ use std::net::SocketAddr; use packet::{ClientboundPacket, Packet, PacketError, ServerboundPacket}; -use pumpkin_config::{ADVANCED_CONFIG, RCONConfig}; +use pumpkin_config::{RCONConfig, advanced_config}; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -83,7 +83,7 @@ impl RCONClient { let Some(packet) = self.receive_packet().await? else { return Ok(()); }; - let config = &ADVANCED_CONFIG.networking.rcon; + let config = &advanced_config().networking.rcon; match packet.get_type() { ServerboundPacket::Auth => { if packet.get_body() == password { diff --git a/pumpkin/src/server/mod.rs b/pumpkin/src/server/mod.rs index de03108ee..d6b39b1cd 100644 --- a/pumpkin/src/server/mod.rs +++ b/pumpkin/src/server/mod.rs @@ -12,7 +12,7 @@ use crate::{ }; use connection_cache::{CachedBranding, CachedStatus}; use key_store::KeyStore; -use pumpkin_config::{ADVANCED_CONFIG, BASIC_CONFIG}; +use pumpkin_config::{BASIC_CONFIG, advanced_config}; use pumpkin_data::block::Block; use pumpkin_inventory::drag_handler::DragHandler; use pumpkin_inventory::{Container, OpenContainer}; @@ -81,10 +81,10 @@ impl Server { let auth_client = BASIC_CONFIG.online_mode.then(|| { reqwest::Client::builder() .connect_timeout(Duration::from_millis(u64::from( - ADVANCED_CONFIG.networking.authentication.connect_timeout, + advanced_config().networking.authentication.connect_timeout, ))) .read_timeout(Duration::from_millis(u64::from( - ADVANCED_CONFIG.networking.authentication.read_timeout, + advanced_config().networking.authentication.read_timeout, ))) .build() .expect("Failed to to make reqwest client") diff --git a/pumpkin/src/world/mod.rs b/pumpkin/src/world/mod.rs index 12a6c2b46..4dba872ba 100644 --- a/pumpkin/src/world/mod.rs +++ b/pumpkin/src/world/mod.rs @@ -110,10 +110,6 @@ impl PumpkinError for GetBlockError { pub struct World { /// The underlying level, responsible for chunk management and terrain generation. pub level: Arc, - // An LFU cache for storing the most frequently used 16 chunks that are singleton called (not a - // part of chunk loading), if a chunk is called in this manner, it is likely to be called again, - // so we will watch all chunks while they remain in this cache - single_chunk_lfu: Mutex<[Vector2; 16]>, /// A map of active players within the world, keyed by their unique UUID. pub players: Arc>>>, /// A map of active entities within the world, keyed by their unique UUID. @@ -137,7 +133,6 @@ impl World { pub fn load(level: Level, dimension_type: DimensionType) -> Self { Self { level: Arc::new(level), - single_chunk_lfu: Mutex::new([Vector2::default(); 16]), players: Arc::new(RwLock::new(HashMap::new())), entities: Arc::new(RwLock::new(HashMap::new())), scoreboard: Mutex::new(Scoreboard::new()), @@ -1055,12 +1050,11 @@ impl World { let relative = ChunkRelativeBlockCoordinates::from(relative_coordinates); let chunk = self.receive_chunk(chunk_coordinate).await.0; - let replaced_block_state_id = chunk.read().await.subchunks.get_block(relative).unwrap(); - chunk - .write() - .await - .subchunks - .set_block(relative, block_state_id); + let mut chunk = chunk.write().await; + chunk.dirty = true; + let replaced_block_state_id = chunk.subchunks.get_block(relative).unwrap(); + chunk.subchunks.set_block(relative, block_state_id); + drop(chunk); self.broadcast_packet_all(&CBlockUpdate::new( position, @@ -1103,47 +1097,6 @@ impl World { pub async fn receive_chunk(&self, chunk_pos: Vector2) -> (Arc>, bool) { let mut receiver = self.receive_chunks(vec![chunk_pos], false); - // If we are only getting one chunk, we are probably doing something that requires multiple - // calls to it. "Watch" it temp. - - let mut lfu = self.single_chunk_lfu.lock().await; - - #[allow(clippy::single_match_else)] - let pos_to_clean = match lfu.iter().position(|pos| *pos == chunk_pos) { - Some(index) => { - if index > 0 { - lfu[..=index].rotate_right(1); - } - None - } - None => { - // The position isn't in the cache - lfu.rotate_right(1); - let to_remove = lfu[0]; - lfu[0] = chunk_pos; - self.level.mark_chunk_as_newly_watched(chunk_pos).await; - - if to_remove == Vector2::::default() { - // This is our dummy value and spawn chunks are watched anyway - // TODO: What if we call this on our default? - None - } else { - Some(to_remove) - } - } - }; - - if let Some(pos) = pos_to_clean { - self.level.mark_chunk_as_not_watched(pos).await; - if !self.level.is_chunk_watched(&pos) { - log::trace!( - "Chunk {:?} evicted from single chunk cache... cleaning", - chunk_pos - ); - self.level.clean_chunk(&pos).await; - } - } - receiver .recv() .await