Merge pull request #138 from ferrumc-rs/rework/better-chunk-sending

Less terrible chunk sending

Author: Sweattypalms

Cargo.toml

@@ -6,6 +6,7 @@
 # - Workspace lints
 # - Workspace dependencies.
 
+
 [workspace]
 resolver = "2"
 
@@ -104,6 +105,7 @@ ferrumc-utils = { path = "src/lib/utils" }
 ferrumc-world = { path = "src/lib/world" }
 
 
+
 # Asynchronous
 tokio = { version = "1.40.0", features = ["full"] }
 socket2 = "0.5.7"
@@ -114,6 +116,7 @@ async-trait = "0.1.82"
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 log = "0.4.22"
+console-subscriber = "0.4.1"
 
 # Concurrency/Parallelism
 parking_lot = "0.12.3"
@@ -147,6 +150,7 @@ hashbrown = "0.15.0"
 tinyvec = "1.8.0"
 dashmap = "6.1.0"
 uuid = { version = "1.1", features = ["v4", "v3", "serde"] }
+whirlwind = "0.1.1"
 
 # Macros
 lazy_static = "1.5.0"

src/bin/Cargo.toml

@@ -28,7 +28,7 @@ ferrumc-general-purpose = { workspace = true }
 ferrumc-state = { workspace = true }
 
 ctor = { workspace = true }
-parking_lot = { workspace = true }
+parking_lot = { workspace = true, features = ["deadlock_detection"] }
 tracing = { workspace = true }
 tokio = { workspace = true }
 rayon = { workspace = true }

src/bin/src/cli.rs

@@ -6,7 +6,7 @@ pub struct CLIArgs {
     #[command(subcommand)]
     pub command: Option<Command>,
     #[clap(long)]
-    #[arg(value_enum, default_value_t = LogLevel(Level::TRACE))]
+    #[arg(value_enum, default_value_t = LogLevel(Level::DEBUG))]
     pub log: LogLevel,
 }
 

src/bin/src/main.rs

@@ -6,11 +6,14 @@ use crate::errors::BinaryError;
 use clap::Parser;
 use ferrumc_config::statics::get_global_config;
 use ferrumc_config::whitelist::create_whitelist;
+use ferrumc_core::chunks::chunk_receiver::ChunkReceiver;
 use ferrumc_ecs::Universe;
 use ferrumc_general_purpose::paths::get_root_path;
+use ferrumc_net::connection::StreamWriter;
 use ferrumc_net::server::create_server_listener;
 use ferrumc_state::ServerState;
 use ferrumc_world::World;
+use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 use systems::definition;
 use tracing::{error, info};
