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(-) (limited to 'mumd/src') 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(-) (limited to 'mumd/src') 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(+) (limited to 'mumd/src') 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 --- mumd/src/command.rs | 4 +++- mumd/src/main.rs | 5 +++- mumd/src/network/udp.rs | 64 +++++++++++++++++++++++++++++++++++++++---------- mumd/src/state.rs | 2 +- 4 files changed, 59 insertions(+), 16 deletions(-) (limited to 'mumd/src') 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 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(-) (limited to 'mumd/src') 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(-) (limited to 'mumd/src') 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(-) (limited to 'mumd/src') 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(-) (limited to 'mumd/src') 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(-) (limited to 'mumd/src') 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