diff --git a/pumpkin-protocol/src/bedrock/server/player_action.rs b/pumpkin-protocol/src/bedrock/server/player_action.rs index 9a4d81300..5a59c8ca3 100644 --- a/pumpkin-protocol/src/bedrock/server/player_action.rs +++ b/pumpkin-protocol/src/bedrock/server/player_action.rs @@ -22,6 +22,7 @@ pub enum Action { AbortBreak = 1, StopBreak = 2, GetUpdatedBlock = 3, + /// Seems to be not used, or atleast not send by client DropItem = 4, StartSleeping = 5, StopSleeping = 6, diff --git a/pumpkin-world/src/chunk/format/mod.rs b/pumpkin-world/src/chunk/format/mod.rs index 4b2862b36..64a1f9b5c 100644 --- a/pumpkin-world/src/chunk/format/mod.rs +++ b/pumpkin-world/src/chunk/format/mod.rs @@ -12,8 +12,6 @@ use pumpkin_data::{Block, chunk::ChunkStatus, fluid::Fluid}; use pumpkin_nbt::{compound::NbtCompound, nbt_long_array}; use rustc_hash::FxHashMap; use tokio::sync::Mutex; -use tracing::{debug, trace}; -use uuid::Uuid; use crate::{ chunk::{ @@ -289,43 +287,11 @@ impl ChunkEntityData { chunk_entity_data.position[1], ))); } - let mut map = FxHashMap::default(); - for entity_nbt in chunk_entity_data.entities { - if entity_nbt.is_empty() { - continue; - } - let uuid = if let Some(uuid) = entity_nbt.get_int_array("UUID") { - if uuid.len() != 4 { - debug!( - "Entity in chunk {},{} has invalid UUID array length {}: {:?}", - position.x, - position.y, - uuid.len(), - entity_nbt - ); - continue; - } - Uuid::from_u128( - (uuid[0] as u128) << 96 - | (uuid[1] as u128) << 64 - | (uuid[2] as u128) << 32 - | (uuid[3] as u128), - ) - } else { - trace!( - "Entity in chunk {},{} is missing UUID: {:?}", - position.x, position.y, entity_nbt - ); - continue; - }; - - map.insert(uuid, entity_nbt); - } Ok(Self { x: position.x, z: position.y, - data: Mutex::new(map), + data: Mutex::new(chunk_entity_data.entities), dirty: AtomicBool::new(false), }) } @@ -334,7 +300,7 @@ impl ChunkEntityData { let nbt = EntityNbt { data_version: WORLD_DATA_VERSION, position: [self.x, self.z], - entities: self.data.lock().await.values().cloned().collect(), + entities: self.data.lock().await.clone(), }; let mut result = Vec::new(); diff --git a/pumpkin-world/src/chunk/mod.rs b/pumpkin-world/src/chunk/mod.rs index d90bb9f30..d38dfe73b 100644 --- a/pumpkin-world/src/chunk/mod.rs +++ b/pumpkin-world/src/chunk/mod.rs @@ -89,7 +89,7 @@ pub struct ChunkEntityData { pub x: i32, /// Chunk Z pub z: i32, - pub data: Mutex>, + pub data: Mutex>, pub dirty: AtomicBool, } diff --git a/pumpkin-world/src/chunk_system/dag.rs b/pumpkin-world/src/chunk_system/dag.rs index 28528c238..93461c186 100644 --- a/pumpkin-world/src/chunk_system/dag.rs +++ b/pumpkin-world/src/chunk_system/dag.rs @@ -8,6 +8,7 @@ pub struct Node { pub stage: StagedChunkEnum, pub in_degree: u32, pub in_queue: bool, + pub in_flight: bool, pub edge: EdgeKey, } @@ -19,6 +20,7 @@ impl Node { stage, in_degree: 0, in_queue: false, + in_flight: false, edge: EdgeKey::null(), } } diff --git a/pumpkin-world/src/chunk_system/schedule.rs b/pumpkin-world/src/chunk_system/schedule.rs index 7d688fd2e..a4ff7cb2f 100644 --- a/pumpkin-world/src/chunk_system/schedule.rs +++ b/pumpkin-world/src/chunk_system/schedule.rs @@ -77,6 +77,7 @@ pub struct GenerationSchedule { gen_pool: Option>, listener: Arc, lighting_config: LightingEngineConfig, + last_unload: std::time::Instant, } impl GenerationSchedule { @@ -169,6 +170,7 @@ impl GenerationSchedule { listener, chunk_map: Default::default(), lighting_config, + last_unload: std::time::Instant::now(), }; scheduler.work(level_sched); }) @@ -443,9 +445,12 @@ impl GenerationSchedule { { let task = &mut holder.tasks[i]; if !task.is_null() { - self.waiting_for_chunks.remove(task); - self.drop_node(*task); - *task = NodeKey::null(); + let is_in_flight = self.graph.nodes.get(*task).is_some_and(|n| n.in_flight); + if !is_in_flight { + self.waiting_for_chunks.remove(task); + self.drop_node(*task); + *task = NodeKey::null(); + } } } if new_stage == StagedChunkEnum::None @@ -601,20 +606,13 @@ impl GenerationSchedule { self.public_chunk_map.remove(&pos); holder.public = false; } - let sc = Arc::strong_count(&chunk); - if sc == 1 { - if chunk.is_dirty() { - chunks.push((pos, Chunk::Level(chunk))); - } - self.chunk_map.remove(&pos); - } else { - warn!( - "unload_chunk: chunk {pos:?} still has {} strong refs; cannot unload. holder.public={}", - sc, holder.public - ); - self.unload_chunks.insert(pos); - holder.chunk = Some(Chunk::Level(chunk)); + + // Forcefully drop chunks when unloaded to prevent memory leaks + // from dangling strong references + if chunk.is_dirty() { + chunks.push((pos, Chunk::Level(chunk))); } + self.chunk_map.remove(&pos); } Chunk::Proto(chunk) => { debug_assert!(!holder.public); @@ -646,8 +644,8 @@ impl GenerationSchedule { let mut chunks = Vec::with_capacity(self.chunk_map.len()); for (pos, holder) in &mut self.chunk_map { - if let Some(chunk) = holder.chunk.take() { - let should_save = match &chunk { + if let Some(chunk) = &holder.chunk { + let should_save = match chunk { Chunk::Level(sync_chunk) => sync_chunk.is_dirty(), Chunk::Proto(proto) => { save_proto_chunk @@ -660,9 +658,11 @@ impl GenerationSchedule { }; if should_save { - chunks.push((*pos, chunk)); - } else { - holder.chunk = Some(chunk); + let chunk_to_save = match chunk { + Chunk::Level(sync_chunk) => Chunk::Level(sync_chunk.clone()), + Chunk::Proto(_) => holder.chunk.take().unwrap(), + }; + chunks.push((*pos, chunk_to_save)); } } } @@ -803,13 +803,14 @@ impl GenerationSchedule { if was_public { self.apply_lighting_override(&chunk); - holder.chunk = Some(Chunk::Level(chunk.clone())); - self.public_chunk_map.insert(new_pos, chunk.clone()); + let public_chunk = chunk.clone(); + self.public_chunk_map.insert(new_pos, public_chunk); info!( "Notifying players: regenerated chunk at {:?} (was already public)", new_pos ); self.listener.process_new_chunk(new_pos, &chunk); + holder.chunk = Some(Chunk::Level(chunk)); } else { self.apply_lighting_override(&chunk); let public_chunk = chunk.clone(); @@ -1009,9 +1010,13 @@ impl GenerationSchedule { // 1. Get latest world state (player moves, etc) self.resort_work(self.send_level.get()); - // Process unload queue continuously if there are chunks to unload - if !self.unload_chunks.is_empty() { + // Process unload queue periodically (every 1 second) to batch writes together + // and act as a brief memory cache if a player walks back into the chunk. + if !self.unload_chunks.is_empty() + && self.last_unload.elapsed() >= std::time::Duration::from_secs(1) + { self.process_unload_queue(); + self.last_unload = std::time::Instant::now(); } // 2. Process all pending chunk results from workers @@ -1056,6 +1061,7 @@ impl GenerationSchedule { node.in_queue = false; continue; } + node.in_flight = true; let node = node.clone(); if node.stage == StagedChunkEnum::Empty { self.running_task_count += 1; @@ -1114,6 +1120,7 @@ impl GenerationSchedule { if !all_ready { if let Some(n) = self.graph.nodes.get_mut(task.1) { n.in_queue = false; + n.in_flight = false; } self.waiting_for_chunks.insert(task.1); // Close the TOCTOU window: the chunk we're waiting for may diff --git a/pumpkin-world/src/chunk_system/worker_logic.rs b/pumpkin-world/src/chunk_system/worker_logic.rs index e2b054df2..afb0fadbb 100644 --- a/pumpkin-world/src/chunk_system/worker_logic.rs +++ b/pumpkin-world/src/chunk_system/worker_logic.rs @@ -76,17 +76,17 @@ pub async fn io_read_work( let (t_send, mut t_recv) = tokio::sync::mpsc::channel(1000); + let batch_len = batch.len(); let level_clone = level.clone(); - let batch_clone = batch.clone(); let fetch_task = tokio::spawn(async move { level_clone .chunk_saver - .fetch_chunks(&level_clone.level_folder, &batch_clone, t_send) + .fetch_chunks(&level_clone.level_folder, &batch, t_send) .await; }); - for _ in 0..batch.len() { + for _ in 0..batch_len { let data = match t_recv.recv().await { Some(res) => res, None => break, diff --git a/pumpkin-world/src/level.rs b/pumpkin-world/src/level.rs index e9b6d4271..397dd2839 100644 --- a/pumpkin-world/src/level.rs +++ b/pumpkin-world/src/level.rs @@ -24,7 +24,7 @@ use pumpkin_data::dimension::Dimension; use pumpkin_data::{Block, block_properties::has_random_ticks, fluid::Fluid}; use pumpkin_util::math::{position::BlockPos, vector2::Vector2}; use pumpkin_util::world_seed::Seed; -use rustc_hash::{FxHashMap, FxHashSet}; +use rustc_hash::FxHashSet; use std::sync::{Arc, Mutex, Weak}; use std::time::Duration; use std::{ @@ -119,8 +119,10 @@ pub struct RandomTickSample { pub struct LevelFolder { pub root_folder: PathBuf, + pub dim_folder: PathBuf, pub region_folder: PathBuf, pub entities_folder: PathBuf, + pub poi_folder: PathBuf, } impl Level { @@ -132,16 +134,26 @@ impl Level { dimension: Dimension, gen_pool: Option>, ) -> Arc { - let region_folder = root_folder.join("region"); - let entities_folder = root_folder.join("entities"); + let (namespace, name) = match dimension.minecraft_name.split_once(':') { + Some((ns, n)) => (ns, n), + None => ("minecraft", dimension.minecraft_name), + }; + let dim_folder = root_folder.join("dimensions").join(namespace).join(name); + + let region_folder = dim_folder.join("region"); + let entities_folder = dim_folder.join("entities"); + let poi_folder = dim_folder.join("poi"); std::fs::create_dir_all(®ion_folder).expect("Failed to create Region folder"); std::fs::create_dir_all(&entities_folder).expect("Failed to create Entities folder"); + std::fs::create_dir_all(&poi_folder).expect("Failed to create POI folder"); let level_folder = Arc::new(LevelFolder { root_folder, + dim_folder, region_folder, entities_folder, + poi_folder, }); let seed = Seed(seed as u64); @@ -225,7 +237,7 @@ impl Level { let arc_chunk = Arc::new(ChunkEntityData { x: pos.x, z: pos.y, - data: tokio::sync::Mutex::new(FxHashMap::default()), + data: tokio::sync::Mutex::new(Vec::new()), dirty: AtomicBool::new(false), }); @@ -246,7 +258,7 @@ impl Level { let arc_chunk = Arc::new(ChunkEntityData { x: pos.x, z: pos.y, - data: tokio::sync::Mutex::new(FxHashMap::default()), + data: tokio::sync::Mutex::new(Vec::new()), dirty: AtomicBool::new(false), }); @@ -322,8 +334,9 @@ impl Level { self.tasks.wait().await; self.chunk_system_tasks.wait().await; - info!("Flushing data to disk for {}...", world_id); + info!("Flushing chunk data to disk for {}...", world_id); self.chunk_saver.block_and_await_ongoing_tasks().await; + info!("Flushing entity data to disk for {}...", world_id); self.entity_saver.block_and_await_ongoing_tasks().await; // save all chunks currently in memory @@ -521,9 +534,7 @@ impl Level { .map(|entry| *entry.key()) .collect(); - if !entity_chunks_to_remove.is_empty() { - self.clean_entity_chunks(&entity_chunks_to_remove); - } + // We do not clean them here because we want the caller to save any active entities in them first. // if the difference is too big, we can shrink the loaded chunks // (1024 chunks is the equivalent to a 32x32 chunks area) diff --git a/pumpkin-world/src/poi/mod.rs b/pumpkin-world/src/poi/mod.rs index ad4fccc58..c12a6cc7c 100644 --- a/pumpkin-world/src/poi/mod.rs +++ b/pumpkin-world/src/poi/mod.rs @@ -376,9 +376,9 @@ pub struct PoiStorage { impl PoiStorage { #[must_use] - pub fn new(world_folder: &Path) -> Self { + pub fn new(poi_folder: PathBuf) -> Self { Self { - folder: world_folder.join("poi"), + folder: poi_folder, regions: HashMap::new(), } } @@ -532,7 +532,7 @@ mod tests { let dir = std::env::temp_dir().join("pumpkin_poi_mca_test"); let _ = std::fs::remove_dir_all(&dir); - let mut storage = PoiStorage::new(&dir); + let mut storage = PoiStorage::new(dir.join("poi")); storage.add_portal(BlockPos(Vector3::new(100, 64, 100))); storage.add_portal(BlockPos(Vector3::new(110, 64, 100))); @@ -552,7 +552,7 @@ mod tests { assert!(mca_path.exists()); // Reload and verify - let mut storage2 = PoiStorage::new(&dir); + let mut storage2 = PoiStorage::new(dir.join("poi")); let results2 = storage2.get_in_square( BlockPos(Vector3::new(105, 64, 100)), 16, diff --git a/pumpkin/src/entity/mod.rs b/pumpkin/src/entity/mod.rs index 2ad1b31fc..0d1365a32 100644 --- a/pumpkin/src/entity/mod.rs +++ b/pumpkin/src/entity/mod.rs @@ -3300,6 +3300,13 @@ impl NBTStorage for Entity { if self.has_visual_fire.load(Relaxed) { nbt.put_bool("HasVisualFire", true); } + nbt.put_int("TicksFrozen", self.frozen_ticks.load(Relaxed)); + if let Some(custom_name) = &**self.custom_name.load() + && let Ok(name_json) = pumpkin_util::serde_json::to_string(custom_name) + { + nbt.put_string("CustomName", name_json); + } + nbt.put_bool("CustomNameVisible", self.custom_name_visible.load(Relaxed)); // todo more... }) @@ -3340,6 +3347,15 @@ impl NBTStorage for Entity { .store(nbt.get_int("PortalCooldown").unwrap_or(0) as u32, Relaxed); self.has_visual_fire .store(nbt.get_bool("HasVisualFire").unwrap_or(false), Relaxed); + self.frozen_ticks + .store(nbt.get_int("TicksFrozen").unwrap_or(0), Relaxed); + if let Some(name_json) = nbt.get_string("CustomName") + && let Ok(component) = pumpkin_util::serde_json::from_str(name_json) + { + self.custom_name.store(Arc::new(Some(component))); + } + self.custom_name_visible + .store(nbt.get_bool("CustomNameVisible").unwrap_or(false), Relaxed); // todo more... }) } diff --git a/pumpkin/src/entity/player.rs b/pumpkin/src/entity/player.rs index 242105510..f10ed8be5 100644 --- a/pumpkin/src/entity/player.rs +++ b/pumpkin/src/entity/player.rs @@ -886,11 +886,15 @@ impl Player { let chunks_to_clean = level.mark_chunks_as_not_watched(&radial_chunks).await; // Remove chunks with no watchers from the cache if !chunks_to_clean.is_empty() { + world.remove_entities_in_chunks(&chunks_to_clean).await; level.clean_entity_chunks(&chunks_to_clean); - world.remove_entities_in_chunks(&chunks_to_clean); } // Remove left over entries from all possiblily loaded chunks - level.clean_memory(); + let cleaned_chunks = level.clean_memory(); + if !cleaned_chunks.is_empty() { + world.remove_entities_in_chunks(&cleaned_chunks).await; + level.clean_entity_chunks(&cleaned_chunks); + } debug!( "Removed player id {} from world {} ({} chunks remain cached)", @@ -3093,6 +3097,7 @@ impl Player { .await { screen_handler.set_received_stack(slot_index, updated_stack); + screen_handler.send_content_updates().await; } } @@ -4869,19 +4874,170 @@ impl InventoryPlayer for Player { packet: &'a CSetContainerContent, ) -> PlayerFuture<'a, ()> { Box::pin(async move { - self.client.enqueue_packet(packet).await; + match self.client.as_ref() { + ClientPlatform::Java(java) => { + java.enqueue_packet(packet).await; + } + ClientPlatform::Bedrock(bedrock) => { + use pumpkin_protocol::bedrock::{ + client::inventory_content::CInventoryContent, + network_item::{ + ContainerName, FullContainerName, NetworkItemStackDescriptor, + }, + }; + use pumpkin_protocol::codec::var_uint::VarUInt; + + let window_id = packet.window_id.0 as u32; + let slots: Vec = packet + .slot_data + .iter() + .map(|s| NetworkItemStackDescriptor::from(&*s.0)) + .collect(); + + if window_id == 0 { + let bedrock_packet = CInventoryContent { + container_id: VarUInt(0), + slots, + full_container_name: FullContainerName { + container_name: ContainerName::Inventory, + dynamic_id: None, + }, + storage_item: NetworkItemStackDescriptor::default(), + }; + bedrock.enqueue_packet(&bedrock_packet).await; + } + } + } }) } fn enqueue_slot_packet<'a>(&'a self, packet: &'a CSetContainerSlot) -> PlayerFuture<'a, ()> { Box::pin(async move { - self.client.enqueue_packet(packet).await; + match self.client.as_ref() { + ClientPlatform::Java(java) => { + java.enqueue_packet(packet).await; + } + ClientPlatform::Bedrock(bedrock) => { + use pumpkin_protocol::bedrock::{ + client::inventory_content::CInventoryContent, + client::inventory_slot::CInventorySlot, + network_item::{ + ContainerName, FullContainerName, NetworkItemStackDescriptor, + }, + }; + use pumpkin_protocol::codec::var_uint::VarUInt; + + let window_id = packet.window_id; + tracing::info!( + "enqueue_slot_packet: window_id={}, slot={}", + window_id, + packet.slot + ); + + if window_id == 0 { + tracing::info!( + "enqueue_slot_packet: window_id is 0, sending CInventoryContent to Bedrock client" + ); + let mut slots = Vec::with_capacity(36); + let main_inventory = &self.inventory().main_inventory; + for s in main_inventory { + let stack = s.lock().await; + slots.push(NetworkItemStackDescriptor::from(&*stack)); + } + + let bedrock_packet = CInventoryContent { + container_id: VarUInt(0), + slots, + full_container_name: FullContainerName { + container_name: ContainerName::Inventory, + dynamic_id: None, + }, + storage_item: NetworkItemStackDescriptor::default(), + }; + bedrock.enqueue_packet(&bedrock_packet).await; + } else { + let slot_idx = packet.slot as usize; + let item_desc = NetworkItemStackDescriptor::from(&*packet.slot_data.0); + + // Container screen + let current_handler = self.current_screen_handler.lock().await.clone(); + let handler = current_handler.lock().await; + let window_type = handler.window_type(); + let total_slots = handler.get_behaviour().slots.len(); + let bedrock_info = if total_slots >= 36 { + let container_slots = total_slots - 36; + if slot_idx < container_slots { + if window_type == Some(WindowType::Crafting) { + if slot_idx == 0 { + Some((ContainerName::CreatedOutput, 0)) + } else { + Some(( + ContainerName::CraftingInput, + (32 + slot_idx - 1) as u8, + )) + } + } else { + Some((ContainerName::LevelEntity, slot_idx as u8)) + } + } else { + let inv_slot = slot_idx - container_slots; + if inv_slot < 27 { + Some((ContainerName::Inventory, (inv_slot + 9) as u8)) + } else { + Some((ContainerName::Inventory, (inv_slot - 27) as u8)) + } + } + } else { + None + }; + + if let Some((container_name, slot_id)) = bedrock_info { + let bedrock_packet = CInventorySlot { + window_id: VarUInt(window_id as u32), + inventory_slot: VarUInt(slot_id as u32), + container_name: Some(FullContainerName { + container_name, + dynamic_id: None, + }), + storage: None, + item: item_desc, + }; + bedrock.enqueue_packet(&bedrock_packet).await; + } + } + } + } }) } fn enqueue_cursor_packet<'a>(&'a self, packet: &'a CSetCursorItem) -> PlayerFuture<'a, ()> { Box::pin(async move { - self.client.enqueue_packet(packet).await; + match self.client.as_ref() { + ClientPlatform::Java(java) => { + java.enqueue_packet(packet).await; + } + ClientPlatform::Bedrock(bedrock) => { + use pumpkin_protocol::bedrock::{ + client::inventory_content::CInventoryContent, + network_item::{ + ContainerName, FullContainerName, NetworkItemStackDescriptor, + }, + }; + use pumpkin_protocol::codec::var_uint::VarUInt; + + let item_desc = NetworkItemStackDescriptor::from(&*packet.stack.0); + let bedrock_packet = CInventoryContent { + container_id: VarUInt(59), + slots: vec![item_desc], + full_container_name: FullContainerName { + container_name: ContainerName::Cursor, + dynamic_id: None, + }, + storage_item: NetworkItemStackDescriptor::default(), + }; + bedrock.enqueue_packet(&bedrock_packet).await; + } + } }) } @@ -4899,7 +5055,42 @@ impl InventoryPlayer for Player { packet: &'a CSetPlayerInventory, ) -> PlayerFuture<'a, ()> { Box::pin(async move { - self.client.enqueue_packet(packet).await; + match self.client.as_ref() { + ClientPlatform::Java(java) => { + java.enqueue_packet(packet).await; + } + ClientPlatform::Bedrock(bedrock) => { + use pumpkin_protocol::bedrock::{ + client::inventory_content::CInventoryContent, + network_item::{ + ContainerName, FullContainerName, NetworkItemStackDescriptor, + }, + }; + use pumpkin_protocol::codec::var_uint::VarUInt; + + tracing::info!( + "enqueue_slot_set_packet: slot={}, sending CInventoryContent to Bedrock client", + packet.slot.0 + ); + let mut slots = Vec::with_capacity(36); + let main_inventory = &self.inventory().main_inventory; + for s in main_inventory { + let stack = s.lock().await; + slots.push(NetworkItemStackDescriptor::from(&*stack)); + } + + let bedrock_packet = CInventoryContent { + container_id: VarUInt(0), + slots, + full_container_name: FullContainerName { + container_name: ContainerName::Inventory, + dynamic_id: None, + }, + storage_item: NetworkItemStackDescriptor::default(), + }; + bedrock.enqueue_packet(&bedrock_packet).await; + } + } }) } diff --git a/pumpkin/src/net/bedrock/play.rs b/pumpkin/src/net/bedrock/play.rs index 18f5ef892..f52bccff7 100644 --- a/pumpkin/src/net/bedrock/play.rs +++ b/pumpkin/src/net/bedrock/play.rs @@ -451,10 +451,69 @@ impl BedrockClient { player: &Arc, packet: SInventoryTransaction, ) { + tracing::info!("handle_inventory_action: packet={:?}", packet); let mut inventory_updated = false; + let mut updates = Vec::new(); + let result = 0u8; + + if packet.actions.is_empty() && packet.legacy_request_id.0 != 0 { + let mut player_screen_handler = player.player_screen_handler.lock().await; + for legacy_slot in &packet.legacy_set_item_slots { + let mapped_window_id = match legacy_slot.container_id { + 28 | 29 => 0, // HotBar or Inventory + 6 | 120 => 120, // Armor + 34 | 119 => 119, // Offhand + other => other as i32, + }; + for &slot_id in &legacy_slot.slots { + if let Some(screen_slot) = + map_bedrock_slot_to_screen_handler(mapped_window_id, slot_id as u32) + { + let current_stack = player_screen_handler + .get_slot(screen_slot) + .get_cloned_stack() + .await; + if !current_stack.is_empty() { + player.drop_item(current_stack.clone()).await; + + player_screen_handler + .get_slot(screen_slot) + .set_stack(ItemStack::EMPTY.clone()) + .await; + player_screen_handler + .set_received_stack(screen_slot, ItemStack::EMPTY.clone()); + + record_update( + &mut updates, + FullContainerName { + container_name: match legacy_slot.container_id { + 28 => ContainerName::HotBar, + _ => ContainerName::Inventory, + }, + dynamic_id: None, + }, + slot_id, + 0, + VarInt(0), + ); + inventory_updated = true; + } + } + } + } + player_screen_handler.send_content_updates().await; + } for action in &packet.actions { - if let Some(window_id) = action.window_id { + use pumpkin_protocol::bedrock::server::inventory_transaction::InventoryActionSource; + let source_type = InventoryActionSource::from(action.source_type); + if source_type == InventoryActionSource::World { + let old_stack = descriptor_to_stack(&action.old_item); + let new_stack = descriptor_to_stack(&action.new_item); + if old_stack.is_empty() && !new_stack.is_empty() { + player.drop_item(new_stack).await; + } + } else if let Some(window_id) = action.window_id { if let Some(screen_slot) = map_bedrock_slot_to_screen_handler(window_id, action.inventory_slot) { @@ -635,6 +694,52 @@ impl BedrockClient { player.living_entity.clear_active_hand().await; } } + + if packet.legacy_request_id.0 != 0 { + use pumpkin_protocol::bedrock::client::item_stack_response::{ + CItemStackResponse, ItemStackResponse, ItemStackResponseContainerInfo, + ItemStackResponseSlotInfo, + }; + + let mut container_infos = Vec::new(); + if result == 0 { + for update in updates { + let container_info = container_infos.iter_mut().find( + |info: &&mut ItemStackResponseContainerInfo| { + info.container_name == update.container_name + }, + ); + + let slot_info = ItemStackResponseSlotInfo { + slot: update.slot_id, + hotbar_slot: update.slot_id, + count: update.count, + item_stack_id: update.stack_id, + custom_name: String::new(), + filtered_custom_name: String::new(), + durability_correction: VarInt(0), + }; + + if let Some(info) = container_info { + info.slots.push(slot_info); + } else { + container_infos.push(ItemStackResponseContainerInfo { + container_name: update.container_name.clone(), + slots: vec![slot_info], + }); + } + } + } + + self.enqueue_packet(&CItemStackResponse { + responses: vec![ItemStackResponse { + result, + request_id: packet.legacy_request_id, + container_infos, + }], + }) + .await; + } } pub async fn handle_interaction(&self, player: &Arc, packet: SInteraction) { @@ -868,6 +973,9 @@ impl BedrockClient { player.mining.store(false, Ordering::Relaxed); world.set_block_breaking(entity, location, -1).await; } + PlayerAction::DropItem => { + player.drop_held_item(false).await; + } // TODO _ => {} } @@ -953,6 +1061,7 @@ impl BedrockClient { let mut result = 0u8; // 0 = Success, 1 = Error for action in request.actions { + tracing::info!("Processing ItemStackRequestAction: {:?}", action); match action { ItemStackRequestAction::CraftCreative { creative_item_id, @@ -1041,6 +1150,29 @@ impl BedrockClient { } source_stack.decrement(count); + if source.container_name.container_name == ContainerName::CreatedOutput + { + if let Some(ref mut stack) = created_item { + stack.decrement(count); + if stack.is_empty() { + created_item = None; + } + } + } else if source.container_name.container_name == ContainerName::Cursor + { + let cursor_is_empty = screen_handler + .get_behaviour() + .cursor_stack + .lock() + .await + .is_empty(); + if cursor_is_empty && let Some(ref mut stack) = created_item { + stack.decrement(count); + if stack.is_empty() { + created_item = None; + } + } + } let source_stack = if source_stack.is_empty() { ItemStack::EMPTY.clone() } else { @@ -1554,11 +1686,23 @@ fn map_bedrock_container_slot( ContainerName::Cursor => None, ContainerName::CraftingInput => { if is_player_screen { - (slot_id < 4).then(|| 1 + slot_id as usize) + if slot_id < 4 { + Some(1 + slot_id as usize) + } else if (28..32).contains(&slot_id) { + Some(1 + (slot_id - 28) as usize) + } else { + None + } } else if screen_handler.window_type() == Some(pumpkin_data::screen::WindowType::Crafting) { - (slot_id < 9).then(|| 1 + slot_id as usize) + if slot_id < 9 { + Some(1 + slot_id as usize) + } else if (32..41).contains(&slot_id) { + Some(1 + (slot_id - 32) as usize) + } else { + None + } } else { None } @@ -1800,6 +1944,11 @@ async fn get_slot_stack( } if slot_info.container_name.container_name == ContainerName::Cursor { let cursor_lock = screen_handler.get_behaviour().cursor_stack.lock().await; + if cursor_lock.is_empty() + && let Some(stack) = created_item + { + return stack.clone(); + } return cursor_lock.clone(); } if let Some(screen_slot) = map_bedrock_container_slot( diff --git a/pumpkin/src/plugin/loader/wasm/wasm_host/state.rs b/pumpkin/src/plugin/loader/wasm/wasm_host/state.rs index 7ef2cb96b..166bd4f35 100644 --- a/pumpkin/src/plugin/loader/wasm/wasm_host/state.rs +++ b/pumpkin/src/plugin/loader/wasm/wasm_host/state.rs @@ -31,7 +31,6 @@ use crate::{ server::{RecipeManager, Server}, world::World, }; -use pumpkin_world::level::SyncChunk; pub struct WasmResource { pub provider: T, @@ -44,7 +43,7 @@ pub type JavaPlayerResource = WasmResource>; pub type BedrockPlayerResource = WasmResource>; pub type EntityResource = WasmResource>; pub type WorldResource = WasmResource>; -pub type ChunkResource = WasmResource<(Arc, SyncChunk)>; +pub type ChunkResource = WasmResource<(Arc, Weak)>; pub type WorldBorderResource = WasmResource>; pub type ScoreboardResource = WasmResource>; pub type GuiResource = WasmResource>>; @@ -157,7 +156,7 @@ impl PluginHostState { pub fn add_chunk( &mut self, world: Arc, - chunk: SyncChunk, + chunk: Weak, ) -> wasmtime::Result> { let resource = self.resource_table.push(ChunkResource { provider: (world, chunk), diff --git a/pumpkin/src/plugin/loader/wasm/wasm_host/wit/v0_1/world.rs b/pumpkin/src/plugin/loader/wasm/wasm_host/wit/v0_1/world.rs index cd65d98f9..af383ad4a 100644 --- a/pumpkin/src/plugin/loader/wasm/wasm_host/wit/v0_1/world.rs +++ b/pumpkin/src/plugin/loader/wasm/wasm_host/wit/v0_1/world.rs @@ -229,7 +229,7 @@ impl pumpkin::plugin::world::HostWorld for PluginHostState { .get(&pos) .map(|c| c.value().clone()); if let Some(chunk) = chunk { - let res = self.add_chunk(world_provider, chunk)?; + let res = self.add_chunk(world_provider, std::sync::Arc::downgrade(&chunk))?; Ok(Some(res)) } else { Ok(None) @@ -696,12 +696,18 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { async fn get_x(&mut self, chunk: Resource) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; Ok(chunk_data.x) } async fn get_z(&mut self, chunk: Resource) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; Ok(chunk_data.z) } @@ -712,6 +718,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; Ok(chunk_data .section .get_block_absolute_y(pos.x as usize, pos.y, pos.z as usize) @@ -725,6 +734,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; let id = chunk_data .section .get_block_absolute_y(pos.x as usize, pos.y, pos.z as usize) @@ -783,6 +795,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result<()> { let chunk_res = self.get_chunk_res(&chunk)?; let (world, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; let replaced = chunk_data.set_block_absolute_y(pos.x as usize, pos.y, pos.z as usize, state); @@ -804,6 +819,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; let id = chunk_data .section .get_rough_biome_absolute_y(pos.x as usize, pos.y, pos.z as usize) @@ -821,6 +839,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result> { let chunk_res = self.get_chunk_res(&chunk)?; let (world, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; let absolute_pos = BlockPos::new(chunk_data.x * 16 + pos.x, pos.y, chunk_data.z * 16 + pos.z); let block_entity = world.get_block_entity(&absolute_pos); @@ -836,6 +857,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; Ok(chunk_data.heightmap.lock().unwrap().get( ChunkHeightmapType::WorldSurface, x, @@ -851,6 +875,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; let section_index = (pos.y - chunk_data.section.min_y) as usize / 16; Ok(chunk_data .light_engine @@ -870,6 +897,9 @@ impl pumpkin::plugin::world::HostChunk for PluginHostState { ) -> wasmtime::Result { let chunk_res = self.get_chunk_res(&chunk)?; let (_, chunk_data) = &chunk_res.provider; + let Some(chunk_data) = chunk_data.upgrade() else { + return Err(wasmtime::Error::msg("Chunk unloaded")); + }; let section_index = (pos.y - chunk_data.section.min_y) as usize / 16; Ok(chunk_data .light_engine diff --git a/pumpkin/src/server/mod.rs b/pumpkin/src/server/mod.rs index b03592b3f..010db0c8d 100644 --- a/pumpkin/src/server/mod.rs +++ b/pumpkin/src/server/mod.rs @@ -196,12 +196,13 @@ impl Server { let defaultgamemode = Mutex::new(DefaultGamemode { gamemode: basic_config.default_gamemode, }); + let players_dir = world_path.join("players"); let player_data_storage = ServerPlayerData::new( - world_path.join("playerdata"), + players_dir.join("data"), Duration::from_secs(advanced_config.player_data.save_player_cron_interval), advanced_config.player_data.save_player_data, ); - let advancement_manager = Arc::new(AdvancementManager::new(world_path.clone(), true)); + let advancement_manager = Arc::new(AdvancementManager::new(players_dir.clone(), true)); let white_list = AtomicBool::new(basic_config.white_list); let tick_rate_manager = Arc::new(ServerTickRateManager::new(basic_config.tps)); diff --git a/pumpkin/src/world/chunker.rs b/pumpkin/src/world/chunker.rs index 4178bd582..202c8562c 100644 --- a/pumpkin/src/world/chunker.rs +++ b/pumpkin/src/world/chunker.rs @@ -114,8 +114,8 @@ pub async fn update_position(player: &Arc) { .await; if !chunks_to_clean.is_empty() { + world.remove_entities_in_chunks(&chunks_to_clean).await; world.level.clean_entity_chunks(&chunks_to_clean); - world.remove_entities_in_chunks(&chunks_to_clean); } if !loading_chunks.is_empty() { diff --git a/pumpkin/src/world/mod.rs b/pumpkin/src/world/mod.rs index 2d7295587..48b85d81b 100644 --- a/pumpkin/src/world/mod.rs +++ b/pumpkin/src/world/mod.rs @@ -285,7 +285,7 @@ impl World { let generation_settings = GenerationSettings::from_dimension(&dimension); // Load portal POI from disk (PoiStorage::new automatically loads from disk if files exist) - let portal_poi = portal::PortalPoiStorage::new(&level.level_folder.root_folder); + let portal_poi = portal::PortalPoiStorage::new(level.level_folder.poi_folder.clone()); let dragon_fight = (dimension.minecraft_name == Dimension::THE_END.minecraft_name) .then(|| Mutex::new(dragon_fight::DragonFight::new())); Self { @@ -377,7 +377,6 @@ impl World { // 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 uuid = base_entity.entity_uuid; let current_chunk_coordinate = base_entity.block_pos.load().chunk_position(); let mut nbt = NbtCompound::new(); entity.write_nbt(&mut nbt).await; @@ -388,37 +387,21 @@ impl World { chunk.mark_dirty(true); let mut data = chunk.data.lock().await; if old_chunk == current_chunk_coordinate { - data.insert(uuid, nbt); + data.push(nbt); return; } // The chunk has changed, lets remove the entity from the old chunk - data.remove(&uuid); + // TODO? + data.clear(); } // 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 data = chunk.data.lock().await; - data.insert(uuid, nbt); + data.push(nbt); chunk.mark_dirty(true); } - async fn remove_entity_data(&self, entity: &Entity) { - let current_chunk_coordinate = entity.block_pos.load().chunk_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; - chunk.mark_dirty(true); - if old_chunk == current_chunk_coordinate { - chunk.data.lock().await.remove(&entity.entity_uuid); - } else { - let chunk = self.level.get_entity_chunk(current_chunk_coordinate).await; - // The chunk has changed, lets remove the entity from the old chunk - chunk.data.lock().await.remove(&entity.entity_uuid); - chunk.mark_dirty(true); - } - } - } - /// Sends an entity status update to all players tracking the specified entity. pub fn send_entity_status(&self, entity: &Entity, status: EntityStatus) { let chunk_pos = entity.chunk_pos.load(); @@ -1134,7 +1117,8 @@ impl World { self.level.should_unload.store(true, Relaxed); let cleaned_chunks = self.level.clean_memory(); if !cleaned_chunks.is_empty() { - self.remove_entities_in_chunks(&cleaned_chunks); + self.remove_entities_in_chunks(&cleaned_chunks).await; + self.level.clean_entity_chunks(&cleaned_chunks); } // If autosave is configured and this tick will trigger an autosave, don't double notify if self.level.autosave_ticks == 0 { @@ -3526,6 +3510,7 @@ impl World { let mut entity_receiver = self.level.receive_entity_chunks(chunks); let level = self.level.clone(); let world = self.clone(); + player.clone().spawn_task(async move { 'main: loop { let recv_result = tokio::select! { @@ -3553,62 +3538,46 @@ impl World { "Received chunk {:?}, but it is no longer watched... cleaning", &position ); - let mut ids_to_remove = Vec::new(); - for (uuid, entity_nbt) in chunk.data.lock().await.iter() { - let Some(id) = entity_nbt.get_string("id") else { - warn!("Entity has no ID"); - continue; - }; - let Some(entity_type) = - EntityType::from_name(id.strip_prefix("minecraft:").unwrap_or(id)) - else { - 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_non_mut(entity_nbt).await; - let base_entity = entity.get_entity(); + if first_load { + for entity_nbt in chunk.data.lock().await.iter() { + let Some(id) = entity_nbt.get_string("id") else { + continue; + }; + let Some(entity_type) = + EntityType::from_name(id.strip_prefix("minecraft:").unwrap_or(id)) + else { + continue; + }; - ids_to_remove.push(VarInt(base_entity.entity_id)); + let entity = from_type( + entity_type, + Vector3::new(0.0, 0.0, 0.0), + &world, + Uuid::new_v4(), + ); + entity.read_nbt_non_mut(entity_nbt).await; + let base_entity = entity.get_entity(); - if first_load { 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 = world.level.get_entity_chunk(old_chunk).await; chunk.mark_dirty(true); - let base_entity = entity.get_entity(); let current_chunk_coordinate = base_entity.block_pos.load().chunk_position(); let mut data = chunk.data.lock().await; if old_chunk == current_chunk_coordinate { - data.insert(*uuid, nbt); + data.push(nbt); continue; } // The chunk has changed, lets remove the entity from the old chunk - data.remove(uuid); - } - } - } - if !ids_to_remove.is_empty() { - match player.client.as_ref() { - crate::net::ClientPlatform::Java(java) => { - java.enqueue_packet(&CRemoveEntities::new(&ids_to_remove)) - .await; - } - crate::net::ClientPlatform::Bedrock(bedrock) => { - for id in &ids_to_remove { - bedrock - .enqueue_packet(&CRemoveActor::new(VarLong(id.0 as i64))) - .await; - } + data.clear(); } } } @@ -3617,7 +3586,7 @@ impl World { // Add all new Entities to the world let mut entities_to_add: Vec> = Vec::new(); - for (uuid, entity_nbt) in chunk.data.lock().await.iter() { + for entity_nbt in chunk.data.lock().await.iter() { let Some(id) = entity_nbt.get_string("id") else { debug!("Entity has no ID"); continue; @@ -3629,29 +3598,17 @@ impl World { continue; }; - // Check if this entity already exists in the world (e.g. another player - // is still online and tracking it). If so, just send the spawn packet - // for the existing entity to the reconnecting player instead of - // creating a duplicate with a different entity_id. - let existing = world - .entities - .load() - .iter() - .find(|e| e.get_entity().entity_uuid == *uuid) - .cloned(); - if let Some(existing_entity) = existing { - let base_entity = existing_entity.get_entity(); - player - .client - .enqueue_packet(&base_entity.create_spawn_packet()) - .await; - existing_entity.init_data_tracker().await; - 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); + let entity = from_type( + entity_type, + Vector3::new(0.0, 0.0, 0.0), + &world, + Uuid::new_v4(), + ); entity.read_nbt_non_mut(entity_nbt).await; + + entity.init_data_tracker().await; + let base_entity = entity.get_entity(); // Clear velocity so the client does not replay the drop animation. @@ -3663,12 +3620,12 @@ impl World { .client .enqueue_packet(&base_entity.create_spawn_packet()) .await; - entity.init_data_tracker().await; if first_load { entities_to_add.push(entity); } } + if first_load && !entities_to_add.is_empty() { world.entities.rcu(|current_entities| { let mut new_entities = (**current_entities).clone(); @@ -4152,7 +4109,7 @@ impl World { self.spawn_state.load().add_entity(self, entity.as_ref()); let chunk = self.level.get_entity_chunk(chunk_coordinate).await; - chunk.data.lock().await.insert(base_entity.entity_uuid, nbt); + chunk.data.lock().await.push(nbt); chunk.mark_dirty(true); self.entities.rcu(|current_entities| { @@ -4162,6 +4119,7 @@ impl World { }); } + #[allow(clippy::unused_async)] pub async fn remove_entity(&self, entity: &dyn EntityBase) { let base_entity = entity.get_entity(); self.spawn_state.load().remove_entity(self, entity); @@ -4177,11 +4135,9 @@ impl World { &CRemoveEntities::new(&[base_entity.entity_id.into()]), &CRemoveActor::new(VarLong(base_entity.entity_id as i64)), ); - - self.remove_entity_data(base_entity).await; } - pub fn remove_entities_in_chunks(&self, chunks: &[Vector2]) { + pub async fn remove_entities_in_chunks(&self, chunks: &[Vector2]) { let chunks_set: FxHashSet<_> = chunks.iter().copied().collect(); let mut entities_to_remove = Vec::new(); @@ -4201,9 +4157,8 @@ impl World { }); for entity in entities_to_remove { + self.save_entity(&entity).await; self.spawn_state.load().remove_entity(self, entity.as_ref()); - // Important: We do NOT call remove_entity_data here because we want the entities - // to persist in the chunk data on disk. We only remove them from the active world (RAM). } self.block_entities