Refacctored network and main thread channel handling to gracefully exit

on unrecoverable errors and properly handle recoverable ones.

Author: SubAt0m1c

src/main.rs

@@ -31,7 +31,7 @@ use crate::server::utils::dvec3::DVec3;
 use crate::server::world::VIEW_DISTANCE;
 use crate::utils::hasher::deterministic_hasher::DeterministicHashMap;
 use crate::utils::seeded_rng::SeededRng;
-use anyhow::Result;
+use anyhow::{bail, Result};
 use chrono::Local;
 use include_dir::include_dir;
 use indoc::formatdoc;
@@ -41,6 +41,7 @@ use std::env;
 use std::ops::Add;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::mpsc::unbounded_channel;
+use tokio::sync::mpsc::error::TryRecvError;
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -165,7 +166,7 @@ async fn main() -> Result<()> {
 
             // test
             pub struct MortImpl;
-            
+
             impl EntityImpl for MortImpl {
                 fn tick(&mut self, _: &mut Entity, _: &mut PacketBuffer) {
                     // rotate
@@ -177,7 +178,7 @@ async fn main() -> Result<()> {
                     player.open_ui(UI::MortReadyUpMenu);
                 }
             }
-            
+
             let id = server.world.spawn_entity(
                 room.get_world_block_pos(&BlockPos { x: 15, y: 69, z: 4 })
                     .as_dvec3()
@@ -225,10 +226,14 @@ async fn main() -> Result<()> {
         tick_interval.tick().await;
         // let start = std::time::Instant::now();
 
-        while let Ok(message) = main_rx.try_recv() {
-            server.process_event(message).unwrap_or_else(|err| eprintln!("Error processing event: {err}"));
+        loop {
+            match main_rx.try_recv() {
+                Ok(message) => server.process_event(message).unwrap_or_else(|err| eprintln!("Error processing event: {err}")),
+                Err(TryRecvError::Empty) => break,
+                Err(TryRecvError::Disconnected) => bail!("Network thread dropped its reciever."),
+            }
         }
-
+        
         server.dungeon.tick()?;
         server.world.tick()?;
 

src/net/client.rs

@@ -29,6 +29,9 @@ impl Client {
     }
 }
 
+// main thread tx errors can be ignored since the network/client threads will close eachother properly regardless of client status, 
+// which will in turn close the client handlers.
+
 pub async fn handle_client(
     client_id: ClientId,
     mut socket: TcpStream,

src/net/internal_packets.rs

@@ -18,6 +18,7 @@ pub enum NetworkThreadMessage {
     /// Disconnects the client from the server.
     /// This sends a close handler message to the client's handler.
     /// It should be sent after the vanilla disconnect packet is sent.
+    /// the main thread should wait for a ClientDisconnected response to handle actually removing the player.
     DisconnectClient {
         client_id: ClientId,
     },
@@ -40,11 +41,8 @@ pub enum MainThreadMessage {
         profile: GameProfile,
     },
 
+    /// sent to the main thread when a client is removed for any reason, even reasons caused by the main thread.
     ClientDisconnected {
         client_id: ClientId,
     },
-
-    Abort {
-        reason: String,
-    },
 }

src/net/run_network.rs

@@ -7,24 +7,22 @@ use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
 use crate::net::internal_packets::{ClientHandlerMessage, MainThreadMessage, NetworkThreadMessage};
 use crate::server::player::player::ClientId;
 
-/// runs the network thread. It is very important that nothing here panics without alerting the main thread.
-/// if a handle_client panics, it should send a disconnect player to the main thread. (however we dont have anything that could panic there atm)
+type ClientMap = HashMap<ClientId, UnboundedSender<ClientHandlerMessage>>;
+
 pub async fn run_network_thread(
     mut network_rx: UnboundedReceiver<NetworkThreadMessage>,
     network_tx: UnboundedSender<NetworkThreadMessage>,
     main_tx: UnboundedSender<MainThreadMessage>,
 ) {
-    let listener = TcpListener::bind("127.0.0.1:4972").await.unwrap_or_else(|err| {
-        let _ = main_tx.send(MainThreadMessage::Abort { reason: format!("TCP failed to bind: {}", err) });
-        panic!("{}", err)
-    });
+    let listener = TcpListener::bind("127.0.0.1:4972").await.unwrap();
     println!("Network thread listening on 127.0.0.1:4972");
 
-    let mut clients: HashMap<ClientId, UnboundedSender<ClientHandlerMessage>> = HashMap::new();
+    let mut clients: ClientMap = HashMap::new();
     let mut client_id_counter: ClientId = 1;
 
     loop {
         tokio::select! {
+            // a client failing to connect here is recoverable and doesnt really do anything, so we can just ignore it.
             Ok((socket, _)) = listener.accept() => {
                 let client_id: ClientId = client_id_counter;
                 client_id_counter += 1;
@@ -35,28 +33,39 @@ pub async fn run_network_thread(
                 tokio::spawn(handle_client(client_id, socket, client_rx, main_tx.clone(), network_tx.clone()));
             }
 
+            // this can never be none since this function owns a network_tx.
             Some(msg) = network_rx.recv() => {
-                match msg {
+                // we can just discard main thread -> network thread messages with a disconnected client_id
+                // as the main thread either already has or will be be informed shortly of this issue
+                match msg { 
                     NetworkThreadMessage::SendPackets { client_id, buffer } => {
                         if let Some(client_tx) = clients.get(&client_id) {
-                            let _ = client_tx.send(ClientHandlerMessage::Send(buffer));
+                            if let Err(e) = client_tx.send(ClientHandlerMessage::Send(buffer)) {
+                                eprintln!("Client {} handler dropped its reciever: {}", client_id, e);
+                                disconnect_client(client_id, &main_tx, &mut clients);
+                            }
                         }
                     }
-
+            
                     NetworkThreadMessage::DisconnectClient { client_id } => {
                         if let Some(client_tx) = clients.get(&client_id) {
-                            let _ = client_tx.send(ClientHandlerMessage::CloseHandler);
-                        } else {
-                            eprintln!("Attempted to disconnect nonexistent client {}", client_id);
+                            if let Err(e) = client_tx.send(ClientHandlerMessage::CloseHandler) {
+                                eprintln!("Client {} handler dropped its reciever: {}", client_id, e);
+                                disconnect_client(client_id, &main_tx, &mut clients);
+                            }
                         }
                     }
-
+            
                     NetworkThreadMessage::ConnectionClosed { client_id } => {
-                        let _ = main_tx.send(MainThreadMessage::ClientDisconnected { client_id });
-                        clients.remove(&client_id);
+                        disconnect_client(client_id, &main_tx, &mut clients);
                     }
                 }
             }
         }
     }
+}
+
+fn disconnect_client(client_id: ClientId, main_tx: &UnboundedSender<MainThreadMessage>, clients: &mut ClientMap) {
+    main_tx.send(MainThreadMessage::ClientDisconnected { client_id }).expect("Main thread should never drop its network reciever.");
+    clients.remove(&client_id);
 }

src/server/server.rs

@@ -1,7 +1,6 @@
 use crate::dungeon::dungeon::Dungeon;
 use crate::net::internal_packets::{MainThreadMessage, NetworkThreadMessage};
 use crate::net::packets::packet::ProcessPacket;
-use crate::net::packets::packet_serialize::PacketSerializable;
 use crate::net::protocol::play::clientbound::{AddEffect, CustomPayload, EntityProperties, JoinGame, PlayerAbilities, PlayerListHeaderFooter, PositionLook};
 use crate::net::var_int::VarInt;
 use crate::server::items::Item;
@@ -193,9 +192,6 @@ impl Server {
                 let player = self.world.players.get_mut(&client_id).context(format!("Player not found for id {client_id}"))?;
                 packet.process_with_player(player);
             },
-            MainThreadMessage::Abort { reason } => {
-                panic!("Network called for shutdown: {}", reason);
-            },
         }
         Ok(())
     }