@@ -28,6 +31,19 @@ async fn main() {
     let cli_args = CLIArgs::parse();
     ferrumc_logging::init_logging(cli_args.log.into());
 
+    check_deadlocks();
+
+    {
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        std::any::TypeId::of::<ChunkReceiver>().hash(&mut hasher);
+        let digest = hasher.finish();
+        println!("ChunkReceiver: {:X}", digest);
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        std::any::TypeId::of::<StreamWriter>().hash(&mut hasher);
+        let digest = hasher.finish();
+        println!("StreamWriter: {:X}", digest);
+    }
+
     match cli_args.command {
         Some(Command::Setup) => {
             info!("Starting setup...");
@@ -109,3 +125,28 @@ async fn create_state() -> Result<ServerState> {
         world: World::new().await,
     })
 }
+fn check_deadlocks() {
+    {
+        use parking_lot::deadlock;
+        use std::thread;
+        use std::time::Duration;
+
+        // Create a background thread which checks for deadlocks every 10s
+        thread::spawn(move || loop {
+            thread::sleep(Duration::from_secs(10));
+            let deadlocks = deadlock::check_deadlock();
+            if deadlocks.is_empty() {
+                continue;
+            }
+
+            println!("{} deadlocks detected", deadlocks.len());
+            for (i, threads) in deadlocks.iter().enumerate() {
+                println!("Deadlock #{}", i);
+                for t in threads {
+                    println!("Thread Id {:#?}", t.thread_id());
+                    println!("{:#?}", t.backtrace());
+                }
+            }
+        });
+    }
+}

src/bin/src/packet_handlers/login_process.rs

@@ -1,4 +1,5 @@
 use ferrumc_config::statics::{get_global_config, get_whitelist};
+use ferrumc_core::chunks::chunk_receiver::ChunkReceiver;
 use ferrumc_core::identity::player_identity::PlayerIdentity;
 use ferrumc_core::transform::grounded::OnGround;
 use ferrumc_core::transform::position::Position;
@@ -38,6 +39,16 @@ async fn handle_login_start(
     let player_identity = PlayerIdentity::new(username.to_string(), uuid);
     debug!("Handling login start event for user: {username}, uuid: {uuid}");
 
+    // Add the player identity component to the ECS for the entity.
+    state
+        .universe
+        .add_component::<PlayerIdentity>(
+            login_start_event.conn_id,
+            PlayerIdentity::new(username.to_string(), uuid),
+        )?
+        .add_component::<ChunkReceiver>(login_start_event.conn_id, ChunkReceiver::default())?;
+
+    //Send a Login Success Response to further the login sequence
     let mut writer = state
         .universe
         .get_mut::<StreamWriter>(login_start_event.conn_id)?;
@@ -184,6 +195,12 @@ async fn handle_ack_finish_configuration(
             &NetEncodeOpts::WithLength,
         )
         .await?;
+
+    let pos = state.universe.get_mut::<Position>(conn_id)?;
+    let mut chunk_recv = state.universe.get_mut::<ChunkReceiver>(conn_id)?;
+    chunk_recv.last_chunk = Some((pos.x as i32, pos.z as i32, String::from("overworld")));
+    chunk_recv.calculate_chunks().await;
+
     send_keep_alive(conn_id, state, &mut writer).await?;
 
     Ok(ack_finish_configuration_event)

src/bin/src/packet_handlers/transform/update_player_position.rs

@@ -1,3 +1,4 @@
+use ferrumc_core::chunks::chunk_receiver::ChunkReceiver;
 use ferrumc_core::transform::grounded::OnGround;
 use ferrumc_core::transform::position::Position;
 use ferrumc_core::transform::rotation::Rotation;
@@ -6,6 +7,7 @@ use ferrumc_net::errors::NetError;
 use ferrumc_net::packets::packet_events::TransformEvent;
 use ferrumc_net::utils::ecs_helpers::EntityExt;
 use ferrumc_state::GlobalState;
+use tracing::trace;
 
 #[event_handler]
 async fn handle_player_move(
@@ -14,19 +16,48 @@ async fn handle_player_move(
 ) -> Result<TransformEvent, NetError> {
     let conn_id = event.conn_id;
     if let Some(ref new_position) = event.position {
-        let mut position = conn_id.get_mut::<Position>(&state)?;
+        trace!("Getting chunk_recv 1 for player move");
+        {
+            let mut chunk_recv = state.universe.get_mut::<ChunkReceiver>(conn_id)?;
+            trace!("Got chunk_recv 1 for player move");
+            if let Some(last_chunk) = &chunk_recv.last_chunk {
+                let new_chunk = (
+                    new_position.x as i32 / 16,
+                    new_position.z as i32 / 16,
+                    String::from("overworld"),
+                );
+                if *last_chunk != new_chunk {
+                    chunk_recv.last_chunk = Some(new_chunk);
+                    chunk_recv.calculate_chunks().await;
+                }
+            } else {
+                chunk_recv.last_chunk = Some((
+                    new_position.x as i32 / 16,
+                    new_position.z as i32 / 16,
+                    String::from("overworld"),
+                ));
+                chunk_recv.calculate_chunks().await;
+            }
+        }
 
+        trace!("Getting position 1 for player move");
+        let mut position = conn_id.get_mut::<Position>(&state)?;
+        trace!("Got position 1 for player move");
         *position = Position::new(new_position.x, new_position.y, new_position.z);
     }
 
     if let Some(ref new_rotation) = event.rotation {
+        trace!("Getting rotation 1 for player move");
         let mut rotation = conn_id.get_mut::<Rotation>(&state)?;
+        trace!("Got rotation 1 for player move");
 
         *rotation = Rotation::new(new_rotation.yaw, new_rotation.pitch);
     }
 
     if let Some(new_grounded) = event.on_ground {
+        trace!("Getting on_ground 1 for player move");
         let mut on_ground = conn_id.get_mut::<OnGround>(&state)?;
+        trace!("Got on_ground 1 for player move");
 
         *on_ground = OnGround(new_grounded);
     }

src/bin/src/systems/chunk_fetcher.rs

@@ -0,0 +1,92 @@
+use crate::errors::BinaryError;
+use crate::systems::definition::System;
+use async_trait::async_trait;
+use ferrumc_core::chunks::chunk_receiver::ChunkReceiver;
+use ferrumc_state::GlobalState;
+use std::collections::HashMap;
+use std::sync::atomic::AtomicBool;
+use std::sync::Arc;
+use tokio::task::JoinSet;
+use tracing::{error, info, trace};
+
+pub struct ChunkFetcher {
+    stop: AtomicBool,
+}
+
+impl ChunkFetcher {
+    pub(crate) fn new() -> Self {
+        Self {
+            stop: AtomicBool::new(false),
+        }
+    }
+}
+
+#[async_trait]
+impl System for ChunkFetcher {
+    async fn start(self: Arc<Self>, state: GlobalState) {
+        info!("Chunk fetcher system started");
+
+        while !self.stop.load(std::sync::atomic::Ordering::Relaxed) {
+            let mut task_set: JoinSet<Result<(), BinaryError>> = JoinSet::new();
+            let players = state.universe.query::<&mut ChunkReceiver>().into_entities();
+            for eid in players {
+                let state = state.clone();
+                task_set.spawn(async move {
+                    // Copy the chunks into a new map so we don't lock the component while fetching
+                    let mut copied_chunks = {
+                        let Ok(chunk_recv) = state.universe.get::<ChunkReceiver>(eid) else {
+                            trace!("A player disconnected before we could get the ChunkReceiver");
+                            return Ok(());
+                        };
+                        let mut copied_chunks = HashMap::new();
+                        for chunk in chunk_recv.needed_chunks.iter() {
+                            let (key, chunk) = chunk.pair();
+                            if chunk.is_none() {
+                                copied_chunks.insert(key.clone(), None);
+                            }
+                        }
+                        copied_chunks
+                    };
+                    // Fetch the chunks
+                    for (key, chunk) in copied_chunks.iter_mut() {
+                        let fetched_chunk =
+                            state.world.load_chunk(key.0, key.1, &key.2.clone()).await?;
+                        *chunk = Some(fetched_chunk);
+                    }
+                    // Insert the fetched chunks back into the component
+                    {
+                        let Ok(chunk_recv) = state.universe.get::<ChunkReceiver>(eid) else {
+                            trace!("A player disconnected before we could get the ChunkReceiver");
+                            return Ok(());
+                        };
+                        for (key, chunk) in copied_chunks.iter() {
+                            chunk_recv.needed_chunks.insert(key.clone(), chunk.clone());
+                        }
+                    }
+                    Ok(())
+                });
+            }
+            while let Some(result) = task_set.join_next().await {
+                match result {
+                    Ok(task_res) => {
+                        if let Err(e) = task_res {
+                            error!("Error fetching chunk: {:?}", e);
+                        }
+                    }
+                    Err(e) => {
+                        error!("Error fetching chunk: {:?}", e);
+                    }
+                }
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+        }
+    }
+
+    async fn stop(self: Arc<Self>, _: GlobalState) {
+        self.stop.store(true, std::sync::atomic::Ordering::Relaxed);
+    }
+
+    fn name(&self) -> &'static str {
+        "Chunk Fetcher"
+    }
+}