Less terrible chunk sending

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"
@@ -145,6 +148,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

@@ -27,6 +27,8 @@ async fn main() {
     let cli_args = CLIArgs::parse();
     ferrumc_logging::init_logging(cli_args.log.into());
 
+    check_deadlocks();
+
     match cli_args.command {
         Some(Command::Setup) => {
             info!("Starting setup...");
@@ -107,3 +109,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,3 +1,4 @@
+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,10 +39,13 @@ async fn handle_login_start(
     debug!("Received login start from user with username {}", username);
 
     // 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),
-    )?;
+    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
@@ -168,6 +172,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,46 @@ 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,96 @@
+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>();
+            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 = {
+                        trace!("Getting chunk_recv 1 for fetcher");
+                        let chunk_recv = state
+                            .universe
+                            .get::<ChunkReceiver>(eid)
+                            .expect("ChunkReceiver not found");
+                        trace!("Got chunk_recv 1 for fetcher");
+                        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
+                    {
+                        trace!("Getting chunk_recv 2 for fetcher");
+                        let Ok(chunk_recv) = state.universe.get::<ChunkReceiver>(eid) else {
+                            trace!("A player disconnected before we could get the ChunkReceiver");
+                            return Ok(());
+                        };
+                        trace!("Got chunk_recv 2 for fetcher");
+                        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"
+    }
+}