From 28ada847437be5d8215d529ad48eff289dc32085 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 10 Apr 2021 17:10:04 +0200 Subject: add date and time to log output --- Cargo.lock | 34 ++++++++++++++++++++++++++++++++++ mumlib/Cargo.toml | 1 + mumlib/src/lib.rs | 3 ++- 3 files changed, 37 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 2e68fa5..ccb8871 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -160,6 +160,19 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "chrono" +version = "0.4.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "670ad68c9088c2a963aaa298cb369688cf3f9465ce5e2d4ca10e6e0098a1ce73" +dependencies = [ + "libc", + "num-integer", + "num-traits", + "time", + "winapi", +] + [[package]] name = "clang-sys" version = "1.1.1" @@ -885,6 +898,7 @@ dependencies = [ name = "mumlib" version = "0.4.0" dependencies = [ + "chrono", "colored", "dirs", "fern", @@ -998,6 +1012,16 @@ dependencies = [ "syn", ] +[[package]] +name = "num-integer" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2cc698a63b549a70bc047073d2949cce27cd1c7b0a4a862d08a8031bc2801db" +dependencies = [ + "autocfg", + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.14" @@ -1572,6 +1596,16 @@ dependencies = [ "syn", ] +[[package]] +name = "time" +version = "0.1.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca8a50ef2360fbd1eeb0ecd46795a87a19024eb4b53c5dc916ca1fd95fe62438" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "tokio" version = "1.4.0" diff --git a/mumlib/Cargo.toml b/mumlib/Cargo.toml index bda026b..5c9d4e1 100644 --- a/mumlib/Cargo.toml +++ b/mumlib/Cargo.toml @@ -13,6 +13,7 @@ readme = "../README.md" [dependencies] colored = "2" +chrono = "0.4" dirs = "3" fern = "0.6" log = "0.4" diff --git a/mumlib/src/lib.rs b/mumlib/src/lib.rs index 36edc10..9b7d686 100644 --- a/mumlib/src/lib.rs +++ b/mumlib/src/lib.rs @@ -16,7 +16,7 @@ pub fn setup_logger>(target: T, color: bool) { .format(move |out, message, record| { let message = message.to_string(); out.finish(format_args!( - "{} {}:{}{}{}", + "{} {} {}:{}{}{}", //TODO runtime flag that disables color if color { match record.level() { @@ -36,6 +36,7 @@ pub fn setup_logger>(target: T, color: bool) { } .normal() }, + chrono::Local::now().format("[%Y-%m-%d][%H:%M:%S%.6f]"), record.file().unwrap(), record.line().unwrap(), if message.chars().any(|e| e == '\n') { -- cgit v1.2.1 From cda329d361cdf4d1d13dbdb58e38c1e5e049762e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 10 Apr 2021 17:10:29 +0200 Subject: send parallel ping commands --- mumctl/src/main.rs | 58 +++++++++++++++++++++++++++++++++--------------------- 1 file changed, 36 insertions(+), 22 deletions(-) diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs index 29c9e44..07e5064 100644 --- a/mumctl/src/main.rs +++ b/mumctl/src/main.rs @@ -3,7 +3,11 @@ use log::*; use mumlib::command::{Command as MumCommand, CommandResponse}; use mumlib::config::{self, Config, ServerConfig}; use mumlib::state::Channel as MumChannel; -use std::{fmt,io::{self, BufRead, Read, Write}, iter, os::unix::net::UnixStream}; +use std::fmt; +use std::io::{self, BufRead, Read, Write}; +use std::iter; +use std::os::unix::net::UnixStream; +use std::thread; use structopt::{clap::Shell, StructOpt}; const INDENTATION: &str = " "; @@ -363,7 +367,7 @@ fn match_opt() -> Result<(), Error> { Ok(()) } -fn match_server_command(server_command: Server, config: &mut Config) -> Result<(), CliError> { +fn match_server_command(server_command: Server, config: &mut Config) -> Result<(), Error> { match server_command { Server::Config { server_name, @@ -496,29 +500,39 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( if config.servers.is_empty() { return Err(CliError::NoServers)?; } - let query = config + let queries: Vec<_> = config .servers .iter() - .map(|e| { - let response = send_command(MumCommand::ServerStatus { - host: e.host.clone(), - port: e.port.unwrap_or(mumlib::DEFAULT_PORT), - }); - response.map(|f| (e, f)) + .map(|s| { + let query = MumCommand::ServerStatus { + host: s.host.clone(), + port: s.port.unwrap_or(mumlib::DEFAULT_PORT), + }; + thread::spawn(move || { + send_command(query) + }) }) - .collect::, _>>()?; - for (server, response) in query - .into_iter() - .filter(|e| e.1.is_ok()) - .map(|e| (e.0, e.1.unwrap().unwrap())) - { - if let CommandResponse::ServerStatus { - users, max_users, .. - } = response - { - println!("{} [{}/{}]", server.name, users, max_users) - } else { - unreachable!() + .collect(); + for (server, response) in config.servers.iter().zip(queries) { + match response.join().unwrap() { + Ok(Ok(response)) => { + if let Some(CommandResponse::ServerStatus { + users, max_users, .. + }) = response + { + println!("{} [{}/{}]", server.name, users, max_users) + } else { + unreachable!() + } + } + Ok(Err(e)) => { + error!("{}", e); + return Err(e)?; + } + Err(e) => { + error!("{}", e); + return Err(e)?; + } } } } -- cgit v1.2.1 From 7070a9b345afb62f0da5d083a416b46c1efae409 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 10 Apr 2021 17:11:02 +0200 Subject: increment status ping id --- mumd/src/command.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/mumd/src/command.rs b/mumd/src/command.rs index 7eec388..04475aa 100644 --- a/mumd/src/command.rs +++ b/mumd/src/command.rs @@ -8,7 +8,7 @@ use crate::state::{ExecutionContext, State}; use log::*; use mumble_protocol::{Serverbound, control::ControlPacket}; use mumlib::command::{Command, CommandResponse}; -use std::sync::{Arc, RwLock}; +use std::sync::{atomic::{AtomicU64, Ordering}, Arc, RwLock}; use tokio::sync::{mpsc, oneshot, watch}; pub async fn handle( @@ -23,6 +23,7 @@ pub async fn handle( mut connection_info_sender: watch::Sender>, ) { debug!("Begin listening for commands"); + let ping_count = AtomicU64::new(0); while let Some((command, response_sender)) = command_receiver.recv().await { debug!("Received command {:?}", command); let mut state = state.write().unwrap(); @@ -47,10 +48,12 @@ pub async fn handle( response_sender.send(generator()).unwrap(); } ExecutionContext::Ping(generator, converter) => { - match generator() { + let ret = generator(); + match ret { Ok(addr) => { + let id = ping_count.fetch_add(1, Ordering::Relaxed); let res = ping_request_sender.send(( - 0, + id, addr, Box::new(move |packet| { response_sender.send(converter(packet)).unwrap(); -- cgit v1.2.1 From b6c4a90703def5fbd28e334f2c417c1b75627fd9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 10 Apr 2021 17:12:39 +0200 Subject: remove unneeded rc --- mumd/src/network/udp.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index cc085b5..94a1d11 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -3,6 +3,7 @@ use crate::network::ConnectionInfo; use crate::state::{State, StatePhase}; use futures_util::{FutureExt, SinkExt, StreamExt}; +use futures_util::future::join4; use futures_util::stream::{SplitSink, SplitStream, Stream}; use log::*; use mumble_protocol::crypt::ClientCryptState; @@ -12,16 +13,13 @@ use mumble_protocol::Serverbound; use std::collections::HashMap; use std::convert::TryFrom; use std::net::{Ipv6Addr, SocketAddr}; -use std::rc::Rc; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, RwLock}; +use std::sync::{atomic::{AtomicU64, Ordering}, Arc, RwLock}; use tokio::{join, net::UdpSocket}; use tokio::sync::{mpsc, watch, Mutex}; use tokio::time::{interval, Duration}; use tokio_util::udp::UdpFramed; use super::{run_until, VoiceStreamType}; -use futures_util::future::join4; pub type PingRequest = (u64, SocketAddr, Box); @@ -228,7 +226,7 @@ pub async fn handle_pings( .await .expect("Failed to bind UDP socket"); - let pending = Rc::new(Mutex::new(HashMap::new())); + let pending = Mutex::new(HashMap::new()); let sender_handle = async { while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await { -- cgit v1.2.1 From 76ac4ee0c993916b50e474a162e278b0a5844052 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 10 Apr 2021 17:14:15 +0200 Subject: add some debug prints for pings --- mumd/src/command.rs | 1 + mumd/src/network/udp.rs | 1 + 2 files changed, 2 insertions(+) diff --git a/mumd/src/command.rs b/mumd/src/command.rs index 04475aa..1337dce 100644 --- a/mumd/src/command.rs +++ b/mumd/src/command.rs @@ -49,6 +49,7 @@ pub async fn handle( } ExecutionContext::Ping(generator, converter) => { let ret = generator(); + debug!("Ping generated: {:?}", ret); match ret { Ok(addr) => { let id = ping_count.fetch_add(1, Ordering::Relaxed); diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index 94a1d11..d267007 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -230,6 +230,7 @@ pub async fn handle_pings( let sender_handle = async { while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await { + debug!("Sending ping {} to {}", id, socket_addr); let packet = PingPacket { id }; let packet: [u8; 12] = packet.into(); udp_socket.send_to(&packet, &socket_addr).await.unwrap(); -- cgit v1.2.1 From 7e848151aea0ad579acbd51125907d96cc67438b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 10 Apr 2021 19:28:37 +0200 Subject: timeout server pings --- mumctl/src/main.rs | 17 ++++++++----- mumd/src/command.rs | 4 +++- mumd/src/main.rs | 5 +++- mumd/src/network/udp.rs | 64 +++++++++++++++++++++++++++++++++++++++---------- mumd/src/state.rs | 2 +- 5 files changed, 70 insertions(+), 22 deletions(-) diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs index 07e5064..4ce7fc1 100644 --- a/mumctl/src/main.rs +++ b/mumctl/src/main.rs @@ -515,16 +515,21 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( .collect(); for (server, response) in config.servers.iter().zip(queries) { match response.join().unwrap() { - Ok(Ok(response)) => { - if let Some(CommandResponse::ServerStatus { - users, max_users, .. - }) = response + Ok(Ok(Some(response))) => { + if let CommandResponse::ServerStatus { + users, + max_users, + .. + } = response { - println!("{} [{}/{}]", server.name, users, max_users) + println!("{} [{}/{}]", server.name, users, max_users); } else { - unreachable!() + unreachable!(); } } + Ok(Ok(None)) => { + println!("{} offline", server.name); + } Ok(Err(e)) => { error!("{}", e); return Err(e)?; diff --git a/mumd/src/command.rs b/mumd/src/command.rs index 1337dce..e1c91ab 100644 --- a/mumd/src/command.rs +++ b/mumd/src/command.rs @@ -57,7 +57,9 @@ pub async fn handle( id, addr, Box::new(move |packet| { - response_sender.send(converter(packet)).unwrap(); + if let Some(packet) = packet { + response_sender.send(converter(packet)).unwrap(); + } }), )); if res.is_err() { diff --git a/mumd/src/main.rs b/mumd/src/main.rs index d7bc2c0..341300e 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -109,7 +109,10 @@ async fn receive_commands( sender.send((command, tx)).unwrap(); - let response = rx.await.unwrap(); + let response = match rx.await { + Ok(r) => r, + Err(_) => Ok(None), + }; let mut serialized = BytesMut::new(); bincode::serialize_into((&mut serialized).writer(), &response).unwrap(); diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index d267007..59620a3 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -10,18 +10,18 @@ use mumble_protocol::crypt::ClientCryptState; use mumble_protocol::ping::{PingPacket, PongPacket}; use mumble_protocol::voice::VoicePacket; use mumble_protocol::Serverbound; -use std::collections::HashMap; +use std::collections::{hash_map::Entry, HashMap}; use std::convert::TryFrom; use std::net::{Ipv6Addr, SocketAddr}; use std::sync::{atomic::{AtomicU64, Ordering}, Arc, RwLock}; -use tokio::{join, net::UdpSocket}; -use tokio::sync::{mpsc, watch, Mutex}; +use tokio::{join, net::UdpSocket, select}; +use tokio::sync::{mpsc, oneshot, watch, Mutex}; use tokio::time::{interval, Duration}; use tokio_util::udp::UdpFramed; use super::{run_until, VoiceStreamType}; -pub type PingRequest = (u64, SocketAddr, Box); +pub type PingRequest = (u64, SocketAddr, Box) + Send>); type UdpSender = SplitSink, (VoicePacket, SocketAddr)>; type UdpReceiver = SplitStream>; @@ -226,32 +226,70 @@ pub async fn handle_pings( .await .expect("Failed to bind UDP socket"); - let pending = Mutex::new(HashMap::new()); + let pending = Arc::new(Mutex::new(HashMap::new())); - let sender_handle = async { + let sender = async { while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await { debug!("Sending ping {} to {}", id, socket_addr); let packet = PingPacket { id }; let packet: [u8; 12] = packet.into(); udp_socket.send_to(&packet, &socket_addr).await.unwrap(); - pending.lock().await.insert(id, handle); + let (tx, rx) = oneshot::channel(); + match pending.lock().await.entry(id) { + Entry::Occupied(_) => { + warn!("Tried to send duplicate ping {}", id); + continue; + } + Entry::Vacant(v) => { + v.insert(tx); + } + } + + tokio::spawn(async move { + let rx = rx.fuse(); + let timeout = async { + tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; + }; + handle(select! { + r = rx => match r { + Ok(r) => Some(r), + Err(_) => { + warn!("Ping response sender dropped"); + None + } + }, + _ = timeout => None, + }); + }); } }; - let receiver_handle = async { + let receiver = async { let mut buf = vec![0; 24]; + while let Ok(read) = udp_socket.recv(&mut buf).await { - assert_eq!(read, 24); + if read != 24 { + warn!("Ping response had length {}, expected 24", read); + continue; + } + assert_eq!(read, 24); // just checked let packet = PongPacket::try_from(buf.as_slice()).unwrap(); - if let Some(handler) = pending.lock().await.remove(&packet.id) { - handler(packet); + match pending.lock().await.entry(packet.id) { + Entry::Occupied(o) => { + let id = *o.key(); + if o.remove().send(packet).is_err() { + debug!("Received response to ping {} too late", id); + } + } + Entry::Vacant(v) => { + warn!("Received ping {} that we didn't send", v.key()); + } } } }; debug!("Waiting for ping requests"); - - join!(sender_handle, receiver_handle); + join!(sender, receiver); } diff --git a/mumd/src/state.rs b/mumd/src/state.rs index b52b330..1f7331a 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -43,7 +43,7 @@ pub enum ExecutionContext { Now(Box mumlib::error::Result>>), Ping( Box mumlib::error::Result>, - Box mumlib::error::Result>>, + Box mumlib::error::Result> + Send>, ), } -- cgit v1.2.1 From dc318b332a0f75349744684f8b05ff188611cbc0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 00:29:09 +0200 Subject: align server status output --- mumctl/src/main.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs index 4ce7fc1..dbd62c6 100644 --- a/mumctl/src/main.rs +++ b/mumctl/src/main.rs @@ -500,6 +500,15 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( if config.servers.is_empty() { return Err(CliError::NoServers)?; } + + let longest = config + .servers + .iter() + .map(|s| s.name.len()) + .max() + .unwrap() // ok since !config.servers.is_empty() above + + 1; + let queries: Vec<_> = config .servers .iter() @@ -513,6 +522,7 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( }) }) .collect(); + for (server, response) in config.servers.iter().zip(queries) { match response.join().unwrap() { Ok(Ok(Some(response))) => { @@ -522,13 +532,13 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( .. } = response { - println!("{} [{}/{}]", server.name, users, max_users); + println!("{0:<1$} [{2:}/{3:}]", server.name, longest, users, max_users); } else { unreachable!(); } } Ok(Ok(None)) => { - println!("{} offline", server.name); + println!("{0:<1$} offline", server.name, longest); } Ok(Err(e)) => { error!("{}", e); -- cgit v1.2.1 From 1be49aaaab973eccf6f47cab36a4f3530143445a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 22:00:04 +0200 Subject: ? -> into --- mumctl/src/main.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs index dbd62c6..5abed50 100644 --- a/mumctl/src/main.rs +++ b/mumctl/src/main.rs @@ -301,7 +301,7 @@ fn match_opt() -> Result<(), Error> { } } _ => { - return Err(CliError::ConfigKeyNotFound(key))?; + return Err(CliError::ConfigKeyNotFound(key).into()); } }, Command::ConfigReload => { @@ -477,7 +477,7 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( password, } => { if config.servers.iter().any(|s| s.name == name) { - return Err(CliError::ServerAlreadyExists(name))?; + return Err(CliError::ServerAlreadyExists(name).into()); } else { config.servers.push(ServerConfig { name, @@ -498,7 +498,7 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( } Server::List => { if config.servers.is_empty() { - return Err(CliError::NoServers)?; + return Err(CliError::NoServers.into()); } let longest = config @@ -542,11 +542,11 @@ fn match_server_command(server_command: Server, config: &mut Config) -> Result<( } Ok(Err(e)) => { error!("{}", e); - return Err(e)?; + return Err(e.into()); } Err(e) => { error!("{}", e); - return Err(e)?; + return Err(e.into()); } } } -- cgit v1.2.1 From dfe6cea0739fe162be95ab0023aef2aa21c94416 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 22:52:01 +0200 Subject: add error log --- mumd/src/main.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/mumd/src/main.rs b/mumd/src/main.rs index 341300e..f298070 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -111,7 +111,10 @@ async fn receive_commands( let response = match rx.await { Ok(r) => r, - Err(_) => Ok(None), + Err(_) => { + error!("Internal command response sender dropped"); + Ok(None) + } }; let mut serialized = BytesMut::new(); bincode::serialize_into((&mut serialized).writer(), &response).unwrap(); -- cgit v1.2.1 From c8e92b3fe0ec73cd0d87e778ee85eac9a0d6c4e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 22:52:21 +0200 Subject: remove arc --- mumd/src/network/udp.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index 59620a3..808d853 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -226,7 +226,7 @@ pub async fn handle_pings( .await .expect("Failed to bind UDP socket"); - let pending = Arc::new(Mutex::new(HashMap::new())); + let pending = Mutex::new(HashMap::new()); let sender = async { while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await { -- cgit v1.2.1 From 30264e1fa0e2602d29141409ecc8326bce1917aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 22:52:41 +0200 Subject: correct output --- mumd/src/network/udp.rs | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index 808d853..d8cbc2a 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -230,14 +230,14 @@ pub async fn handle_pings( let sender = async { while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await { - debug!("Sending ping {} to {}", id, socket_addr); + debug!("Sending ping with id {} to {}", id, socket_addr); let packet = PingPacket { id }; let packet: [u8; 12] = packet.into(); udp_socket.send_to(&packet, &socket_addr).await.unwrap(); let (tx, rx) = oneshot::channel(); match pending.lock().await.entry(id) { Entry::Occupied(_) => { - warn!("Tried to send duplicate ping {}", id); + warn!("Tried to send duplicate ping with id {}", id); continue; } Entry::Vacant(v) => { @@ -272,7 +272,6 @@ pub async fn handle_pings( warn!("Ping response had length {}, expected 24", read); continue; } - assert_eq!(read, 24); // just checked let packet = PongPacket::try_from(buf.as_slice()).unwrap(); @@ -280,11 +279,11 @@ pub async fn handle_pings( Entry::Occupied(o) => { let id = *o.key(); if o.remove().send(packet).is_err() { - debug!("Received response to ping {} too late", id); + debug!("Received response to ping with id {} too late", id); } } Entry::Vacant(v) => { - warn!("Received ping {} that we didn't send", v.key()); + warn!("Received ping with id {} that we didn't send", v.key()); } } } -- cgit v1.2.1 From 6a0349287066261d7953b4220de5dd4df05049c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 22:53:04 +0200 Subject: use tokio timeout --- mumd/src/network/udp.rs | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index d8cbc2a..2ca8f85 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -14,9 +14,9 @@ use std::collections::{hash_map::Entry, HashMap}; use std::convert::TryFrom; use std::net::{Ipv6Addr, SocketAddr}; use std::sync::{atomic::{AtomicU64, Ordering}, Arc, RwLock}; -use tokio::{join, net::UdpSocket, select}; +use tokio::{join, net::UdpSocket}; use tokio::sync::{mpsc, oneshot, watch, Mutex}; -use tokio::time::{interval, Duration}; +use tokio::time::{interval, timeout, Duration}; use tokio_util::udp::UdpFramed; use super::{run_until, VoiceStreamType}; @@ -246,20 +246,19 @@ pub async fn handle_pings( } tokio::spawn(async move { - let rx = rx.fuse(); - let timeout = async { - tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; - }; - handle(select! { - r = rx => match r { - Ok(r) => Some(r), + handle( + match timeout(Duration::from_secs(1), rx).await { + Ok(Ok(r)) => Some(r), + Ok(Err(_)) => { + warn!("Ping response sender for server {}, ping id {} dropped", socket_addr, id); + None + } Err(_) => { - warn!("Ping response sender dropped"); + debug!("Server {} timed out when sending ping id {}", socket_addr, id); None } - }, - _ = timeout => None, - }); + } + ); }); } }; -- cgit v1.2.1 From 4da882513c9a692161ff00e4421325ffc7d4af24 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sun, 11 Apr 2021 22:53:11 +0200 Subject: always send response to server list --- mumd/src/command.rs | 4 +--- mumd/src/state.rs | 4 ++-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/mumd/src/command.rs b/mumd/src/command.rs index e1c91ab..1337dce 100644 --- a/mumd/src/command.rs +++ b/mumd/src/command.rs @@ -57,9 +57,7 @@ pub async fn handle( id, addr, Box::new(move |packet| { - if let Some(packet) = packet { - response_sender.send(converter(packet)).unwrap(); - } + response_sender.send(converter(packet)).unwrap(); }), )); if res.is_err() { diff --git a/mumd/src/state.rs b/mumd/src/state.rs index 1f7331a..b6e2abb 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -43,7 +43,7 @@ pub enum ExecutionContext { Now(Box mumlib::error::Result>>), Ping( Box mumlib::error::Result>, - Box mumlib::error::Result> + Send>, + Box) -> mumlib::error::Result> + Send>, ), } @@ -387,7 +387,7 @@ impl State { } }), Box::new(move |pong| { - Ok(Some(CommandResponse::ServerStatus { + Ok(pong.map(|pong| CommandResponse::ServerStatus { version: pong.version, users: pong.users, max_users: pong.max_users, -- cgit v1.2.1