From a31c9ec703fa924f8ae5b07ceddacbd5c83cb9c8 Mon Sep 17 00:00:00 2001 From: Alexander Medvedev Date: Wed, 25 Jun 2025 03:47:05 +0200 Subject: [PATCH] Implement Entity data Saving/Reading --- README.md | 2 +- pumpkin-nbt/src/compound.rs | 12 +- pumpkin-nbt/src/deserializer.rs | 64 ++--- pumpkin-nbt/src/serializer.rs | 4 +- pumpkin-nbt/src/tag.rs | 14 +- pumpkin-world/src/chunk/format/anvil.rs | 3 + pumpkin-world/src/chunk/format/linear.rs | 14 +- pumpkin-world/src/chunk/format/mod.rs | 108 +++++++- pumpkin-world/src/chunk/mod.rs | 7 + pumpkin-world/src/inventory/inventory.rs | 2 +- pumpkin-world/src/level.rs | 321 ++++++++++++++++++++++- pumpkin/src/block/blocks/piston/mod.rs | 8 +- pumpkin/src/block/blocks/pumpkin.rs | 10 +- pumpkin/src/block/blocks/tnt.rs | 18 +- pumpkin/src/command/commands/summon.rs | 5 +- pumpkin/src/entity/experience_orb.rs | 9 +- pumpkin/src/entity/living.rs | 10 +- pumpkin/src/entity/mob/mod.rs | 27 +- pumpkin/src/entity/mod.rs | 63 ++++- pumpkin/src/entity/player.rs | 63 +++-- pumpkin/src/entity/type.rs | 37 +++ pumpkin/src/item/items/egg.rs | 10 +- pumpkin/src/item/items/hoe.rs | 10 +- pumpkin/src/item/items/snowball.rs | 10 +- pumpkin/src/net/packet/play.rs | 6 +- pumpkin/src/world/mod.rs | 202 ++++++++++++-- 26 files changed, 864 insertions(+), 175 deletions(-) create mode 100644 pumpkin/src/entity/type.rs diff --git a/README.md b/README.md index 018362dca..c5246bacf 100644 --- a/README.md +++ b/README.md @@ -83,10 +83,10 @@ and customizable experience. It prioritizes performance and player enjoyment whi - [x] Mobs (W.I.P) - [x] Animals (W.I.P) - [x] Entity AI (W.I.P) - - [ ] Entity Saving - [ ] Boss - [ ] Villagers - [ ] Mobs Inventory + - [X] Entity Saving - Server - [x] Plugins (W.I.P) - [x] Query diff --git a/pumpkin-nbt/src/compound.rs b/pumpkin-nbt/src/compound.rs index 0bd356202..9e389d9d8 100644 --- a/pumpkin-nbt/src/compound.rs +++ b/pumpkin-nbt/src/compound.rs @@ -112,7 +112,7 @@ impl NbtCompound { self.put(name, NbtTag::String(value)); } - pub fn put_list(&mut self, name: &str, value: Box<[NbtTag]>) { + pub fn put_list(&mut self, name: &str, value: Vec) { self.put(name, NbtTag::List(value)); } @@ -153,12 +153,10 @@ impl NbtCompound { #[inline] pub fn get(&self, name: &str) -> Option<&NbtTag> { - for (key, value) in &self.child_tags { - if key.as_str() == name { - return Some(value); - } - } - None + self.child_tags + .iter() + .find(|k| k.0.as_str() == name) + .map(|r| &r.1) } pub fn get_short(&self, name: &str) -> Option { diff --git a/pumpkin-nbt/src/deserializer.rs b/pumpkin-nbt/src/deserializer.rs index 7859ef3ec..10dda35a4 100644 --- a/pumpkin-nbt/src/deserializer.rs +++ b/pumpkin-nbt/src/deserializer.rs @@ -1,5 +1,8 @@ +use std::vec::IntoIter; + use crate::*; use io::Read; +use serde::de::value::SeqDeserializer; use serde::de::{self, DeserializeSeed, IntoDeserializer, MapAccess, SeqAccess, Visitor}; use serde::{Deserialize, forward_to_deserialize_any}; @@ -113,8 +116,6 @@ pub struct Deserializer { // Yes, this breaks with recursion. Just an attempt at a sanity check in_list: bool, is_named: bool, - // For debugging - key_stack: Vec, } impl Deserializer { @@ -124,7 +125,6 @@ impl Deserializer { tag_to_deserialize_stack: Vec::new(), in_list: false, is_named, - key_stack: Vec::new(), } } } @@ -181,28 +181,21 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer { END_ID => Err(Error::SerdeError( "Trying to deserialize an END tag!".to_string(), )), - LIST_ID | INT_ARRAY_ID | LONG_ARRAY_ID | BYTE_ARRAY_ID => { - let list_type = match tag_to_deserialize { - LIST_ID => self.input.get_u8_be()?, - INT_ARRAY_ID => INT_ID, - LONG_ARRAY_ID => LONG_ID, - BYTE_ARRAY_ID => BYTE_ID, - _ => unreachable!(), - }; + LIST_ID => { + let list_type = self.input.get_u8_be()?; let remaining_values = self.input.get_i32_be()?; if remaining_values < 0 { return Err(Error::NegativeLength(remaining_values)); } - let result = visitor.visit_seq(ListAccess { + visitor.visit_seq(ListAccess { de: self, list_type, remaining_values: remaining_values as usize, - })?; - Ok(result) + }) } - COMPOUND_ID => visitor.visit_map(CompoundAccess { de: self }), + COMPOUND_ID => self.deserialize_map(visitor), _ => { let result = match NbtTag::deserialize_data(&mut self.input, tag_to_deserialize)? { NbtTag::Byte(value) => visitor.visit_i8::(value)?, @@ -212,6 +205,22 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer { NbtTag::Float(value) => visitor.visit_f32::(value)?, NbtTag::Double(value) => visitor.visit_f64::(value)?, NbtTag::String(value) => visitor.visit_string::(value)?, + NbtTag::LongArray(value) => visitor + .visit_seq::, Error>>( + value.into_deserializer(), + )?, + NbtTag::IntArray(value) => visitor + .visit_seq::, Error>>( + value.into_deserializer(), + )?, + NbtTag::ByteArray(value) => { + // For compatibility, we serialize byte arrays as Vec + // It could be probably changed in the future + let array: Vec<_> = value.iter().map(|&byte| byte as i8).collect(); + visitor.visit_seq::, Error>>( + array.into_deserializer(), + )? + } _ => unreachable!(), }; Ok(result) @@ -268,21 +277,11 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer { if *tag_id == BYTE_ID { let value = self.input.get_u8_be()?; if value != 0 { - visitor.visit_bool(true) - } else { - visitor.visit_bool(false) + return visitor.visit_bool(true); } - } else { - Err(Error::UnsupportedType(format!( - "Non-byte bool (found type {tag_id})" - ))) } - } else { - Err(Error::SerdeError( - "Wanted to deserialize a bool, but there was no type hint on the stack!" - .to_string(), - )) } + visitor.visit_bool(false) } fn deserialize_enum( @@ -313,11 +312,7 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer { if let Some(tag_id) = self.tag_to_deserialize_stack.pop() { if tag_id != COMPOUND_ID { return Err(Error::SerdeError(format!( - "Trying to deserialize a map without a compound ID ({} with id {})", - self.key_stack - .last() - .cloned() - .unwrap_or_else(|| "compound root".to_string()), + "Trying to deserialize a map without a compound ID (with id {})", tag_id ))); } @@ -387,9 +382,7 @@ impl<'de, R: Read> MapAccess<'de> for CompoundAccess<'_, R> { where V: DeserializeSeed<'de>, { - let result = seed.deserialize(&mut *self.de); - self.de.key_stack.pop(); - result + seed.deserialize(&mut *self.de) } } @@ -405,7 +398,6 @@ impl<'de, R: Read> de::Deserializer<'de> for MapKey<'_, R> { V: de::Visitor<'de>, { let key = get_nbt_string(&mut self.de.input)?; - self.de.key_stack.push(key.clone()); visitor.visit_string(key) } diff --git a/pumpkin-nbt/src/serializer.rs b/pumpkin-nbt/src/serializer.rs index b571a1f16..9257635b0 100644 --- a/pumpkin-nbt/src/serializer.rs +++ b/pumpkin-nbt/src/serializer.rs @@ -370,7 +370,9 @@ impl ser::Serializer for &mut Serializer { } else { return Err(Error::UnsupportedType("newtype variant".to_string())); } - value.serialize(self) + value.serialize(self)?; + + Ok(()) } fn serialize_seq(self, len: Option) -> Result { diff --git a/pumpkin-nbt/src/tag.rs b/pumpkin-nbt/src/tag.rs index e25a80649..ed6e6de2f 100644 --- a/pumpkin-nbt/src/tag.rs +++ b/pumpkin-nbt/src/tag.rs @@ -18,10 +18,10 @@ pub enum NbtTag { Double(f64) = DOUBLE_ID, ByteArray(Box<[u8]>) = BYTE_ARRAY_ID, String(String) = STRING_ID, - List(Box<[NbtTag]>) = LIST_ID, + List(Vec) = LIST_ID, Compound(NbtCompound) = COMPOUND_ID, - IntArray(Box<[i32]>) = INT_ARRAY_ID, - LongArray(Box<[i64]>) = LONG_ARRAY_ID, + IntArray(Vec) = INT_ARRAY_ID, + LongArray(Vec) = LONG_ARRAY_ID, } impl NbtTag { @@ -232,7 +232,7 @@ impl NbtTag { assert_eq!(tag.get_type_id(), tag_type_id); list.push(tag); } - Ok(NbtTag::List(list.into_boxed_slice())) + Ok(NbtTag::List(list)) } COMPOUND_ID => Ok(NbtTag::Compound(NbtCompound::deserialize_content(reader)?)), INT_ARRAY_ID => { @@ -247,7 +247,7 @@ impl NbtTag { let int = reader.get_i32_be()?; int_array.push(int); } - Ok(NbtTag::IntArray(int_array.into_boxed_slice())) + Ok(NbtTag::IntArray(int_array)) } LONG_ARRAY_ID => { let len = reader.get_i32_be()?; @@ -261,7 +261,7 @@ impl NbtTag { let long = reader.get_i64_be()?; long_array.push(long); } - Ok(NbtTag::LongArray(long_array.into_boxed_slice())) + Ok(NbtTag::LongArray(long_array)) } _ => Err(Error::UnknownTagId(tag_id)), } @@ -500,7 +500,7 @@ impl<'de> Deserialize<'de> for NbtTag { while let Some(value) = seq.next_element()? { vec.push(value); } - Ok(NbtTag::List(vec.into_boxed_slice())) + Ok(NbtTag::List(vec)) } fn visit_map(self, map: A) -> Result diff --git a/pumpkin-world/src/chunk/format/anvil.rs b/pumpkin-world/src/chunk/format/anvil.rs index 99b107541..a41c69cf3 100644 --- a/pumpkin-world/src/chunk/format/anvil.rs +++ b/pumpkin-world/src/chunk/format/anvil.rs @@ -904,6 +904,7 @@ mod tests { &LevelFolder { root_folder: PathBuf::from(""), region_folder: region_path, + entities_folder: PathBuf::from(""), }, &[Vector2::new(0, 0)], send, @@ -932,6 +933,7 @@ mod tests { let level_folder = LevelFolder { root_folder: temp_dir.path().to_path_buf(), region_folder: temp_dir.path().join("region"), + entities_folder: PathBuf::from("entities"), }; fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder"); let chunk_saver = ChunkFileManager::>::default(); @@ -1209,6 +1211,7 @@ mod tests { let level_folder = LevelFolder { root_folder: temp_dir.path().to_path_buf(), region_folder: temp_dir.path().join("region"), + entities_folder: PathBuf::from("entities"), }; fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder"); let chunk_saver = ChunkFileManager::>::default(); diff --git a/pumpkin-world/src/chunk/format/linear.rs b/pumpkin-world/src/chunk/format/linear.rs index 0d1b19044..88ea82f69 100644 --- a/pumpkin-world/src/chunk/format/linear.rs +++ b/pumpkin-world/src/chunk/format/linear.rs @@ -5,7 +5,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use crate::chunk::format::anvil::{AnvilChunkFile, SingleChunkDataSerializer}; use crate::chunk::io::{ChunkSerializer, LoadedData}; -use crate::chunk::{ChunkData, ChunkReadingError, ChunkWritingError}; +use crate::chunk::{ChunkReadingError, ChunkWritingError}; use async_trait::async_trait; use bytes::{Buf, BufMut, Bytes}; use log::error; @@ -164,7 +164,7 @@ impl Default for LinearFile { #[async_trait] impl ChunkSerializer for LinearFile { - type Data = ChunkData; + type Data = S; type WriteBackend = PathBuf; fn should_write(&self, is_watched: bool) -> bool { @@ -320,7 +320,7 @@ impl ChunkSerializer for LinearFile { }) } - async fn update_chunk(&mut self, chunk: &ChunkData) -> Result<(), ChunkWritingError> { + async fn update_chunk(&mut self, chunk: &Self::Data) -> Result<(), ChunkWritingError> { let index = LinearFile::::get_chunk_index(chunk.position()); let chunk_raw: Bytes = chunk .to_bytes() @@ -343,7 +343,7 @@ impl ChunkSerializer for LinearFile { async fn get_chunks( &self, chunks: &[Vector2], - stream: tokio::sync::mpsc::Sender>, + stream: tokio::sync::mpsc::Sender>, ) { // Don't par iter here so we can prevent backpressure with the await in the async // runtime @@ -352,9 +352,7 @@ impl ChunkSerializer for LinearFile { let linear_chunk_data = &self.chunks_data[index]; let result = if let Some(data) = linear_chunk_data { - match ChunkData::internal_from_bytes(data, chunk) - .map_err(ChunkReadingError::ParsingError) - { + match S::from_bytes(data.clone(), chunk) { Ok(chunk) => LoadedData::Loaded(chunk), Err(err) => LoadedData::Error((chunk, err)), } @@ -420,6 +418,7 @@ mod tests { &LevelFolder { root_folder: PathBuf::from(""), region_folder: region_path, + entities_folder: PathBuf::from(""), }, &[Vector2::new(0, 0)], send, @@ -443,6 +442,7 @@ mod tests { let level_folder = LevelFolder { root_folder: temp_dir.path().to_path_buf(), region_folder: temp_dir.path().join("region"), + entities_folder: PathBuf::from("entities"), }; fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder"); let chunk_saver = ChunkFileManager::>::default(); diff --git a/pumpkin-world/src/chunk/format/mod.rs b/pumpkin-world/src/chunk/format/mod.rs index eb2d99058..3fb1fb625 100644 --- a/pumpkin-world/src/chunk/format/mod.rs +++ b/pumpkin-world/src/chunk/format/mod.rs @@ -5,11 +5,12 @@ use bytes::Bytes; use futures::future::join_all; use pumpkin_data::{Block, chunk::ChunkStatus}; use pumpkin_nbt::{compound::NbtCompound, from_bytes, nbt_long_array}; +use uuid::Uuid; use crate::{ block::entities::block_entity_from_nbt, chunk::{ - ChunkReadingError, ChunkSerializingError, + ChunkEntityData, ChunkReadingError, ChunkSerializingError, format::anvil::{SingleChunkDataSerializer, WORLD_DATA_VERSION}, io::{Dirtiable, file_manager::PathFromLevelFolder}, }, @@ -316,6 +317,103 @@ impl ChunkData { } } +impl PathFromLevelFolder for ChunkEntityData { + #[inline] + fn file_path(folder: &LevelFolder, file_name: &str) -> PathBuf { + folder.entities_folder.join(file_name) + } +} + +impl Dirtiable for ChunkEntityData { + #[inline] + fn mark_dirty(&mut self, flag: bool) { + self.dirty = flag; + } + + #[inline] + fn is_dirty(&self) -> bool { + self.dirty + } +} + +#[async_trait] +impl SingleChunkDataSerializer for ChunkEntityData { + #[inline] + fn from_bytes(bytes: Bytes, pos: Vector2) -> Result { + Self::internal_from_bytes(&bytes, pos).map_err(ChunkReadingError::ParsingError) + } + + #[inline] + async fn to_bytes(&self) -> Result { + self.internal_to_bytes() + } + + #[inline] + fn position(&self) -> &Vector2 { + &self.chunk_position + } +} + +impl ChunkEntityData { + fn internal_from_bytes( + chunk_data: &[u8], + position: Vector2, + ) -> Result { + let chunk_entity_data = pumpkin_nbt::from_bytes::(chunk_data) + .map_err(|e| ChunkParsingError::ErrorDeserializingChunk(e.to_string()))?; + + if chunk_entity_data.position[0] != position.x + || chunk_entity_data.position[1] != position.z + { + return Err(ChunkParsingError::ErrorDeserializingChunk(format!( + "Expected data for entity chunk {},{} but got it for {},{}!", + position.x, + position.z, + chunk_entity_data.position[0], + chunk_entity_data.position[1], + ))); + } + let mut map = HashMap::new(); + for entity_nbt in chunk_entity_data.entities { + // TODO: This is wrong, we should use an int array, but our NBT lib for some reason does not work with int arrays and + // Just gives me a list when putting in a int array + let uuid = match entity_nbt.get_list("UUID") { + Some(uuid) => uuid, + None => { + log::warn!("TODO: use int arrays for UUID"); + continue; + } + }; + let uuid = Uuid::from_u128( + (uuid.first().unwrap().extract_int().unwrap() as u128) << 96 + | (uuid.get(1).unwrap().extract_int().unwrap() as u128) << 64 + | (uuid.get(2).unwrap().extract_int().unwrap() as u128) << 32 + | (uuid.get(3).unwrap().extract_int().unwrap() as u128), + ); + map.insert(uuid, entity_nbt); + } + + Ok(ChunkEntityData { + chunk_position: position, + data: map, + dirty: false, + }) + } + + fn internal_to_bytes(&self) -> Result { + let nbt = EntityNbt { + data_version: WORLD_DATA_VERSION, + position: [self.chunk_position.x, self.chunk_position.z], + entities: self.data.values().cloned().collect(), + }; + + let mut result = Vec::new(); + pumpkin_nbt::to_bytes(&nbt, &mut result) + .map_err(ChunkSerializingError::ErrorSerializingChunk)?; + Ok(result.into()) + } +} + #[derive(Serialize, Deserialize, Debug)] struct ChunkSectionNBT { #[serde(skip_serializing_if = "Option::is_none")] @@ -471,3 +569,11 @@ struct ChunkNbt { #[serde(rename = "isLightOn")] light_correct: bool, } + +#[derive(Serialize, Deserialize, Debug)] +#[serde(rename_all = "PascalCase")] +struct EntityNbt { + data_version: i32, + position: [i32; 2], + entities: Vec, +} diff --git a/pumpkin-world/src/chunk/mod.rs b/pumpkin-world/src/chunk/mod.rs index 2ba27e085..bfd57427e 100644 --- a/pumpkin-world/src/chunk/mod.rs +++ b/pumpkin-world/src/chunk/mod.rs @@ -122,6 +122,13 @@ pub struct ChunkData { pub dirty: bool, } +pub struct ChunkEntityData { + pub chunk_position: Vector2, + pub data: HashMap, + + pub dirty: bool, +} + /// Represents pure block data for a chunk. /// Subchunks are vertical portions of a chunk. They are 16 blocks tall. /// There are currently 24 subchunks per chunk. diff --git a/pumpkin-world/src/inventory/inventory.rs b/pumpkin-world/src/inventory/inventory.rs index 73f5b410e..183145de2 100644 --- a/pumpkin-world/src/inventory/inventory.rs +++ b/pumpkin-world/src/inventory/inventory.rs @@ -53,7 +53,7 @@ pub trait Inventory: Send + Sync + Debug + Clearable { return; } - nbt.put("Items", NbtTag::List(slots.into_boxed_slice())); + nbt.put("Items", NbtTag::List(slots)); } fn read_data( diff --git a/pumpkin-world/src/level.rs b/pumpkin-world/src/level.rs index bc69fb5dc..0d26a921f 100644 --- a/pumpkin-world/src/level.rs +++ b/pumpkin-world/src/level.rs @@ -5,7 +5,7 @@ use pumpkin_config::{advanced_config, chunk::ChunkFormat}; use pumpkin_data::Block; use pumpkin_util::math::{position::BlockPos, vector2::Vector2}; use std::{ - collections::VecDeque, + collections::{HashMap, VecDeque}, path::PathBuf, sync::{ Arc, @@ -26,9 +26,10 @@ use crate::{ BlockStateId, block::RawBlockState, chunk::{ - ChunkData, ChunkParsingError, ChunkReadingError, ScheduledTick, TickPriority, + ChunkData, ChunkEntityData, ChunkParsingError, ChunkReadingError, ScheduledTick, + TickPriority, format::{anvil::AnvilChunkFile, linear::LinearFile}, - io::{FileIO, LoadedData, file_manager::ChunkFileManager}, + io::{Dirtiable, FileIO, LoadedData, file_manager::ChunkFileManager}, }, dimension::Dimension, generation::{Seed, get_world_gen, implementation::WorldGenerator}, @@ -36,6 +37,7 @@ use crate::{ }; pub type SyncChunk = Arc>; +pub type SyncEntityChunk = Arc>; /// The `Level` module provides functionality for working with chunks within or outside a Minecraft world. /// @@ -57,9 +59,13 @@ pub struct Level { // 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>>, + loaded_entity_chunks: Arc, SyncEntityChunk>>, + chunk_watchers: Arc, usize>>, chunk_saver: Arc>, + entity_saver: Arc>, + world_gen: Arc, block_ticks: Arc>>, @@ -75,6 +81,7 @@ pub struct Level { pub struct LevelFolder { pub root_folder: PathBuf, pub region_folder: PathBuf, + pub entities_folder: PathBuf, } impl Level { @@ -89,9 +96,14 @@ impl Level { if !region_folder.exists() { std::fs::create_dir_all(®ion_folder).expect("Failed to create Region folder"); } + let entities_folder = root_folder.join("entities"); + if !entities_folder.exists() { + std::fs::create_dir_all(®ion_folder).expect("Failed to create Entities folder"); + } let level_folder = LevelFolder { root_folder, region_folder, + entities_folder, }; // TODO: Load info correctly based on world format type @@ -105,6 +117,15 @@ impl Level { Arc::new(ChunkFileManager::>::default()) } }; + let entity_saver: Arc> = + match advanced_config().chunk.format { + ChunkFormat::Linear => { + Arc::new(ChunkFileManager::>::default()) + } + ChunkFormat::Anvil => { + Arc::new(ChunkFileManager::>::default()) + } + }; Self { seed, @@ -112,8 +133,10 @@ impl Level { world_gen, level_folder, chunk_saver, + entity_saver, spawn_chunks: Arc::new(DashMap::new()), loaded_chunks: Arc::new(DashMap::new()), + loaded_entity_chunks: Arc::new(DashMap::new()), chunk_watchers: Arc::new(DashMap::new()), tasks: TaskTracker::new(), shutdown_notifier: Notify::new(), @@ -140,7 +163,7 @@ impl Level { self.tasks.close(); log::debug!("Awaiting level tasks"); self.tasks.wait().await; - log::debug!("Done awaiting level tasks"); + log::debug!("Done awaiting level chunk tasks"); // wait for chunks currently saving in other threads self.chunk_saver.block_and_await_ongoing_tasks().await; @@ -156,6 +179,23 @@ impl Level { // TODO: I think the chunk_saver should be at the server level self.chunk_saver.clear_watched_chunks().await; self.write_chunks(chunks_to_write).await; + + log::debug!("Done awaiting level entity tasks"); + + // wait for chunks currently saving in other threads + self.entity_saver.block_and_await_ongoing_tasks().await; + + // save all chunks currently in memory + let chunks_to_write = self + .loaded_entity_chunks + .iter() + .map(|chunk| (*chunk.key(), chunk.value().clone())) + .collect::>(); + self.loaded_entity_chunks.clear(); + + // TODO: I think the chunk_saver should be at the server level + self.entity_saver.clear_watched_chunks().await; + self.write_entity_chunks(chunks_to_write).await; } pub fn loaded_chunk_count(&self) -> usize { @@ -164,6 +204,7 @@ impl Level { pub async fn clean_up_log(&self) { self.chunk_saver.clean_up_log().await; + self.entity_saver.clean_up_log().await; } pub fn list_cached(&self) { @@ -197,6 +238,9 @@ impl Level { self.chunk_saver .watch_chunks(&self.level_folder, chunks) .await; + self.entity_saver + .watch_chunks(&self.level_folder, chunks) + .await; } #[inline] @@ -233,6 +277,9 @@ impl Level { self.chunk_saver .unwatch_chunks(&self.level_folder, chunks) .await; + self.entity_saver + .unwatch_chunks(&self.level_folder, chunks) + .await; chunks_to_clean } @@ -286,6 +333,50 @@ impl Level { }); } + pub async fn clean_entity_chunks(self: &Arc, chunks: &[Vector2]) { + // Care needs to be take here because of interweaving case: + // 1) Remove chunk from cache + // 2) Another player wants same chunk + // 3) Load (old) chunk from serializer + // 4) Write (new) chunk from serializer + // Now outdated chunk data is cached and will be written later + + let chunks_with_no_watchers = chunks + .iter() + .filter_map(|pos| { + // Only chunks that have no entry in the watcher map or have 0 watchers + if self + .chunk_watchers + .get(pos) + .is_none_or(|count| count.is_zero()) + { + self.loaded_entity_chunks + .get(pos) + .map(|chunk| (*pos, chunk.value().clone())) + } else { + None + } + }) + .collect::>(); + + let level = self.clone(); + self.spawn_task(async move { + let chunks_to_remove = chunks_with_no_watchers.clone(); + level.write_entity_chunks(chunks_with_no_watchers).await; + // Only after we have written the chunks to the serializer do we remove them from the + // cache + for (pos, _) in chunks_to_remove { + let _ = level.loaded_entity_chunks.remove_if(&pos, |_, _| { + // Recheck that there is no one watching + level + .chunk_watchers + .get(&pos) + .is_none_or(|count| count.is_zero()) + }); + } + }); + } + pub async fn tick_block_entities(&self, world: Arc) { for chunk in self.loaded_chunks.iter() { let chunk = chunk.read().await; @@ -301,6 +392,10 @@ impl Level { self.clean_chunks(&[*chunk]).await; } + pub async fn clean_entity_chunk(self: &Arc, chunk: &Vector2) { + self.clean_entity_chunks(&[*chunk]).await; + } + pub fn is_chunk_watched(&self, chunk: &Vector2) -> bool { self.chunk_watchers.get(chunk).is_some() } @@ -309,6 +404,8 @@ impl Level { self.chunk_watchers.retain(|_, watcher| !watcher.is_zero()); self.loaded_chunks .retain(|at, _| self.chunk_watchers.get(at).is_some()); + self.loaded_entity_chunks + .retain(|at, _| self.chunk_watchers.get(at).is_some()); // if the difference is too big, we can shrink the loaded chunks // (1024 chunks is the equivalent to a 32x32 chunks area) @@ -321,6 +418,10 @@ impl Level { if self.loaded_chunks.capacity() - self.loaded_chunks.len() >= 4096 { self.loaded_chunks.shrink_to_fit(); } + + if self.loaded_entity_chunks.capacity() - self.loaded_entity_chunks.len() >= 4096 { + self.loaded_entity_chunks.shrink_to_fit(); + } } // Stream the chunks (don't collect them and then do stuff with them) @@ -348,6 +449,27 @@ impl Level { receiver } + pub fn receive_entity_chunks( + self: &Arc, + chunks: Vec>, + ) -> UnboundedReceiver<(SyncEntityChunk, bool)> { + let (sender, receiver) = mpsc::unbounded_channel(); + // Put this in another thread so we aren't blocking on it + let level = self.clone(); + self.spawn_task(async move { + let cancel_notifier = level.shutdown_notifier.notified(); + let fetch_task = level.fetch_entity_chunks(&chunks, sender); + + // Don't continue to handle chunks if we are shutting down + select! { + () = cancel_notifier => {}, + () = fetch_task => {} + }; + }); + + receiver + } + pub async fn get_chunk( self: &Arc, chunk_coordinate: Vector2, @@ -358,6 +480,16 @@ impl Level { } } + pub async fn get_entity_chunk( + self: &Arc, + chunk_coordinate: Vector2, + ) -> Arc> { + match self.try_get_entity_chunk(chunk_coordinate) { + Some(chunk) => chunk.clone(), + None => self.receive_entity_chunk(chunk_coordinate).await.0, + } + } + pub async fn receive_chunk( self: &Arc, chunk_pos: Vector2, @@ -370,6 +502,18 @@ impl Level { .expect("Channel closed for unknown reason") } + pub async fn receive_entity_chunk( + self: &Arc, + chunk_pos: Vector2, + ) -> (Arc>, bool) { + let mut receiver = self.receive_entity_chunks(vec![chunk_pos]); + + receiver + .recv() + .await + .expect("Channel closed for unknown reason") + } + pub async fn get_block_state(self: &Arc, position: &BlockPos) -> RawBlockState { let (chunk_coordinate, relative) = position.chunk_and_chunk_relative_position(); let chunk = self.get_chunk(chunk_coordinate).await; @@ -404,7 +548,7 @@ impl Level { return block_state_id; } - chunk.dirty = true; + chunk.mark_dirty(true); chunk.section.set_block_absolute_y( relative.x as usize, @@ -463,6 +607,23 @@ impl Level { } } + pub async fn write_entity_chunks(&self, chunks_to_write: Vec<(Vector2, SyncEntityChunk)>) { + if chunks_to_write.is_empty() { + return; + } + + let chunk_saver = self.entity_saver.clone(); + let level_folder = self.level_folder.clone(); + + trace!("Sending chunks to ChunkIO {:}", chunks_to_write.len()); + if let Err(error) = chunk_saver + .save_chunks(&level_folder, chunks_to_write) + .await + { + log::error!("Failed writing Chunk to disk {error}"); + } + } + /// Initializes the spawn chunks to these chunks pub async fn read_spawn_chunks(self: &Arc, chunks: &[Vector2]) { let (send, mut recv) = mpsc::unbounded_channel(); @@ -647,6 +808,149 @@ impl Level { tracker.wait().await; } + pub async fn fetch_entity_chunks( + self: &Arc, + chunks: &[Vector2], + channel: mpsc::UnboundedSender<(SyncEntityChunk, bool)>, + ) { + if chunks.is_empty() { + return; + } + + // If false, stop loading chunks because the channel has closed. + let send_chunk = + move |is_new: bool, + chunk: SyncEntityChunk, + channel: &mpsc::UnboundedSender<(SyncEntityChunk, bool)>| { + channel.send((chunk, is_new)).is_ok() + }; + + // First send all chunks that we have cached + // We expect best case scenario to have all cached + let mut remaining_chunks = Vec::new(); + for chunk in chunks { + let is_ok = if let Some(chunk) = self.loaded_entity_chunks.get(chunk) { + send_chunk(false, chunk.value().clone(), &channel) + } else { + remaining_chunks.push(*chunk); + true + }; + + if !is_ok { + return; + } + } + + if remaining_chunks.is_empty() { + return; + } + + // These just pass data between async tasks, each of which do not block on anything, so + // these do not need to hold a lot + let (load_bridge_send, mut load_bridge_recv) = + tokio::sync::mpsc::channel::>(16); + let (generate_bridge_send, mut generate_bridge_recv) = tokio::sync::mpsc::channel(16); + + let load_channel = channel.clone(); + let loaded_chunks = self.loaded_entity_chunks.clone(); + let handle_load = async move { + while let Some(data) = load_bridge_recv.recv().await { + let is_ok = match data { + LoadedData::Loaded(chunk) => { + let position = chunk.read().await.chunk_position; + + let value = loaded_chunks + .entry(position) + .or_insert(chunk) + .value() + .clone(); + send_chunk(false, value, &load_channel) + } + LoadedData::Missing(pos) => generate_bridge_send.send(pos).await.is_ok(), + LoadedData::Error((pos, error)) => { + match error { + // this is expected, and is not an error + ChunkReadingError::ChunkNotExist + | ChunkReadingError::ParsingError( + ChunkParsingError::ChunkNotGenerated, + ) => {} + // this is an error, and we should log it + error => { + log::error!( + "Failed to load chunk at {pos:?}: {error} (regenerating)" + ); + } + }; + + generate_bridge_send.send(pos).await.is_ok() + } + }; + + if !is_ok { + // This isn't recoverable, so stop listening + return; + } + } + }; + + let loaded_chunks = self.loaded_entity_chunks.clone(); + let handle_generate = async move { + let continue_to_generate = Arc::new(AtomicBool::new(true)); + while let Some(pos) = generate_bridge_recv.recv().await { + if !continue_to_generate.load(Ordering::Relaxed) { + return; + } + + let loaded_chunks = loaded_chunks.clone(); + let channel = channel.clone(); + let cloned_continue_to_generate = continue_to_generate.clone(); + + tokio::spawn(async move { + // Rayon tasks are queued, so also check it here + if !cloned_continue_to_generate.load(Ordering::Relaxed) { + return; + } + + let result = { + let entry = loaded_chunks.entry(pos); // Get the entry for the position + + // Check if the entry already exists. + // If not, generate the chunk asynchronously and insert it. + match entry { + Entry::Occupied(entry) => entry.into_ref(), + Entry::Vacant(entry) => { + let generated_chunk = ChunkEntityData { + chunk_position: pos, + data: HashMap::new(), + dirty: true, + }; + entry.insert(Arc::new(RwLock::new(generated_chunk))) + } + } + .value() + .clone() + }; + + if !send_chunk(true, result, &channel) { + // Stop any additional queued generations + cloned_continue_to_generate.store(false, Ordering::Relaxed); + } + }); + } + }; + + let tracker = TaskTracker::new(); + tracker.spawn(handle_load); + tracker.spawn(handle_generate); + + self.entity_saver + .fetch_chunks(&self.level_folder, &remaining_chunks, load_bridge_send) + .await; + + tracker.close(); + tracker.wait().await; + } + pub fn try_get_chunk( &self, coordinates: Vector2, @@ -654,6 +958,13 @@ impl Level { self.loaded_chunks.try_get(&coordinates).try_unwrap() } + pub fn try_get_entity_chunk( + &self, + coordinates: Vector2, + ) -> Option, Arc>>> { + self.loaded_entity_chunks.try_get(&coordinates).try_unwrap() + } + pub async fn get_and_tick_block_ticks(&self) -> Arc>> { let mut block_ticks = self.block_ticks.lock().await; let mut ticks = VecDeque::new(); diff --git a/pumpkin/src/block/blocks/piston/mod.rs b/pumpkin/src/block/blocks/piston/mod.rs index 67bd1b635..0b38aceb4 100644 --- a/pumpkin/src/block/blocks/piston/mod.rs +++ b/pumpkin/src/block/blocks/piston/mod.rs @@ -50,13 +50,7 @@ impl<'a> PistonHandler<'a> { self.moved_blocks.clear(); self.broken_blocks.clear(); let (block, block_state) = self.world.get_block_and_block_state(&self.pos_to).await; - dbg!(PistonBlock::is_movable( - &block, - &block_state, - self.motion_direction, - false, - self.piston_direction, - )); + if !PistonBlock::is_movable( &block, &block_state, diff --git a/pumpkin/src/block/blocks/pumpkin.rs b/pumpkin/src/block/blocks/pumpkin.rs index dfb98b45f..d5d0c8642 100644 --- a/pumpkin/src/block/blocks/pumpkin.rs +++ b/pumpkin/src/block/blocks/pumpkin.rs @@ -1,3 +1,4 @@ +use crate::entity::Entity; use crate::entity::item::ItemEntity; use crate::server::Server; use crate::world::World; @@ -11,6 +12,7 @@ use pumpkin_util::math::position::BlockPos; use pumpkin_world::item::ItemStack; use pumpkin_world::world::BlockFlags; use std::sync::Arc; +use uuid::Uuid; #[pumpkin_block("minecraft:pumpkin")] pub struct PumpkinBlock; @@ -36,7 +38,13 @@ impl crate::block::pumpkin_block::PumpkinBlock for PumpkinBlock { BlockFlags::NOTIFY_ALL, ) .await; - let entity = world.create_entity(pos.to_f64(), EntityType::ITEM); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + pos.to_f64(), + EntityType::ITEM, + false, + ); let item_entity = Arc::new(ItemEntity::new(entity, ItemStack::new(4, &Item::PUMPKIN_SEEDS)).await); world.spawn_entity(item_entity).await; diff --git a/pumpkin/src/block/blocks/tnt.rs b/pumpkin/src/block/blocks/tnt.rs index fb3fa54ce..83fdc890a 100644 --- a/pumpkin/src/block/blocks/tnt.rs +++ b/pumpkin/src/block/blocks/tnt.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use crate::block::pumpkin_block::PumpkinBlock; use crate::block::registry::BlockActionResult; +use crate::entity::Entity; use crate::entity::player::Player; use crate::entity::tnt::TNTEntity; use crate::server::Server; @@ -17,6 +18,7 @@ use pumpkin_util::math::vector3::Vector3; use pumpkin_world::BlockStateId; use pumpkin_world::world::BlockFlags; use rand::Rng; +use uuid::Uuid; use super::redstone::block_receives_redstone_power; @@ -25,7 +27,13 @@ pub struct TNTBlock; impl TNTBlock { pub async fn prime(world: &Arc, location: &BlockPos) { - let entity = world.create_entity(location.to_f64(), EntityType::TNT); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + location.to_f64(), + EntityType::TNT, + false, + ); let pos = entity.pos.load(); let tnt = Arc::new(TNTEntity::new(entity, DEFAULT_POWER, DEFAULT_FUSE)); world.spawn_entity(tnt).await; @@ -93,7 +101,13 @@ impl PumpkinBlock for TNTBlock { } async fn explode(&self, _block: &Block, world: &Arc, location: BlockPos) { - let entity = world.create_entity(location.to_f64(), EntityType::TNT); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + location.to_f64(), + EntityType::TNT, + false, + ); let angle = rand::random::() * std::f64::consts::TAU; entity .set_velocity(Vector3::new(-angle.sin() * 0.02, 0.2, -angle.cos() * 0.02)) diff --git a/pumpkin/src/command/commands/summon.rs b/pumpkin/src/command/commands/summon.rs index 92cb9a342..efdb5f3a9 100644 --- a/pumpkin/src/command/commands/summon.rs +++ b/pumpkin/src/command/commands/summon.rs @@ -1,5 +1,6 @@ use async_trait::async_trait; use pumpkin_util::{math::vector3::Vector3, text::TextComponent}; +use uuid::Uuid; use crate::{ command::{ @@ -10,7 +11,7 @@ use crate::{ }, tree::{CommandTree, builder::argument}, }, - entity::mob, + entity::r#type::from_type, }; const NAMES: [&str; 1] = ["summon"]; @@ -57,7 +58,7 @@ impl CommandExecutor for Executor { (player.world().await, pos) } }; - let mob = mob::from_type(entity, pos, &world); + let mob = from_type(entity, pos, &world, Uuid::new_v4()); world.spawn_entity(mob).await; sender diff --git a/pumpkin/src/entity/experience_orb.rs b/pumpkin/src/entity/experience_orb.rs index 792f5c56e..c15edc98e 100644 --- a/pumpkin/src/entity/experience_orb.rs +++ b/pumpkin/src/entity/experience_orb.rs @@ -3,6 +3,7 @@ use std::sync::{Arc, atomic::AtomicU32}; use async_trait::async_trait; use pumpkin_data::{damage::DamageType, entity::EntityType}; use pumpkin_util::math::vector3::Vector3; +use uuid::Uuid; use crate::{server::Server, world::World}; @@ -29,7 +30,13 @@ impl ExperienceOrbEntity { while amount > 0 { let i = Self::round_to_orb_size(amount); amount -= i; - let entity = world.create_entity(position, EntityType::EXPERIENCE_ORB); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + position, + EntityType::EXPERIENCE_ORB, + false, + ); let orb = Arc::new(Self::new(entity, i)); world.spawn_entity(orb).await; } diff --git a/pumpkin/src/entity/living.rs b/pumpkin/src/entity/living.rs index 7915ca069..11621708e 100644 --- a/pumpkin/src/entity/living.rs +++ b/pumpkin/src/entity/living.rs @@ -371,10 +371,7 @@ impl EntityBase for LivingEntity { fn get_living_entity(&self) -> Option<&LivingEntity> { Some(self) } -} -#[async_trait] -impl NBTStorage for LivingEntity { async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) { self.entity.write_nbt(nbt).await; nbt.put("Health", NbtTag::Float(self.health.load())); @@ -389,17 +386,14 @@ impl NBTStorage for LivingEntity { effect.write_nbt(&mut effect_nbt).await; effects_list.push(NbtTag::Compound(effect_nbt)); } - nbt.put( - "active_effects", - NbtTag::List(effects_list.into_boxed_slice()), - ); + nbt.put("active_effects", NbtTag::List(effects_list)); } } //TODO: write equipment // todo more... } - async fn read_nbt(&mut self, nbt: &mut pumpkin_nbt::compound::NbtCompound) { + async fn read_nbt(&self, nbt: &pumpkin_nbt::compound::NbtCompound) { self.entity.read_nbt(nbt).await; self.health.store(nbt.get_float("Health").unwrap_or(0.0)); self.fall_distance diff --git a/pumpkin/src/entity/mob/mod.rs b/pumpkin/src/entity/mob/mod.rs index 6a1b3978e..462a2ad0f 100644 --- a/pumpkin/src/entity/mob/mod.rs +++ b/pumpkin/src/entity/mob/mod.rs @@ -1,12 +1,9 @@ use std::sync::Arc; use async_trait::async_trait; -use pumpkin_data::entity::EntityType; -use pumpkin_util::math::vector3::Vector3; use tokio::sync::Mutex; -use zombie::Zombie; -use crate::{server::Server, world::World}; +use crate::server::Server; use super::{ Entity, EntityBase, @@ -50,25 +47,3 @@ impl EntityBase for MobEntity { Some(&self.living_entity) } } - -pub fn from_type( - entity_type: EntityType, - position: Vector3, - world: &Arc, -) -> Arc { - let entity = world.create_entity(position, entity_type); - - #[allow(clippy::single_match)] - let mob = match entity_type { - EntityType::ZOMBIE => Zombie::make(entity), - // TODO - _ => MobEntity { - living_entity: LivingEntity::new(entity), - goals: Mutex::new(vec![]), - navigator: Mutex::new(Navigator::default()), - }, - }; - Arc::new(mob) -} - -impl MobEntity {} diff --git a/pumpkin/src/entity/mod.rs b/pumpkin/src/entity/mod.rs index f87b91c02..248ffb530 100644 --- a/pumpkin/src/entity/mod.rs +++ b/pumpkin/src/entity/mod.rs @@ -51,6 +51,7 @@ pub mod mob; pub mod player; pub mod projectile; pub mod tnt; +pub mod r#type; mod combat; @@ -73,6 +74,22 @@ pub trait EntityBase: Send + Sync { } } + async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) { + if let Some(living) = self.get_living_entity() { + living.write_nbt(nbt).await; + } else { + self.get_entity().write_nbt(nbt).await; + } + } + + async fn read_nbt(&self, nbt: &pumpkin_nbt::compound::NbtCompound) { + if let Some(living) = self.get_living_entity() { + living.read_nbt(nbt).await; + } else { + self.get_entity().read_nbt(nbt).await; + } + } + async fn init_data_tracker(&self) {} async fn teleport( @@ -151,6 +168,8 @@ pub struct Entity { pub fire_ticks: AtomicI32, pub has_visual_fire: AtomicBool, + pub first_loaded_chunk_position: AtomicCell>>, + pub portal_cooldown: AtomicU32, pub portal_manager: Mutex>>, @@ -191,6 +210,7 @@ impl Entity { velocity: AtomicCell::new(Vector3::new(0.0, 0.0, 0.0)), standing_eye_height: entity_type.eye_height, pose: AtomicCell::new(EntityPose::Standing), + first_loaded_chunk_position: AtomicCell::new(None), bounding_box: AtomicCell::new(BoundingBox::new_from_pos( position.x, position.y, @@ -788,28 +808,43 @@ impl EntityBase for Entity { fn get_living_entity(&self) -> Option<&LivingEntity> { None } -} -#[async_trait] -impl NBTStorage for Entity { async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) { let position = self.pos.load(); + nbt.put_string( + "id", + format!("minecraft:{}", self.entity_type.resource_name), + ); + let uuid = self.entity_uuid.as_u128(); + nbt.put( + "UUID", + NbtTag::IntArray(vec![ + (uuid >> 96) as i32, + ((uuid >> 64) & 0xFFFF_FFFF) as i32, + ((uuid >> 32) & 0xFFFF_FFFF) as i32, + (uuid & 0xFFFF_FFFF) as i32, + ]), + ); nbt.put( "Pos", - NbtTag::List( - vec![position.x.into(), position.y.into(), position.z.into()].into_boxed_slice(), - ), + NbtTag::List(vec![ + position.x.into(), + position.y.into(), + position.z.into(), + ]), ); let velocity = self.velocity.load(); nbt.put( "Motion", - NbtTag::List( - vec![velocity.x.into(), velocity.y.into(), velocity.z.into()].into_boxed_slice(), - ), + NbtTag::List(vec![ + velocity.x.into(), + velocity.y.into(), + velocity.z.into(), + ]), ); nbt.put( "Rotation", - NbtTag::List(vec![self.yaw.load().into(), self.pitch.load().into()].into_boxed_slice()), + NbtTag::List(vec![self.yaw.load().into(), self.pitch.load().into()]), ); nbt.put_short("Fire", self.fire_ticks.load(Relaxed) as i16); nbt.put_bool("OnGround", self.on_ground.load(Relaxed)); @@ -822,12 +857,14 @@ impl NBTStorage for Entity { // todo more... } - async fn read_nbt(&mut self, nbt: &mut pumpkin_nbt::compound::NbtCompound) { + async fn read_nbt(&self, nbt: &pumpkin_nbt::compound::NbtCompound) { let position = nbt.get_list("Pos").unwrap(); let x = position[0].extract_double().unwrap_or(0.0); let y = position[1].extract_double().unwrap_or(0.0); let z = position[2].extract_double().unwrap_or(0.0); - self.set_pos(Vector3::new(x, y, z)); + let pos = Vector3::new(x, y, z); + self.set_pos(pos); + self.first_loaded_chunk_position.store(Some(pos.to_i32())); let velocity = nbt.get_list("Motion").unwrap(); let x = velocity[0].extract_double().unwrap_or(0.0); let y = velocity[1].extract_double().unwrap_or(0.0); @@ -854,7 +891,7 @@ impl NBTStorage for Entity { #[async_trait] pub trait NBTStorage: Send + Sync + Sized { - async fn write_nbt(&self, nbt: &mut NbtCompound); + async fn write_nbt(&self, _nbt: &mut NbtCompound) {} async fn read_nbt(&mut self, _nbt: &mut NbtCompound) {} diff --git a/pumpkin/src/entity/player.rs b/pumpkin/src/entity/player.rs index ca2eafa5f..811294a73 100644 --- a/pumpkin/src/entity/player.rs +++ b/pumpkin/src/entity/player.rs @@ -9,6 +9,7 @@ use std::time::{Duration, Instant}; use async_trait::async_trait; use crossbeam::atomic::AtomicCell; use log::warn; +use pumpkin_world::chunk::{ChunkData, ChunkEntityData}; use pumpkin_world::inventory::Inventory; use tokio::sync::{Mutex, RwLock}; use tokio::task::JoinHandle; @@ -69,7 +70,7 @@ use pumpkin_world::entity::entity_data_flags::{ DATA_PLAYER_MAIN_HAND, DATA_PLAYER_MODE_CUSTOMISATION, SLEEPING_POS_ID, }; use pumpkin_world::item::ItemStack; -use pumpkin_world::level::SyncChunk; +use pumpkin_world::level::{SyncChunk, SyncEntityChunk}; use crate::block::blocks::bed::BedBlock; use crate::command::client_suggestions; @@ -104,6 +105,7 @@ enum BatchState { pub struct ChunkManager { chunks_per_tick: usize, chunk_queue: VecDeque<(Vector2, SyncChunk)>, + entity_chunk_queue: VecDeque<(Vector2, SyncEntityChunk)>, batches_sent_since_ack: BatchState, } @@ -115,6 +117,7 @@ impl ChunkManager { Self { chunks_per_tick, chunk_queue: VecDeque::new(), + entity_chunk_queue: VecDeque::new(), batches_sent_since_ack: BatchState::Initial, } } @@ -128,6 +131,10 @@ impl ChunkManager { self.chunk_queue.push_back((position, chunk)); } + pub fn push_entity(&mut self, position: Vector2, chunk: SyncEntityChunk) { + self.entity_chunk_queue.push_back((position, chunk)); + } + #[must_use] pub fn can_send_chunk(&self) -> bool { let state_available = match self.batches_sent_since_ack { @@ -141,12 +148,30 @@ impl ChunkManager { pub fn next_chunk(&mut self) -> Box<[SyncChunk]> { let chunk_size = self.chunk_queue.len().min(self.chunks_per_tick); - let mut chunks = Vec::with_capacity(chunk_size); - chunks.extend( - self.chunk_queue - .drain(0..chunk_size) - .map(|(_, chunk)| chunk), - ); + let chunks: Vec>> = self + .chunk_queue + .drain(0..chunk_size) + .map(|(_, chunk)| chunk) + .collect(); + + match &mut self.batches_sent_since_ack { + BatchState::Count(count) => { + count.add_assign(1); + } + state @ BatchState::Initial => *state = BatchState::Waiting, + BatchState::Waiting => unreachable!(), + } + + chunks.into_boxed_slice() + } + + pub fn next_entity(&mut self) -> Box<[SyncEntityChunk]> { + let chunk_size = self.entity_chunk_queue.len().min(self.chunks_per_tick); + let chunks: Vec>> = self + .entity_chunk_queue + .drain(0..chunk_size) + .map(|(_, chunk)| chunk) + .collect(); match &mut self.batches_sent_since_ack { BatchState::Count(count) => { @@ -404,6 +429,7 @@ impl Player { let chunks_to_clean = level.mark_chunks_as_not_watched(&radial_chunks).await; // Remove chunks with no watchers from the cache level.clean_chunks(&chunks_to_clean).await; + level.clean_entity_chunks(&chunks_to_clean).await; // Remove left over entries from all possiblily loaded chunks level.clean_memory(); @@ -970,17 +996,6 @@ impl Player { .await; } - /// Sends a world's mobs to only this player. - // TODO: This should be optimized for larger servers based on the player's current chunk. - pub async fn send_mobs(&self, world: &World) { - let entities = world.entities.read().await.clone(); - for entity in entities.values() { - self.client - .enqueue_packet(&entity.get_entity().create_spawn_packet()) - .await; - } - } - async fn unload_watched_chunks(&self, world: &World) { let radial_chunks = self.watched_section.load().all_chunks_within(); let level = &world.level; @@ -1430,10 +1445,14 @@ impl Player { } pub async fn drop_item(&self, item_stack: ItemStack) { - let entity = self.world().await.create_entity( - self.living_entity.entity.pos.load() - + Vector3::new(0.0, f64::from(EntityType::PLAYER.eye_height) - 0.3, 0.0), + let item_pos = self.living_entity.entity.pos.load() + + Vector3::new(0.0, f64::from(EntityType::PLAYER.eye_height) - 0.3, 0.0); + let entity = Entity::new( + Uuid::new_v4(), + self.world().await, + item_pos, EntityType::ITEM, + false, ); let pitch = f64::from(self.living_entity.entity.pitch.load()).to_radians(); @@ -1945,7 +1964,7 @@ impl NBTStorage for PlayerInventory { } // Save the inventory list - nbt.put("Inventory", NbtTag::List(vec.into_boxed_slice())); + nbt.put("Inventory", NbtTag::List(vec)); } async fn read_nbt_non_mut(&self, nbt: &mut NbtCompound) { diff --git a/pumpkin/src/entity/type.rs b/pumpkin/src/entity/type.rs new file mode 100644 index 000000000..4b7d9d584 --- /dev/null +++ b/pumpkin/src/entity/type.rs @@ -0,0 +1,37 @@ +use std::sync::Arc; + +use pumpkin_data::entity::EntityType; +use pumpkin_util::math::vector3::Vector3; +use tokio::sync::Mutex; +use uuid::Uuid; + +use crate::{ + entity::{ + Entity, EntityBase, + ai::path::Navigator, + living::LivingEntity, + mob::{MobEntity, zombie::Zombie}, + }, + world::World, +}; + +pub fn from_type( + entity_type: EntityType, + position: Vector3, + world: &Arc, + uuid: Uuid, +) -> Arc { + let entity = Entity::new(uuid, world.clone(), position, entity_type, false); + + #[allow(clippy::single_match)] + let mob = match entity_type { + EntityType::ZOMBIE => Zombie::make(entity), + // TODO + _ => MobEntity { + living_entity: LivingEntity::new(entity), + goals: Mutex::new(vec![]), + navigator: Mutex::new(Navigator::default()), + }, + }; + Arc::new(mob) +} diff --git a/pumpkin/src/item/items/egg.rs b/pumpkin/src/item/items/egg.rs index 41864be70..84cc15015 100644 --- a/pumpkin/src/item/items/egg.rs +++ b/pumpkin/src/item/items/egg.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use crate::entity::Entity; use crate::entity::player::Player; use crate::entity::projectile::ThrownItemEntity; use crate::item::pumpkin_item::{ItemMetadata, PumpkinItem}; @@ -7,6 +8,7 @@ use async_trait::async_trait; use pumpkin_data::entity::EntityType; use pumpkin_data::item::Item; use pumpkin_data::sound::Sound; +use uuid::Uuid; pub struct EggItem; @@ -31,7 +33,13 @@ impl PumpkinItem for EggItem { ) .await; // TODO: Implement eggs the right way, so there is a chance of spawning chickens - let entity = world.create_entity(position, EntityType::EGG); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + position, + EntityType::EGG, + false, + ); let snowball = ThrownItemEntity::new(entity, &player.living_entity.entity); let yaw = player.living_entity.entity.yaw.load(); let pitch = player.living_entity.entity.pitch.load(); diff --git a/pumpkin/src/item/items/hoe.rs b/pumpkin/src/item/items/hoe.rs index 1e24366cd..b1022e9cd 100644 --- a/pumpkin/src/item/items/hoe.rs +++ b/pumpkin/src/item/items/hoe.rs @@ -1,3 +1,4 @@ +use crate::entity::Entity; use crate::entity::item::ItemEntity; use crate::entity::player::Player; use crate::item::pumpkin_item::{ItemMetadata, PumpkinItem}; @@ -12,6 +13,7 @@ use pumpkin_util::math::position::BlockPos; use pumpkin_world::item::ItemStack; use pumpkin_world::world::BlockFlags; use std::sync::Arc; +use uuid::Uuid; pub struct HoeItem; @@ -89,7 +91,13 @@ impl PumpkinItem for HoeItem { BlockDirection::West => location.up().to_f64().add_raw(-1.0, -0.4, 0.0), BlockDirection::East => location.up().to_f64().add_raw(1.0, -0.4, 0.0), }; - let entity = world.create_entity(location, EntityType::ITEM); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + location, + EntityType::SNOWBALL, + false, + ); // TODO: Merge stacks together let item_entity = Arc::new( ItemEntity::new(entity, ItemStack::new(1, &Item::HANGING_ROOTS)).await, diff --git a/pumpkin/src/item/items/snowball.rs b/pumpkin/src/item/items/snowball.rs index 1a8172a01..e3ea37e7f 100644 --- a/pumpkin/src/item/items/snowball.rs +++ b/pumpkin/src/item/items/snowball.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use crate::entity::Entity; use crate::entity::player::Player; use crate::entity::projectile::ThrownItemEntity; use crate::item::pumpkin_item::{ItemMetadata, PumpkinItem}; @@ -7,6 +8,7 @@ use async_trait::async_trait; use pumpkin_data::entity::EntityType; use pumpkin_data::item::Item; use pumpkin_data::sound::Sound; +use uuid::Uuid; pub struct SnowBallItem; @@ -30,7 +32,13 @@ impl PumpkinItem for SnowBallItem { &position, ) .await; - let entity = world.create_entity(position, EntityType::SNOWBALL); + let entity = Entity::new( + Uuid::new_v4(), + world.clone(), + position, + EntityType::SNOWBALL, + false, + ); let snowball = ThrownItemEntity::new(entity, &player.living_entity.entity); let yaw = player.living_entity.entity.yaw.load(); let pitch = player.living_entity.entity.pitch.load(); diff --git a/pumpkin/src/net/packet/play.rs b/pumpkin/src/net/packet/play.rs index 4456d417d..8b33108e8 100644 --- a/pumpkin/src/net/packet/play.rs +++ b/pumpkin/src/net/packet/play.rs @@ -46,12 +46,14 @@ use pumpkin_world::block::entities::command_block::CommandBlockEntity; use pumpkin_world::block::entities::sign::SignBlockEntity; use pumpkin_world::item::ItemStack; use pumpkin_world::world::BlockFlags; +use uuid::Uuid; use crate::block::registry::BlockActionResult; use crate::block::{self, BlockIsReplacing}; use crate::command::CommandSender; +use crate::entity::EntityBase; use crate::entity::player::{ChatMode, ChatSession, Hand, Player}; -use crate::entity::{EntityBase, mob}; +use crate::entity::r#type::from_type; use crate::error::PumpkinError; use crate::net::PlayerConfig; use crate::plugin::player::player_chat::PlayerChatEvent; @@ -1622,7 +1624,7 @@ impl Player { let world = self.world().await; // Create a new mob and UUID based on the spawn egg id - let mob = mob::from_type(EntityType::from_raw(entity_type.id).unwrap(), pos, &world); + let mob = from_type(entity_type, pos, &world, Uuid::new_v4()); // Set the rotation mob.get_entity().set_rotation(yaw, 0.0); diff --git a/pumpkin/src/world/mod.rs b/pumpkin/src/world/mod.rs index 22c09569f..ad4c24b3b 100644 --- a/pumpkin/src/world/mod.rs +++ b/pumpkin/src/world/mod.rs @@ -12,7 +12,7 @@ use crate::{ PLUGIN_MANAGER, block::{self, registry::BlockRegistry}, command::client_suggestions, - entity::{Entity, EntityBase, EntityId, player::Player}, + entity::{Entity, EntityBase, EntityId, player::Player, r#type::from_type}, error::PumpkinError, plugin::{ block::block_break::BlockBreakEvent, @@ -80,7 +80,7 @@ use pumpkin_util::{ }; use pumpkin_world::{ BlockStateId, GENERATION_SETTINGS, GeneratorSetting, biome, block::entities::BlockEntity, - item::ItemStack, + chunk::io::Dirtiable, item::ItemStack, }; use pumpkin_world::{chunk::ChunkData, world::BlockAccessor}; use pumpkin_world::{chunk::TickPriority, level::Level}; @@ -102,6 +102,7 @@ pub mod custom_bossbar; pub mod scoreboard; pub mod weather; +use uuid::Uuid; use weather::Weather; type FlowingFluidProperties = pumpkin_data::fluid::FlowingWaterLikeFluidProperties; @@ -196,9 +197,62 @@ impl World { } pub async fn shutdown(&self) { + for (uuid, entity) in self.entities.read().await.iter() { + self.save_entity(uuid, entity).await; + } self.level.shutdown().await; } + async fn save_entity(&self, uuid: &uuid::Uuid, entity: &Arc) { + // First lets see if the entity was saved on an other chunk, and if the current chunk does not match we remove it + // Otherwise we just update the nbt data + let base_entity = entity.get_entity(); + let (current_chunk_coordinate, _) = base_entity + .block_pos + .load() + .chunk_and_chunk_relative_position(); + let mut nbt = NbtCompound::new(); + entity.write_nbt(&mut nbt).await; + if let Some(old_chunk) = base_entity.first_loaded_chunk_position.load() { + let old_chunk = old_chunk.to_vec2_i32(); + let chunk = self.level.get_entity_chunk(old_chunk).await; + let mut chunk = chunk.write().await; + chunk.mark_dirty(true); + if old_chunk == current_chunk_coordinate { + chunk.data.insert(*uuid, nbt); + return; + } + + // The chunk has changed, lets remove the entity from the old chunk + chunk.data.remove(uuid); + } + // We did not continue, so lets save data in a new chunk + let chunk = self.level.get_entity_chunk(current_chunk_coordinate).await; + let mut chunk = chunk.write().await; + chunk.data.insert(*uuid, nbt); + chunk.mark_dirty(true); + } + + async fn remove_entity_data(&self, entity: &Entity) { + let (current_chunk_coordinate, _) = + entity.block_pos.load().chunk_and_chunk_relative_position(); + if let Some(old_chunk) = entity.first_loaded_chunk_position.load() { + let old_chunk = old_chunk.to_vec2_i32(); + let chunk = self.level.get_entity_chunk(old_chunk).await; + let mut chunk = chunk.write().await; + chunk.mark_dirty(true); + if old_chunk == current_chunk_coordinate { + chunk.data.remove(&entity.entity_uuid); + } else { + let chunk = self.level.get_entity_chunk(current_chunk_coordinate).await; + let mut chunk = chunk.write().await; + // The chunk has changed, lets remove the entity from the old chunk + chunk.data.remove(&entity.entity_uuid); + chunk.mark_dirty(true); + } + } + } + pub async fn send_entity_status(&self, entity: &Entity, status: EntityStatus) { // TODO: only nearby self.broadcast_packet_all(&CEntityStatus::new(entity.entity_id, status as i8)) @@ -895,7 +949,6 @@ impl World { // } player.has_played_before.store(true, Ordering::Relaxed); - player.send_mobs(self).await; player .on_screen_handler_opened(player.player_screen_handler.clone()) .await; @@ -1083,7 +1136,7 @@ impl World { /// IMPORTANT: Chunks have to be non-empty #[allow(clippy::too_many_lines)] fn spawn_world_chunks( - &self, + self: &Arc, player: Arc, chunks: Vec>, center_chunk: Vector2, @@ -1107,8 +1160,12 @@ impl World { rel_x * rel_x + rel_z * rel_z }); - let mut receiver = self.level.receive_chunks(chunks); + let mut receiver = self.level.receive_chunks(chunks.clone()); + let level = self.level.clone(); + let player1 = player.clone(); + let world = self.clone(); + let world1 = self.clone(); player.clone().spawn_task(async move { 'main: loop { @@ -1145,11 +1202,11 @@ impl World { } let (world, chunk) = if level.is_chunk_watched(&position) { - (player.world().await.clone(), chunk) + (world.clone(), chunk) } else { send_cancellable! {{ ChunkSave { - world: player.world().await.clone(), + world: world.clone(), chunk, cancelled: false, }; @@ -1199,6 +1256,102 @@ impl World { } } + #[cfg(debug_assertions)] + log::debug!("Chunks queued after {}ms", inst.elapsed().as_millis()); + }); + let mut entity_receiver = self.level.receive_entity_chunks(chunks); + let level = self.level.clone(); + let player = player1.clone(); + let world = world1.clone(); + player.clone().spawn_task(async move { + 'main: loop { + let recv_result = tokio::select! { + () = player.client.await_close_interrupt() => { + log::debug!("Canceling player packet processing"); + None + }, + recv_result = entity_receiver.recv() => { + recv_result + } + }; + + let Some((chunk, _first_load)) = recv_result else { + break; + }; + let position = chunk.read().await.chunk_position; + + let chunk = if level.is_chunk_watched(&position) { + chunk + } else { + log::trace!( + "Received chunk {:?}, but it is no longer watched... cleaning", + &position + ); + let mut ids = Vec::new(); + // Remove all the entities from the world + let entity_chunk = chunk.read().await; + let mut entities = world.entities.write().await; + for (uuid, entity_nbt) in &entity_chunk.data { + let Some(id) = entity_nbt.get_string("id") else { + log::warn!("Entity has no ID"); + continue; + }; + let Some(entity_type) = + EntityType::from_name(id.strip_prefix("minecraft:").unwrap_or(id)) + else { + log::warn!("Entity has no valid Entity Type {id}"); + continue; + }; + // Pos is zero since it will read from nbt + let entity = + from_type(entity_type, Vector3::new(0.0, 0.0, 0.0), &world, *uuid); + entity.read_nbt(entity_nbt).await; + let base_entity = entity.get_entity(); + + entities.remove(&base_entity.entity_uuid); + ids.push(VarInt(base_entity.entity_id)); + + world.save_entity(uuid, &entity).await; + } + if !ids.is_empty() { + player + .client + .enqueue_packet(&CRemoveEntities::new(&ids)) + .await; + } + level.clean_entity_chunk(&position).await; + + continue 'main; + }; + + let entity_chunk = chunk.read().await; + // Add all new Entities to the world + let mut current_entities = world.entities.write().await; + + for (uuid, entity_nbt) in &entity_chunk.data { + let Some(id) = entity_nbt.get_string("id") else { + log::warn!("Entity has no ID"); + continue; + }; + let Some(entity_type) = + EntityType::from_name(id.strip_prefix("minecraft:").unwrap_or(id)) + else { + log::warn!("Entity has no valid Entity Type {id}"); + continue; + }; + // Pos is zero since it will read from nbt + let entity = from_type(entity_type, Vector3::new(0.0, 0.0, 0.0), &world, *uuid); + entity.read_nbt(entity_nbt).await; + let base_entity = entity.get_entity(); + player + .client + .enqueue_packet(&base_entity.create_spawn_packet()) + .await; + entity.init_data_tracker().await; + current_entities.insert(base_entity.entity_uuid, entity); + } + } + #[cfg(debug_assertions)] log::debug!("Chunks queued after {}ms", inst.elapsed().as_millis()); }); @@ -1427,29 +1580,34 @@ impl World { } } - pub fn create_entity( - self: &Arc, - position: Vector3, - entity_type: EntityType, - ) -> Entity { - let uuid = uuid::Uuid::new_v4(); - Entity::new(uuid, self.clone(), position, entity_type, false) - } - /// Adds an entity to the world. pub async fn spawn_entity(&self, entity: Arc) { let base_entity = entity.get_entity(); self.broadcast_packet_all(&base_entity.create_spawn_packet()) .await; entity.init_data_tracker().await; - let mut current_living_entities = self.entities.write().await; - current_living_entities.insert(base_entity.entity_uuid, entity); + + let (chunk_coordinate, _) = base_entity + .block_pos + .load() + .chunk_and_chunk_relative_position(); + let chunk = self.level.get_entity_chunk(chunk_coordinate).await; + let mut chunk = chunk.write().await; + let mut nbt = NbtCompound::new(); + entity.write_nbt(&mut nbt).await; + chunk.data.insert(base_entity.entity_uuid, nbt); + chunk.mark_dirty(true); + + let mut current_entities = self.entities.write().await; + current_entities.insert(base_entity.entity_uuid, entity); } pub async fn remove_entity(&self, entity: &Entity) { self.entities.write().await.remove(&entity.entity_uuid); self.broadcast_packet_all(&CRemoveEntities::new(&[entity.entity_id.into()])) .await; + + self.remove_entity_data(entity).await; } pub async fn set_block_breaking(&self, from: &Entity, location: BlockPos, progress: i32) { @@ -1483,7 +1641,7 @@ impl World { return block_state_id; } - chunk.dirty = true; + chunk.mark_dirty(true); chunk.section.set_block_absolute_y( relative.x as usize, @@ -1683,7 +1841,7 @@ impl World { f64::from(pos.0.z) + 0.5 + rand::rng().random_range(-0.25..0.25), ); - let entity = self.create_entity(pos, EntityType::ITEM); + let entity = Entity::new(Uuid::new_v4(), self.clone(), pos, EntityType::ITEM, false); let item_entity = Arc::new(ItemEntity::new(entity, stack).await); self.spawn_entity(item_entity).await; } @@ -1875,7 +2033,7 @@ impl World { block_pos, (block_entity_nbt.unwrap_or_default(), block_entity), ); - chunk.dirty = true; + chunk.mark_dirty(true); } pub async fn remove_block_entity(&self, block_pos: &BlockPos) { @@ -1885,7 +2043,7 @@ impl World { .await; let mut chunk: tokio::sync::RwLockWriteGuard = chunk.write().await; chunk.block_entities.remove(block_pos); - chunk.dirty = true; + chunk.mark_dirty(true); } fn intersects_aabb_with_direction(