majjo/Pumpkin
1
0
Fork 0
mirror of https://github.com/Pumpkin-MC/Pumpkin.git synced 2026-08-30 20:14:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

ChunkIO api and world saving optimizations (#624)

Browse Source
This commit is contained in:
kralverde
2025-03-12 08:40:53 -10:00
committed by GitHub
parent 2bc19bacb5
commit 300ccde108
26 changed files with 997 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pumpkin-config/Cargo.toml
View File

@@ -10,3 +10,7 @@ log.workspace = true
uuid.workspace = true
toml = "0.8"
[features]
# Adds helper to change the config at runtime
test_helper = []

5
pumpkin-config/src/chunk.rs
View File

@@ -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,

47
pumpkin-config/src/lib.rs
View File

@@ -36,10 +36,46 @@ use resource_pack::ResourcePackConfig;
const CONFIG_ROOT_FOLDER: &str = "config/";
pub static ADVANCED_CONFIG: LazyLock<AdvancedConfiguration> =
LazyLock::new(AdvancedConfiguration::load);
pub static BASIC_CONFIG: LazyLock<BasicConfiguration> = LazyLock::new(|| {
let exec_dir = env::current_dir().unwrap();
BasicConfiguration::load(&exec_dir)
});
pub static BASIC_CONFIG: LazyLock<BasicConfiguration> = LazyLock::new(BasicConfiguration::load);
#[cfg(not(feature = "test_helper"))]
static ADVANCED_CONFIG: LazyLock<AdvancedConfiguration> = 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");

11
pumpkin-world/Cargo.toml
View File

@@ -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"

22
pumpkin-world/benches/chunk_noise_populate.rs
View File

@@ -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<Level>, positions: Vec<Vector2<i32>>) {
let _ = x;
}
}
/*
async fn test_reads_parallel(level: &Arc<Level>, positions: Vec<Vector2<i32>>, threads: usize) {
let mut tasks = JoinSet::new();
@@ -45,11 +47,13 @@ async fn test_reads_parallel(level: &Arc<Level>, positions: Vec<Vector2<i32>>, t
let _ = tasks.join_all().await;
}
*/
async fn test_writes(level: &Arc<Level>, chunks: Vec<(Vector2<i32>, Arc<RwLock<ChunkData>>)>) {
level.write_chunks(chunks).await;
}
/*
async fn test_writes_parallel(
level: &Arc<Level>,
chunks: Vec<(Vector2<i32>, Arc<RwLock<ChunkData>>)>,
@@ -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);

789
pumpkin-world/src/chunk/format/anvil.rs
View File

@@ -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<usize>),
}
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<AnvilChunkData>; CHUNK_COUNT],
chunks_data: [Option<AnvilChunkMetadata>; CHUNK_COUNT],
end_sector: u32,
write_action: Mutex<WriteAction>,
}
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<Self, ChunkReadingError> {
let mut bytes = bytes;
// Minus one for the compression byte
@@ -269,14 +304,19 @@ impl AnvilChunkData {
Ok(chunk)
}
fn from_chunk(chunk: &ChunkData) -> Result<Self, ChunkWritingError> {
fn from_chunk(
chunk: &ChunkData,
compression: Option<Compression>,
) -> Result<Self, ChunkWritingError> {
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<i32>) -> 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::<Vec<_>>();
// 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<i32>) -> 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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
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<i32>],
stream: tokio::sync::mpsc::Sender<LoadedData<SyncChunk, ChunkReadingError>>,
stream: tokio::sync::mpsc::Sender<LoadedData<ChunkData, ChunkReadingError>>,
) {
// 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<Vec<u8>, 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<AnvilChunkFile>,
folder: &LevelFolder,
chunks: &[(Vector2<i32>, 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::<Vec<_>>();
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::<Vec<_>>();
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::<AnvilChunkFile>::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::<AnvilChunkFile>::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::<AnvilChunkFile>::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::<AnvilChunkFile>::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::<AnvilChunkFile>::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::<Vec<_>>();
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::<Vec<_>>();
// Create a new manager to ensure nothing is cached
let chunk_saver = ChunkFileManager::<AnvilChunkFile>::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

86
pumpkin-world/src/chunk/format/linear.rs
View File

@@ -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<i32>) -> 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<u8> = 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<i32>],
stream: tokio::sync::mpsc::Sender<LoadedData<SyncChunk, ChunkReadingError>>,
stream: tokio::sync::mpsc::Sender<LoadedData<ChunkData, ChunkReadingError>>,
) {
// 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,

2
pumpkin-world/src/chunk/format/mod.rs
View File

@@ -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,
})
}
}

125
pumpkin-world/src/chunk/io/chunk_file_manager.rs
View File

@@ -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<S: ChunkSerializer> {
pub struct ChunkFileManager<S: ChunkSerializer<WriteBackend = PathBuf>> {
// Dashmap has rw-locks on shards, but we want per-serializer
file_locks: RwLock<BTreeMap<PathBuf, SerializerCacheEntry<S>>>,
watchers: RwLock<BTreeMap<PathBuf, usize>>,
@@ -37,7 +38,7 @@ pub struct ChunkFileManager<S: ChunkSerializer> {
//to avoid clippy warnings we extract the type alias
type SerializerCacheEntry<S> = OnceCell<Arc<RwLock<S>>>;
impl<S: ChunkSerializer> Default for ChunkFileManager<S> {
impl<S: ChunkSerializer<WriteBackend = PathBuf>> Default for ChunkFileManager<S> {
fn default() -> Self {
Self {
file_locks: RwLock::new(BTreeMap::new()),
@@ -46,12 +47,12 @@ impl<S: ChunkSerializer> Default for ChunkFileManager<S> {
}
}
impl<S: ChunkSerializer> ChunkFileManager<S> {
impl<S: ChunkSerializer<WriteBackend = PathBuf>> ChunkFileManager<S> {
fn map_key(folder: &LevelFolder, file_name: &str) -> PathBuf {
folder.region_folder.join(file_name)
}
pub async fn read_file(&self, path: &Path) -> Result<Arc<RwLock<S>>, ChunkReadingError> {
async fn read_file(&self, path: &Path) -> Result<Arc<RwLock<S>>, 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<S: ChunkSerializer> ChunkFileManager<S> {
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<S, D> ChunkIO<D> for ChunkFileManager<S>
impl<S> ChunkIO for ChunkFileManager<S>
where
D: 'static + Send + Sync + Sized,
S: ChunkSerializer<Data = D>,
S: ChunkSerializer<Data = ChunkData, WriteBackend = PathBuf>,
{
type Data = SyncChunk;
async fn watch_chunks(&self, folder: &LevelFolder, chunks: &[Vector2<i32>]) {
// 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<i32>],
stream: tokio::sync::mpsc::Sender<LoadedData<D, ChunkReadingError>>,
stream: tokio::sync::mpsc::Sender<LoadedData<SyncChunk, ChunkReadingError>>,
) {
let mut regions_chunks: BTreeMap<String, Vec<Vector2<i32>>> = 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<RwLock>
let (send, mut recv) = mpsc::channel::<LoadedData<ChunkData, ChunkReadingError>>(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<i32>, D)>,
chunks_data: Vec<(Vector2<i32>, SyncChunk)>,
) -> Result<(), ChunkWritingError> {
let mut regions_chunks: BTreeMap<String, Vec<D>> = BTreeMap::new();
let mut regions_chunks: BTreeMap<String, Vec<SyncChunk>> = 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

29
pumpkin-world/src/chunk/io/mod.rs
View File

@@ -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<i32>, Err)),
}
impl<D: Send, E: error::Error> LoadedData<D, E> {
pub fn map_loaded<D2: Send>(self, map: impl FnOnce(D) -> D2) -> LoadedData<D2, E> {
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<D>
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<i32>],
stream: tokio::sync::mpsc::Sender<LoadedData<D, ChunkReadingError>>,
stream: tokio::sync::mpsc::Sender<LoadedData<Self::Data, ChunkReadingError>>,
);
/// Persist the chunks data
async fn save_chunks(
&self,
folder: &LevelFolder,
chunks_data: Vec<(Vector2<i32>, D)>,
chunks_data: Vec<(Vector2<i32>, 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<i32>) -> 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<Self, ChunkReadingError>;
/// 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(

1
pumpkin-world/src/chunk/mod.rs
View File

@@ -61,6 +61,7 @@ pub struct ChunkData {
/// See `https://minecraft.wiki/w/Heightmap` for more info
pub heightmap: ChunkHeightmaps,
pub position: Vector2<i32>,
pub dirty: bool,
}
/// # Subchunks

2
pumpkin-world/src/generation/generic_generator.rs
View File

@@ -76,6 +76,8 @@ impl<B: BiomeGenerator, T: PerlinTerrainGenerator> WorldGenerator for GenericGen
subchunks,
heightmap: Default::default(),
position: at,
// We just generated this chunk! Mark it as dirty
dirty: true,
}
}
}

2
pumpkin-world/src/generation/implementation/test.rs
View File

@@ -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,
}
}
}

37
pumpkin-world/src/level.rs
View File

@@ -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<dyn WorldInfoWriter>,
level_folder: LevelFolder,
// Holds this level's spawn chunks, which are always loaded
spawn_chunks: Arc<DashMap<Vector2<i32>, 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<DashMap<Vector2<i32>, SyncChunk>>,
chunk_watchers: Arc<DashMap<Vector2<i32>, usize>>,
chunk_saver: Arc<dyn ChunkIO<SyncChunk>>,
chunk_saver: Arc<dyn ChunkIO<Data = SyncChunk>>,
world_gen: Arc<dyn WorldGenerator>,
// 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<dyn ChunkIO<SyncChunk>> = match ADVANCED_CONFIG.chunk.format {
let chunk_saver: Arc<dyn ChunkIO<Data = SyncChunk>> = match advanced_config().chunk.format {
//ChunkFormat::Anvil => (Arc::new(AnvilChunkFormat), Arc::new(AnvilChunkFormat)),
ChunkFormat::Linear => Arc::new(ChunkFileManager::<LinearFile>::default()),
ChunkFormat::Anvil => Arc::new(ChunkFileManager::<AnvilChunkFile>::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<Self>, chunks: &[Vector2<i32>]) {
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);
}

4
pumpkin/src/entity/living.rs
View File

@@ -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)

6
pumpkin/src/entity/player.rs
View File

@@ -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();

26
pumpkin/src/lib.rs
View File

@@ -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<Option<(ReadlineLogWrapper, LevelFilter)>> = 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<Option<(ReadlineLogWrapper, LevelFilter)>> = 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<Option<(ReadlineLogWrapper, LevelFilter)>> = 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<Option<(ReadlineLogWrapper, LevelFilter)>> = 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));
}

8
pumpkin/src/net/authentication.rs
View File

@@ -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<GameProfile, AuthError> {
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

6
pumpkin/src/net/lan_broadcast.rs
View File

@@ -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

4
pumpkin/src/net/packet/config.rs
View File

@@ -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());

42
pumpkin/src/net/packet/login.rs
View File

@@ -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<Vec<Link>> = LazyLock::new(|| {
let mut links: Vec<Link> = 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<Vec<Link>> = 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(

6
pumpkin/src/net/packet/play.rs
View File

@@ -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;

4
pumpkin/src/net/query.rs
View File

@@ -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<Server>, 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);
}

4
pumpkin/src/net/rcon/mod.rs
View File

@@ -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 {

6
pumpkin/src/server/mod.rs
View File

@@ -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")

57
pumpkin/src/world/mod.rs
View File

@@ -110,10 +110,6 @@ impl PumpkinError for GetBlockError {
pub struct World {
/// The underlying level, responsible for chunk management and terrain generation.
pub level: Arc<Level>,
// 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<i32>; 16]>,
/// A map of active players within the world, keyed by their unique UUID.
pub players: Arc<RwLock<HashMap<uuid::Uuid, Arc<Player>>>>,
/// 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<i32>) -> (Arc<RwLock<ChunkData>>, 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::<i32>::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