feat: add rudimentary .pump format

also this should fix a dead lock

README.md

@@ -48,9 +48,9 @@ and customizable experience. It prioritizes performance and player enjoyment whi
   - [x] Lighting
   - [x] Entity Spawning
   - [x] Bossbar
-  - [x] Chunk Loading (Vanilla, Linear)
+  - [x] Chunk Loading (Vanilla, Linear, Pump)
   - [Chunk Generation](https://github.com/Pumpkin-MC/Pumpkin/issues/36)
-  - [x] Chunk Saving (Vanilla, Linear)
+  - [x] Chunk Saving (Vanilla, Linear, Pump)
   - [Redstone](https://github.com/Pumpkin-MC/Pumpkin/issues/1402)
   - [x] Liquid Physics
   - ...

pumpkin-config/src/chunk.rs

@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
 /// Configuration for chunk storage format.
 ///
 /// Supports multiple chunk formats, currently `Anvil` and `Linear`.
-#[derive(Deserialize, Serialize, Clone)]
+#[derive(Deserialize, Default, Serialize, Clone)]
 #[serde(tag = "type")]
 pub enum ChunkConfig {
     /// Standard Anvil chunk storage.
@@ -14,12 +14,10 @@ pub enum ChunkConfig {
     /// Linear chunk storage format.
     #[serde(rename = "linear")]
     Linear,
-}
+    /// Pumpkin's own optimized world format.
-
+    #[serde(rename = "pump")]
-impl Default for ChunkConfig {
+    #[default]
-    fn default() -> Self {
+    Pump,
-        Self::Anvil(AnvilChunkConfig::default())
-    }
 }
 
 /// Configuration for Anvil chunk storage.

pumpkin-world/src/chunk/format/mod.rs

@@ -35,6 +35,7 @@ use super::{
 };
 pub mod anvil;
 pub mod linear;
+pub mod pump;
 
 impl SingleChunkDataSerializer for ChunkData {
     #[inline]

pumpkin-world/src/chunk/format/pump.rs

@@ -0,0 +1,149 @@
+use std::marker::PhantomData;
+use std::path::PathBuf;
+
+use crate::chunk::format::anvil::{CHUNK_COUNT, SingleChunkDataSerializer};
+use crate::chunk::io::{ChunkSerializer, LoadedData};
+use crate::chunk::{ChunkReadingError, ChunkWritingError};
+use bytes::Bytes;
+use pumpkin_util::math::vector2::Vector2;
+use ruzstd::decoding::StreamingDecoder;
+use ruzstd::encoding::{CompressionLevel, compress_to_vec};
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+pub struct PumpFile<D> {
+    pub data: PumpData,
+    _phantom: PhantomData<D>,
+}
+
+#[derive(Serialize, Deserialize, Default)]
+pub struct PumpData {
+    pub x: i32,
+    pub z: i32,
+    pub chunks: Vec<Option<Vec<u8>>>,
+}
+
+impl<D> Default for PumpFile<D> {
+    fn default() -> Self {
+        Self {
+            data: PumpData {
+                x: 0,
+                z: 0,
+                chunks: vec![None; CHUNK_COUNT],
+            },
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<D> ChunkSerializer for PumpFile<D>
+where
+    D: SingleChunkDataSerializer + Send + Sync + Sized,
+{
+    type Data = D;
+    type WriteBackend = PathBuf;
+    type ChunkConfig = ();
+
+    fn get_chunk_key(chunk: &Vector2<i32>) -> String {
+        let region_x = chunk.x >> 5;
+        let region_z = chunk.y >> 5;
+        format!("r.{region_x}.{region_z}.pump")
+    }
+
+    fn should_write(&self, _is_watched: bool) -> bool {
+        true
+    }
+
+    async fn write(&self, backend: &Self::WriteBackend) -> Result<(), std::io::Error> {
+        let bytes =
+            pumpkin_nbt::to_pnbt(&self.data).map_err(|e| std::io::Error::other(e.to_string()))?;
+
+        let compressed = compress_to_vec(&bytes[..], CompressionLevel::Fastest);
+
+        tokio::fs::write(backend, compressed).await
+    }
+
+    fn read(r: Bytes) -> Result<Self, ChunkReadingError> {
+        let mut decoder = StreamingDecoder::new(&r[..])
+            .map_err(|e| ChunkReadingError::IoError(std::io::Error::other(e.to_string()).kind()))?;
+        let mut decompressed = Vec::new();
+        std::io::Read::read_to_end(&mut decoder, &mut decompressed)
+            .map_err(|e| ChunkReadingError::IoError(e.kind()))?;
+
+        let data: PumpData = pumpkin_nbt::from_pnbt(&decompressed).map_err(|e| {
+            ChunkReadingError::ParsingError(
+                crate::chunk::ChunkParsingError::ErrorDeserializingChunk(e.to_string()),
+            )
+        })?;
+
+        Ok(Self {
+            data,
+            _phantom: PhantomData,
+        })
+    }
+
+    async fn update_chunk(
+        &mut self,
+        chunk_data: &Self::Data,
+        _chunk_config: &Self::ChunkConfig,
+    ) -> Result<(), ChunkWritingError> {
+        let (x, z) = chunk_data.position();
+        self.data.x = x >> 5;
+        self.data.z = z >> 5;
+        let rel_x = x.rem_euclid(32);
+        let rel_z = z.rem_euclid(32);
+        let index = (rel_x + rel_z * 32) as usize;
+
+        let bytes = chunk_data
+            .to_bytes()
+            .await
+            .map_err(|e| ChunkWritingError::ChunkSerializingError(e.to_string()))?;
+
+        if index < self.data.chunks.len() {
+            self.data.chunks[index] = Some(bytes.to_vec());
+        } else {
+            warn!("Chunk position {}, {} out of range for Pump region", x, z);
+        }
+
+        Ok(())
+    }
+
+    async fn get_chunks(
+        &self,
+        chunks: Vec<Vector2<i32>>,
+        stream: tokio::sync::mpsc::Sender<LoadedData<Self::Data, ChunkReadingError>>,
+    ) {
+        for pos in chunks {
+            let rel_x = pos.x.rem_euclid(32);
+            let rel_z = pos.y.rem_euclid(32);
+            let index = (rel_x + rel_z * 32) as usize;
+
+            if index < self.data.chunks.len() {
+                if let Some(chunk_bytes) = &self.data.chunks[index] {
+                    let bytes = Bytes::copy_from_slice(chunk_bytes);
+                    match D::from_bytes(&bytes, pos) {
+                        Ok(data) => {
+                            let _ = stream.send(LoadedData::Loaded(data)).await;
+                        }
+                        Err(e) => {
+                            let _ = stream.send(LoadedData::Error((pos, e))).await;
+                        }
+                    }
+                } else {
+                    let _ = stream.send(LoadedData::Missing(pos)).await;
+                }
+            } else {
+                let _ = stream
+                    .send(LoadedData::Error((
+                        pos,
+                        ChunkReadingError::ParsingError(
+                            crate::chunk::ChunkParsingError::ErrorDeserializingChunk(
+                                "Chunk out of range".to_string(),
+                            ),
+                        ),
+                    )))
+                    .await;
+            }
+        }
+    }
+}

pumpkin-world/src/level.rs

@@ -1,4 +1,5 @@
 use crate::chunk::format::linear::LinearV2File;
+use crate::chunk::format::pump::PumpFile;
 use crate::chunk_system::{ChunkListener, ChunkLoading, GenerationSchedule, LevelChannel};
 use crate::generation::generator::VanillaGenerator;
 use crate::lighting::DynamicLightEngine;
@@ -168,6 +169,7 @@ impl Level {
             ChunkConfig::Anvil(config) => Arc::new(
                 ChunkFileManager::<AnvilChunkFile<ChunkData>>::new(config.clone()),
             ),
+            ChunkConfig::Pump => Arc::new(ChunkFileManager::<PumpFile<ChunkData>>::new(())),
         };
         let entity_saver: Arc<dyn FileIO<Data = SyncEntityChunk>> = match &level_config.chunk {
             ChunkConfig::Linear => {
@@ -176,6 +178,7 @@ impl Level {
             ChunkConfig::Anvil(config) => Arc::new(ChunkFileManager::<
                 AnvilChunkFile<ChunkEntityData>,
             >::new(config.clone())),
+            ChunkConfig::Pump => Arc::new(ChunkFileManager::<PumpFile<ChunkEntityData>>::new(())),
         };
 
         let pending_entity_generations = Arc::new(DashMap::new());

pumpkin/src/entity/item.rs

@@ -127,9 +127,22 @@ impl ItemEntity {
     }
 
     async fn try_merge_with(&self, other: &Self) {
-        let self_stack = self.item_stack.lock().await;
+        // Always lock in entity_id order to prevent deadlock when two
+        // items try to merge with each other concurrently.
+        let (low, high) = if self.entity.entity_id < other.entity.entity_id {
+            (self, other)
+        } else {
+            (other, self)
+        };
 
-        let other_stack = other.item_stack.lock().await;
+        let low_stack = low.item_stack.lock().await;
+        let high_stack = high.item_stack.lock().await;
+
+        let (self_stack, other_stack) = if self.entity.entity_id < other.entity.entity_id {
+            (low_stack, high_stack)
+        } else {
+            (high_stack, low_stack)
+        };
 
         if !self_stack.are_equal(&other_stack)
             || self_stack.item_count + other_stack.item_count > self_stack.get_max_stack_size()
@@ -419,7 +432,6 @@ impl EntityBase for ItemEntity {
                 return false;
             }
 
-            // Thread safe damage application
             loop {
                 let current = self.health.load(Relaxed);
                 let new = current - amount;
@@ -446,20 +458,26 @@ impl EntityBase for ItemEntity {
                 return;
             }
 
-            if player
+            let inserted = {
-                .inventory
+                let mut stack = self.item_stack.lock().await;
-                .insert_stack_anywhere(&mut *self.item_stack.lock().await)
+                player.inventory.insert_stack_anywhere(&mut stack).await
-                .await
+            };
-                || player.is_creative()
+
-            {
+            if inserted || player.is_creative() {
+                let (item_count, is_empty) = {
+                    let stack = self.item_stack.lock().await;
+                    (stack.item_count, stack.is_empty())
+                };
+
                 player
                     .client
                     .enqueue_packet(&CTakeItemEntity::new(
                         self.entity.entity_id.into(),
                         player.entity_id().into(),
-                        self.item_stack.lock().await.item_count.into(),
+                        item_count.into(),
                     ))
                     .await;
+
                 player
                     .current_screen_handler
                     .lock()
@@ -468,10 +486,10 @@ impl EntityBase for ItemEntity {
                     .await
                     .send_content_updates()
                     .await;
-                if self.item_stack.lock().await.is_empty() {
+
+                if is_empty {
                     self.entity.remove().await;
                 } else {
-                    // Update entity
                     self.init_data_tracker().await;
                 }
             }

pumpkin/src/entity/living.rs

@@ -1135,8 +1135,7 @@ impl LivingEntity {
             let yaw = f64::from(self.entity.yaw.load()).to_radians();
 
             velo.x -= yaw.sin() * 0.2;
-
+            velo.z += yaw.cos() * 0.2;
-            velo.y += yaw.cos() * 0.2;
         }
 
         self.entity.velocity.store(velo);
@@ -1147,7 +1146,9 @@ impl LivingEntity {
     async fn get_jump_velocity(&self, mut strength: f64) -> f64 {
         strength *= self.get_attribute_value(&Attributes::JUMP_STRENGTH);
         strength *= f64::from(self.entity.get_jump_velocity_multiplier().await);
-
+        if let Some(effect) = self.get_effect(&StatusEffect::JUMP_BOOST).await {
+            strength += 0.1 * f64::from(effect.amplifier + 1);
+        }
         strength
     }
 
@@ -2031,8 +2032,7 @@ impl EntityBase for LivingEntity {
     }
 
     fn get_gravity(&self) -> f64 {
-        const GRAVITY: f64 = 0.08;
+        self.get_attribute_value(&Attributes::GRAVITY)
-        GRAVITY
     }
 
     #[allow(clippy::too_many_lines)]