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/network/udp.rs') 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/network/udp.rs | 1 + 1 file changed, 1 insertion(+) (limited to 'mumd/src/network/udp.rs') 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/network/udp.rs | 64 +++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 51 insertions(+), 13 deletions(-) (limited to 'mumd/src/network/udp.rs') 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); } -- 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/network/udp.rs') 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/network/udp.rs') 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/network/udp.rs') 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