diff --git a/pumpkin-protocol/src/client/play/c_chunk_data.rs b/pumpkin-protocol/src/client/play/c_chunk_data.rs index 0a68eace2..85c12effd 100644 --- a/pumpkin-protocol/src/client/play/c_chunk_data.rs +++ b/pumpkin-protocol/src/client/play/c_chunk_data.rs @@ -6,9 +6,9 @@ use pumpkin_macros::packet; use pumpkin_world::{chunk::ChunkData, DIRECT_PALETTE_BITS}; #[packet(0x27)] -pub struct CChunkData<'a>(pub &'a ChunkData); +pub struct CChunkData(pub ChunkData); -impl<'a> ClientPacket for CChunkData<'a> { +impl ClientPacket for CChunkData { fn write(&self, buf: &mut crate::bytebuf::ByteBuffer) { // Chunk X buf.put_i32(self.0.position.x); diff --git a/pumpkin-world/src/chunk/anvil.rs b/pumpkin-world/src/chunk/anvil.rs index 983b3fea1..06705fb08 100644 --- a/pumpkin-world/src/chunk/anvil.rs +++ b/pumpkin-world/src/chunk/anvil.rs @@ -10,6 +10,7 @@ use crate::level::SaveFile; use super::{ChunkData, ChunkReader, ChunkReadingError, CompressionError}; +#[derive(Clone)] pub struct AnvilChunkReader {} impl Default for AnvilChunkReader { diff --git a/pumpkin-world/src/chunk/mod.rs b/pumpkin-world/src/chunk/mod.rs index c75de68a6..9136de62b 100644 --- a/pumpkin-world/src/chunk/mod.rs +++ b/pumpkin-world/src/chunk/mod.rs @@ -54,11 +54,12 @@ pub enum CompressionError { LZ4Error(std::io::Error), } +#[derive(Clone)] pub struct ChunkData { pub blocks: ChunkBlocks, pub position: Vector2, } - +#[derive(Clone)] pub struct ChunkBlocks { // TODO make this a Vec that doesn't store the upper layers that only contain air diff --git a/pumpkin-world/src/level.rs b/pumpkin-world/src/level.rs index 8fff3a6cf..b4207583a 100644 --- a/pumpkin-world/src/level.rs +++ b/pumpkin-world/src/level.rs @@ -1,16 +1,16 @@ use std::{collections::HashMap, path::PathBuf, sync::Arc}; -use parking_lot::{Mutex, RwLock}; -use pumpkin_core::math::vector2::Vector2; -use rayon::prelude::*; -use tokio::sync::mpsc; - use crate::{ chunk::{ anvil::AnvilChunkReader, ChunkData, ChunkParsingError, ChunkReader, ChunkReadingError, }, world_gen::{get_world_gen, Seed, WorldGenerator}, }; +use pumpkin_core::math::vector2::Vector2; +use tokio::sync::mpsc; +use tokio::sync::{Mutex, RwLock}; + +type RAMChunkStorage = Arc, Arc>>>>; /// The `Level` module provides functionality for working with chunks within or outside a Minecraft world. /// @@ -23,12 +23,12 @@ use crate::{ /// For more details on world generation, refer to the `WorldGenerator` module. pub struct Level { save_file: Option, - loaded_chunks: Arc, Arc>>>, + loaded_chunks: RAMChunkStorage, chunk_watchers: Arc, usize>>>, - chunk_reader: Box, - world_gen: Box, + chunk_reader: Arc>, + world_gen: Arc>, } - +#[derive(Clone)] pub struct SaveFile { #[expect(dead_code)] root_folder: PathBuf, @@ -37,7 +37,7 @@ pub struct SaveFile { impl Level { pub fn from_root_folder(root_folder: PathBuf) -> Self { - let world_gen = get_world_gen(Seed(0)); // TODO Read Seed from config. + let world_gen = get_world_gen(Seed(0)).into(); // TODO Read Seed from config. if root_folder.exists() { let region_folder = root_folder.join("region"); @@ -52,7 +52,7 @@ impl Level { root_folder, region_folder, }), - chunk_reader: Box::new(AnvilChunkReader::new()), + chunk_reader: Arc::new(Box::new(AnvilChunkReader::new())), loaded_chunks: Arc::new(RwLock::new(HashMap::new())), chunk_watchers: Arc::new(Mutex::new(HashMap::new())), } @@ -64,7 +64,7 @@ impl Level { Self { world_gen, save_file: None, - chunk_reader: Box::new(AnvilChunkReader::new()), + chunk_reader: Arc::new(Box::new(AnvilChunkReader::new())), loaded_chunks: Arc::new(RwLock::new(HashMap::new())), chunk_watchers: Arc::new(Mutex::new(HashMap::new())), } @@ -76,8 +76,8 @@ impl Level { /// Marks chunks as "watched" by a unique player. When no players are watching a chunk, /// it is removed from memory. Should only be called on chunks the player was not watching /// before - pub fn mark_chunk_as_newly_watched(&self, chunks: &[Vector2]) { - let mut watchers = self.chunk_watchers.lock(); + pub async fn mark_chunk_as_newly_watched(&self, chunks: &[Vector2]) { + let mut watchers = self.chunk_watchers.lock().await; for chunk in chunks { match watchers.entry(*chunk) { std::collections::hash_map::Entry::Occupied(mut occupied) => { @@ -93,9 +93,9 @@ impl Level { /// Marks chunks no longer "watched" by a unique player. When no players are watching a chunk, /// it is removed from memory. Should only be called on chunks the player was watching before - pub fn mark_chunk_as_not_watched_and_clean(&self, chunks: &[Vector2]) { + pub async fn mark_chunk_as_not_watched_and_clean(&self, chunks: &[Vector2]) { let dropped_chunks = { - let mut watchers = self.chunk_watchers.lock(); + let mut watchers = self.chunk_watchers.lock().await; chunks .iter() .filter(|chunk| match watchers.entry(**chunk) { @@ -119,7 +119,7 @@ impl Level { }) .collect::>() }; - let mut loaded_chunks = self.loaded_chunks.write(); + let mut loaded_chunks = self.loaded_chunks.write().await; let dropped_chunk_data = dropped_chunks .iter() .filter_map(|chunk| { @@ -130,7 +130,7 @@ impl Level { self.write_chunks(dropped_chunk_data); } - pub fn write_chunks(&self, _chunks_to_write: Vec<(Vector2, Arc)>) { + pub fn write_chunks(&self, _chunks_to_write: Vec<(Vector2, Arc>)>) { //TODO } @@ -138,58 +138,76 @@ impl Level { /// MUST be called from a tokio runtime thread /// /// Note: The order of the output chunks will almost never be in the same order as the order of input chunks + pub fn fetch_chunks( + &self, + chunks: &[Vector2], + channel: mpsc::Sender>>, + ) { + for chunk in chunks { + { + let chunk_location = *chunk; + let channel = channel.clone(); + let loaded_chunks = self.loaded_chunks.clone(); + let chunk_reader = self.chunk_reader.clone(); + let save_file = self.save_file.clone(); + let world_gen = self.world_gen.clone(); + tokio::spawn(async move { + let loaded_chunks_read = loaded_chunks.read().await; + let possibly_loaded_chunk = loaded_chunks_read.get(&chunk_location).cloned(); + drop(loaded_chunks_read); + match possibly_loaded_chunk { + Some(chunk) => { + let chunk = chunk.clone(); + channel.send(chunk).await.unwrap(); + } + None => { + let chunk_data = match save_file { + Some(save_file) => { + match chunk_reader.read_chunk(&save_file, &chunk_location) { + Ok(data) => Ok(Arc::new(RwLock::new(data))), + Err( + ChunkReadingError::ChunkNotExist + | ChunkReadingError::ParsingError( + ChunkParsingError::ChunkNotGenerated, + ), + ) => { + // This chunk was not generated yet. + let chunk = Arc::new(RwLock::new( + world_gen.generate_chunk(chunk_location), + )); + let mut loaded_chunks = loaded_chunks.write().await; + loaded_chunks.insert(chunk_location, chunk.clone()); + drop(loaded_chunks); + Ok(chunk) + } + Err(err) => Err(err), + } + } + None => { + // There is no savefile yet -> generate the chunks + let chunk = Arc::new(RwLock::new( + world_gen.generate_chunk(chunk_location), + )); - pub fn fetch_chunks(&self, chunks: &[Vector2], channel: mpsc::Sender>) { - chunks.into_par_iter().for_each(|at| { - let channel = channel.clone(); - - let maybe_chunk = { - let loaded_chunks = self.loaded_chunks.read(); - loaded_chunks.get(at).cloned() - } - .or_else(|| { - let chunk_data = match &self.save_file { - Some(save_file) => { - match self.chunk_reader.read_chunk(save_file, at) { - Ok(data) => Ok(Arc::new(data)), - Err( - ChunkReadingError::ChunkNotExist - | ChunkReadingError::ParsingError( - ChunkParsingError::ChunkNotGenerated, - ), - ) => { - // This chunk was not generated yet. - let chunk = Arc::new(self.world_gen.generate_chunk(*at)); - Ok(chunk) + let mut loaded_chunks = loaded_chunks.write().await; + loaded_chunks.insert(chunk_location, chunk.clone()); + Ok(chunk) + } + }; + match chunk_data { + Ok(data) => channel.send(data).await.unwrap(), + Err(err) => { + log::warn!( + "Failed to read chunk {:?}: {:?}", + chunk_location, + err + ); + } } - Err(err) => Err(err), } } - None => { - // There is no savefile yet -> generate the chunks - let chunk = Arc::new(self.world_gen.generate_chunk(*at)); - Ok(chunk) - } - }; - match chunk_data { - Ok(data) => Some(data), - Err(err) => { - // TODO: Panic here? - log::warn!("Failed to read chunk {:?}: {:?}", at, err); - None - } - } - }); - match maybe_chunk { - Some(chunk) => { - channel - .blocking_send(chunk.clone()) - .expect("Failed sending ChunkData."); - } - None => { - log::error!("Unable to send chunk {:?}!", at); - } - }; - }) + }); + } + } } } diff --git a/pumpkin/src/client/player_packet.rs b/pumpkin/src/client/player_packet.rs index feec47f28..30bd72518 100644 --- a/pumpkin/src/client/player_packet.rs +++ b/pumpkin/src/client/player_packet.rs @@ -20,7 +20,7 @@ use pumpkin_protocol::{ client::play::{ Animation, CAcknowledgeBlockChange, CBlockUpdate, CEntityAnimation, CEntityVelocity, CHeadRot, CHurtAnimation, CPingResponse, CPlayerChatMessage, CUpdateEntityPos, - CUpdateEntityPosRot, CUpdateEntityRot, CWorldEvent, FilterType, + CUpdateEntityPosRot, CUpdateEntityRot, FilterType, }, server::play::{ Action, ActionType, SChatCommand, SChatMessage, SClientInformationPlay, SConfirmTeleport, @@ -486,13 +486,7 @@ impl Player { // TODO: currently this is always dirt replace it let entity = &self.living_entity.entity; let world = &entity.world; - world - .broadcast_packet_all(&CWorldEvent::new(2001, &location, 11, false)) - .await; - // AIR - world - .broadcast_packet_all(&CBlockUpdate::new(&location, 0.into())) - .await; + world.break_block(location).await; } } Status::CancelledDigging => { @@ -522,13 +516,7 @@ impl Player { // TODO: currently this is always dirt replace it let entity = &self.living_entity.entity; let world = &entity.world; - world - .broadcast_packet_all(&CWorldEvent::new(2001, &location, 11, false)) - .await; - // AIR - world - .broadcast_packet_all(&CBlockUpdate::new(&location, 0.into())) - .await; + world.break_block(location).await; // TODO: Send this every tick self.client .send_packet(&CAcknowledgeBlockChange::new(player_action.sequence)) diff --git a/pumpkin/src/world/mod.rs b/pumpkin/src/world/mod.rs index 7aec512b8..607244cdd 100644 --- a/pumpkin/src/world/mod.rs +++ b/pumpkin/src/world/mod.rs @@ -1,3 +1,4 @@ +use std::ops::Deref; use std::{collections::HashMap, sync::Arc}; pub mod player_chunker; @@ -6,10 +7,12 @@ use crate::{ client::Client, entity::{player::Player, Entity}, }; -use num_traits::ToPrimitive; +use num_traits::{Euclid, ToPrimitive}; use pumpkin_config::BasicConfiguration; +use pumpkin_core::math::position::WorldPosition; use pumpkin_core::math::vector2::Vector2; use pumpkin_entity::{entity_type::EntityType, EntityId}; +use pumpkin_protocol::client::play::{CBlockUpdate, CWorldEvent}; use pumpkin_protocol::{ client::play::{ CChunkData, CGameEvent, CLogin, CPlayerAbilities, CPlayerInfoUpdate, CRemoveEntities, @@ -17,6 +20,8 @@ use pumpkin_protocol::{ }, ClientPacket, VarInt, }; +use pumpkin_world::block::BlockId; +use pumpkin_world::coordinates::ChunkRelativeBlockCoordinates; use pumpkin_world::level::Level; use scoreboard::Scoreboard; use tokio::sync::mpsc; @@ -273,12 +278,12 @@ impl World { pub async fn mark_chunks_as_not_watched(&self, chunks: &[Vector2]) { let level = self.level.lock().await; - level.mark_chunk_as_not_watched_and_clean(chunks); + level.mark_chunk_as_not_watched_and_clean(chunks).await; } pub async fn mark_chunks_as_watched(&self, chunks: &[Vector2]) { let level = self.level.lock().await; - level.mark_chunk_as_newly_watched(chunks); + level.mark_chunk_as_newly_watched(chunks).await; } fn spawn_world_chunks(&self, client: Arc, chunks: Vec>, distance: i32) { @@ -293,23 +298,24 @@ impl World { let (sender, mut chunk_receiver) = mpsc::channel(distance as usize); let client_id = client.id; - let level = self.level.clone(); - let chunks = Arc::new(chunks); - tokio::spawn(async move { - log::debug!("Spawned chunk fetcher for {}", client_id); - let level = level.lock().await; - level.fetch_chunks(&chunks, sender); - }); + { + let level = self.level.clone(); + tokio::spawn(async move { + log::debug!("Spawned chunk fetcher for {}", client_id); + let level = level.lock().await; + level.fetch_chunks(&chunks, sender); + }); + } tokio::spawn(async move { log::debug!("Spawned chunk sender for {}", client_id); while let Some(chunk_data) = chunk_receiver.recv().await { - // dbg!(chunk_pos); + let chunk_data = chunk_data.read().await.deref().clone(); #[cfg(debug_assertions)] if chunk_data.position == (0, 0).into() { use pumpkin_protocol::bytebuf::ByteBuffer; let mut test = ByteBuffer::empty(); - CChunkData(&chunk_data).write(&mut test); + CChunkData(chunk_data.clone()).write(&mut test); let len = test.buf().len(); log::debug!( "Chunk packet size: {}B {}KB {}MB", @@ -318,11 +324,12 @@ impl World { len / (1024 * 1024) ); } + let packet = CChunkData(chunk_data); // TODO: Queue player packs in a queue so we don't need to check if its closed before // sending if !client.closed.load(std::sync::atomic::Ordering::Relaxed) { - client.send_packet(&CChunkData(&chunk_data)).await; + client.send_packet(&packet).await; } } @@ -374,4 +381,39 @@ impl World { self.broadcast_packet_all(&CRemoveEntities::new(&[entity.entity_id.into()])) .await; } + pub async fn set_block(&self, position: WorldPosition, block_id: BlockId) { + let (z_chunk, z_rem) = position.0.z.div_rem_euclid(&16); + let (x_chunk, x_rem) = position.0.x.div_rem_euclid(&16); + let chunk_coordinate = Vector2 { + x: x_chunk, + z: z_chunk, + }; + + // Since we divide by 16 remnant can never exceed u8 + let relative = ChunkRelativeBlockCoordinates { + x: (x_rem as u8).into(), + z: (z_rem as u8).into(), + y: position.0.y.into(), + }; + + let (sender, mut receive) = mpsc::channel(1024); + { + let level = self.level.clone(); + tokio::spawn( + async move { level.lock().await.fetch_chunks(&[chunk_coordinate], sender) }, + ); + } + if let Some(data) = receive.recv().await { + data.write().await.blocks.set_block(relative, block_id); + self.broadcast_packet_all(&CBlockUpdate::new(&position, (block_id.data as i32).into())) + .await; + } + } + + pub async fn break_block(&self, position: WorldPosition) { + self.set_block(position, BlockId { data: 0 }).await; + + self.broadcast_packet_all(&CWorldEvent::new(2001, &position, 11, false)) + .await; + } }