bunch of changes to prevent deadlocking the server

src/server/mod.rs

@@ -1,11 +1,12 @@
 use std::net::SocketAddr;
-use std::sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}};
+use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
 use std::time::Instant;
 use std::collections::HashMap;
 
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, UdpSocket};
 use tokio::task::{JoinHandle, JoinSet};
+use tokio::sync::{Mutex, mpsc};
 
 use num_enum::IntoPrimitive;
 
@@ -126,7 +127,7 @@ pub struct Server {
     tcp_listener: Arc<TcpListener>,
     pub udp_socket: Arc<UdpSocket>,
 
-    clients_incoming: Arc<Mutex<Vec<Client>>>,
+    clients_incoming_rx: mpsc::Receiver<Client>,
     clients_queue: Vec<(Client, Vec<tokio::sync::oneshot::Receiver<Argument>>, Vec<Argument>)>,
 
     pub clients: Vec<Client>,
@@ -161,25 +162,28 @@ impl Server {
         let plugins = load_plugins();
 
         // Start client runtime
-        let clients_incoming = Arc::new(Mutex::new(Vec::new()));
+        let (clients_incoming_tx, clients_incoming_rx) = mpsc::channel(100);
-        let clients_incoming_ref = Arc::clone(&clients_incoming);
         debug!("Client acception runtime starting...");
         let connect_runtime_handle = tokio::spawn(async move {
             let mut set = JoinSet::new();
             loop {
-                match tcp_listener_ref.accept().await {
+                tokio::select! {
+                    new_conn = tcp_listener_ref.accept() => {
+                        match new_conn {
                             Ok((mut socket, addr)) => {
                                 info!("New client connected: {:?}", addr);
 
                                 let cfg_ref = config_ref.clone();
-                        let ci_ref = clients_incoming_ref.clone();
+                                let ci_ref = clients_incoming_tx.clone();
 
                                 set.spawn(async move {
                                     socket.set_nodelay(true); // TODO: Is this good?
 
                                     socket.readable().await.expect("Failed to wait for socket to become readable!");
                                     let mut tmp = vec![0u8; 1];
-                            while socket.peek(&mut tmp).await.expect("Failed to peek socket!") == 0 {}
+                                    while socket.peek(&mut tmp).await.expect("Failed to peek socket!") == 0 {
+                                        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+                                    }
                                     // Authentication works a little differently than normal
                                     // Not sure why, but the BeamMP source code shows they
                                     // also only read a single byte during authentication
@@ -188,15 +192,12 @@ impl Server {
 
                                     match code as char {
                                         'C' => {
+                                            info!("hi");
                                             let mut client = Client::new(socket).await;
                                             match client.authenticate(&cfg_ref).await {
                                                 Ok(is_client) if is_client => {
-                                            let mut lock = ci_ref
+                                                    info!("bye");
-                                                .lock()
+                                                    ci_ref.send(client).await;
-                                                .map_err(|e| error!("{:?}", e))
-                                                .expect("Failed to acquire lock on mutex!");
-                                            lock.push(client);
-                                            drop(lock);
                                                 },
                                                 Ok(_is_client) => {
                                                     debug!("Downloader?");
@@ -278,20 +279,33 @@ impl Server {
                                                 trace!("Unknown G packet received, not sure what to do!");
                                             }
                                         },
-                                _ => {},
+                                        _ => {
+                                            tokio::task::yield_now().await;
+                                        },
                                     };
                                 });
+                                info!("Client pushed to joinset!");
                             }
                             Err(e) => error!("Failed to accept incoming connection: {:?}", e),
                         }
+                    },
+                    _ = tokio::time::sleep(tokio::time::Duration::from_millis(50)) => {
+                        error!("time out!");
+                    },
+                }
 
                 if set.is_empty() == false {
+                    info!("what!");
                     // Because join_next() is cancel safe, we can simply cancel it after N duration
                     // so at worst this client acceptance loop blocks for N duration
-                    tokio::select!(
+                    tokio::select! {
-                        _ = set.join_next() => {},
+                        _ = tokio::time::sleep(tokio::time::Duration::from_millis(10)) => {
-                        _ = tokio::time::sleep(tokio::time::Duration::from_millis(10)) => {},
+                            error!("join_next timed out!");
-                    )
+                        },
+                        _ = set.join_next() => {
+                            info!("join_next ran!");
+                        },
+                    }
                 }
             }
         });
@@ -301,7 +315,7 @@ impl Server {
             tcp_listener: tcp_listener,
             udp_socket: udp_socket,
 
-            clients_incoming: clients_incoming,
+            clients_incoming_rx,
             clients_queue: Vec::new(),
 
             clients: Vec::new(),
@@ -375,13 +389,8 @@ impl Server {
         // with the client acception runtime. If that one locks, the server won't accept
         // more clients, but it will at least still process all other clients
         let mut joined_names = Vec::new();
-        if let Ok(mut clients_incoming_lock) = self.clients_incoming.try_lock() { // TODO: Why do I use try_lock here?
+        match self.clients_incoming_rx.try_recv() {
-            if clients_incoming_lock.len() > 0 {
+            Ok(client) => {
-                trace!(
-                    "Accepting {} incoming clients...",
-                    clients_incoming_lock.len()
-                );
-                for client in clients_incoming_lock.drain(..) {
                 let userdata = client.get_userdata();
                 let (name, role, is_guest, beammp_id) = (userdata.username.clone(), userdata.roles.clone(), userdata.guest, userdata.uid.clone());
                 info!("Welcome {name}!");
@@ -400,9 +409,9 @@ impl Server {
                     vrx.push(rx);
                 }
                 self.clients_queue.push((client, vrx, Vec::new()));
-                }
+            },
-                trace!("Accepted incoming clients!");
+            Err(mpsc::error::TryRecvError::Empty) => {},
-            }
+            Err(e) => error!("Error while receiving new clients from acception runtime: {:?}", e),
         }
 
         // Bit scuffed but it just polls the return values until all lua plugins have returned