From b86eb8b6c8a78a463058fac99e98ecc5d2d82695 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 17 Jul 2026 00:33:07 +0100 Subject: [PATCH] Pump RUDP acknowledgements under media backpressure --- crates/cultnet-impair/src/main.rs | 15 ++++- crates/muninn-daemon/src/main.rs | 105 ++++++++++++++++++++++++------ vendor/cultnet-rs/src/rudp.rs | 22 ++++--- 3 files changed, 111 insertions(+), 31 deletions(-) diff --git a/crates/cultnet-impair/src/main.rs b/crates/cultnet-impair/src/main.rs index e2ce6ba..d60cbd4 100644 --- a/crates/cultnet-impair/src/main.rs +++ b/crates/cultnet-impair/src/main.rs @@ -10,6 +10,7 @@ use std::time::{Duration, Instant}; const MAX_DATAGRAM: usize = 65_535; const MAX_CLIENT_FLOWS: usize = 64; const CLIENT_FLOW_IDLE_TIMEOUT_MS: u64 = 30_000; +const MAX_DATAGRAMS_PER_FLOW_PER_TURN: usize = 16; #[derive(Clone, Debug, PartialEq, Eq)] struct Profile { @@ -458,9 +459,11 @@ fn run(options: Options) -> Result<(), String> { let mut buffer = vec![0; MAX_DATAGRAM]; loop { let now = elapsed_ms(start); - loop { + let mut did_work = false; + for _ in 0..MAX_DATAGRAMS_PER_FLOW_PER_TURN { match client_socket.recv_from(&mut buffer) { Ok((size, address)) => { + did_work = true; ensure_upstream_flow( &mut flows, address, @@ -484,9 +487,10 @@ fn run(options: Options) -> Result<(), String> { } } for (address, flow) in &mut flows { - loop { + for _ in 0..MAX_DATAGRAMS_PER_FLOW_PER_TURN { match flow.socket.recv(&mut buffer) { Ok(size) => { + did_work = true; flow.last_activity_ms = now; scheduler.admit( now, @@ -501,6 +505,7 @@ fn run(options: Options) -> Result<(), String> { } } while let Some(item) = scheduler.pop_due(now) { + did_work = true; let result = match item.direction { Direction::ClientToUpstream => match flows.get_mut(&item.client) { Some(flow) => { @@ -527,7 +532,11 @@ fn run(options: Options) -> Result<(), String> { reap_idle_flows(&mut flows, now, &mut scheduler.stats); last_flow_reap_at = now; } - thread::sleep(Duration::from_millis(1)); + if did_work { + thread::yield_now(); + } else { + thread::sleep(Duration::from_millis(1)); + } } } diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index 7721097..47332f0 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -79,7 +79,8 @@ const MUNINN_RUDP_FIXED_HEADER_BYTES: usize = 36; const MUNINN_RUDP_MEDIA_MAX_FRAGMENT_BYTES: usize = MUNINN_RUDP_IPV4_UDP_PAYLOAD_BYTES - MUNINN_RUDP_FIXED_HEADER_BYTES - crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL.len(); -const MUNINN_RUDP_MEDIA_RESEND_DELAY_MS: u64 = 5; +const MUNINN_RUDP_MEDIA_RESEND_DELAY_MS: u64 = 8; +const MUNINN_RUDP_MEDIA_MAX_PENDING_RELIABLE_PACKETS: u32 = 256; const MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS: u64 = 100; const MUNINN_RUDP_MEDIA_PAYLOAD_CHANNEL_CAPACITY: usize = 1; const MUNINN_RUDP_MEDIA_PENDING_AUDIO_CAPACITY: usize = 256; @@ -2301,42 +2302,75 @@ enum QueuedMuninnMediaKind { #[derive(Default)] struct PendingMuninnMediaSendQueues { - audio: VecDeque, - video: VecDeque, + active_audio: VecDeque, + audio: VecDeque>, + active_video: VecDeque, + video: VecDeque>, } impl PendingMuninnMediaSendQueues { fn push(&mut self, payload: QueuedMuninnMediaSendPayload) { - match payload.kind { + self.push_group(vec![payload]); + } + + fn push_group(&mut self, payloads: Vec) { + let Some(kind) = payloads.first().map(|payload| payload.kind) else { + return; + }; + debug_assert!(payloads.iter().all(|payload| payload.kind == kind)); + let group = VecDeque::from(payloads); + match kind { QueuedMuninnMediaKind::Audio => { - while self.audio.len() >= MUNINN_RUDP_MEDIA_PENDING_AUDIO_CAPACITY { + while self.audio_len().saturating_add(group.len()) + > MUNINN_RUDP_MEDIA_PENDING_AUDIO_CAPACITY + && !self.audio.is_empty() + { self.audio.pop_front(); } - self.audio.push_back(payload); + self.audio.push_back(group); } QueuedMuninnMediaKind::Video => { - while self.video.len() >= MUNINN_RUDP_MEDIA_PENDING_VIDEO_CAPACITY { + while self.video_len().saturating_add(group.len()) + > MUNINN_RUDP_MEDIA_PENDING_VIDEO_CAPACITY + && !self.video.is_empty() + { self.video.pop_front(); } - self.video.push_back(payload); + self.video.push_back(group); } } } fn pop_next(&mut self) -> Option { - self.audio.pop_front().or_else(|| self.video.pop_front()) + Self::pop_from(&mut self.active_audio, &mut self.audio) + .or_else(|| Self::pop_from(&mut self.active_video, &mut self.video)) + } + + fn pop_from( + active: &mut VecDeque, + queued: &mut VecDeque>, + ) -> Option { + if active.is_empty() + && let Some(group) = queued.pop_front() + { + *active = group; + } + active.pop_front() } fn is_empty(&self) -> bool { - self.audio.is_empty() && self.video.is_empty() + self.active_audio.is_empty() + && self.audio.is_empty() + && self.active_video.is_empty() + && self.video.is_empty() } fn audio_len(&self) -> usize { - self.audio.len() + self.active_audio.len() + self.audio.iter().map(VecDeque::len).sum::() } fn video_len(&self) -> usize { - self.video.len() + self.active_video.len() + self.video.iter().map(VecDeque::len).sum::() } } @@ -2638,9 +2672,7 @@ fn receive_pending_media_payloads( ) -> Result { match rx.recv_timeout(timeout) { Ok(Ok(payloads)) => { - for payload in payloads { - pending.push(payload); - } + pending.push_group(payloads); drain_available_media_payloads(rx, pending) } Ok(Err(error)) => Err(error), @@ -2655,11 +2687,7 @@ fn drain_available_media_payloads( ) -> Result { for _ in 0..MUNINN_RUDP_MEDIA_INGEST_BUDGET_PER_TURN { match rx.try_recv() { - Ok(Ok(payloads)) => { - for payload in payloads { - pending.push(payload); - } - } + Ok(Ok(payloads)) => pending.push_group(payloads), Ok(Err(error)) => return Err(error), Err(mpsc::TryRecvError::Empty) => return Ok(false), Err(mpsc::TryRecvError::Disconnected) => return Ok(true), @@ -2699,6 +2727,11 @@ fn send_rudp_media_payload_with_backpressure( if media_payload_queue_age_exceeded(queued_at, Instant::now(), max_age) { return Ok(false); } + for _ in 0..MUNINN_RUDP_MEDIA_MAX_PENDING_RELIABLE_PACKETS { + if !transport.poll_receive_once()? { + break; + } + } poll_rudp_resends_with_backpressure(transport)?; thread::sleep(Duration::from_millis(1)); } @@ -3051,6 +3084,7 @@ fn muninn_media_rudp_options( CultNetRudpSocketTransportOptions::client("muninn-media", socket, endpoint, connection_id); options.resend_delay_ms = media_profile.sender_resend_delay_ms; options.max_fragment_bytes = Some(media_profile.max_fragment_bytes as u32); + options.max_pending_reliable_packets = Some(MUNINN_RUDP_MEDIA_MAX_PENDING_RELIABLE_PACKETS); options } @@ -11901,6 +11935,10 @@ mod tests { options.max_fragment_bytes, Some(MUNINN_RUDP_MEDIA_MAX_FRAGMENT_BYTES as u32) ); + assert_eq!( + options.max_pending_reliable_packets, + Some(MUNINN_RUDP_MEDIA_MAX_PENDING_RELIABLE_PACKETS) + ); assert_eq!(options.reconnect_policy, None); let transport = CultNetRudpSocketTransportConnection::new(options).unwrap(); let channels = transport @@ -12055,6 +12093,33 @@ mod tests { ); } + #[test] + fn started_video_access_unit_cannot_be_partially_evicted() { + let queued_at = Instant::now(); + let video_group = |marker: u8, count: usize| { + (0..count) + .map(|_| QueuedMuninnMediaSendPayload { + payload: MuninnMediaSendPayload { + channel_id: crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL, + payload: vec![marker], + }, + queued_at, + kind: QueuedMuninnMediaKind::Video, + }) + .collect::>() + }; + let mut pending = PendingMuninnMediaSendQueues::default(); + pending.push_group(video_group(1, 64)); + assert_eq!(pending.pop_next().unwrap().payload.payload, vec![1]); + + pending.push_group(video_group(2, MUNINN_RUDP_MEDIA_PENDING_VIDEO_CAPACITY)); + + for _ in 1..64 { + assert_eq!(pending.pop_next().unwrap().payload.payload, vec![1]); + } + assert_eq!(pending.pop_next().unwrap().payload.payload, vec![2]); + } + #[test] fn rudp_media_ingest_yields_to_send_after_a_bounded_turn() { let (tx, rx) = mpsc::sync_channel(MUNINN_RUDP_MEDIA_INGEST_BUDGET_PER_TURN * 2); diff --git a/vendor/cultnet-rs/src/rudp.rs b/vendor/cultnet-rs/src/rudp.rs index 8b47f9e..2692bb8 100644 --- a/vendor/cultnet-rs/src/rudp.rs +++ b/vendor/cultnet-rs/src/rudp.rs @@ -1004,13 +1004,19 @@ impl CultNetRudpSocketTransportConnection { return Ok(Some(frame)); } + self.poll_receive_once()?; + Ok(self.delivered_frames.pop_front()) + } + + pub fn poll_receive_once(&mut self) -> Result { + let mut wire = vec![0_u8; 65_535]; let (received, remote_addr) = match self.socket.recv_from(&mut wire) { Ok(value) => value, Err(error) if error.kind() == ErrorKind::WouldBlock || error.kind() == ErrorKind::TimedOut => { - return Ok(None); + return Ok(false); } Err(error) => return Err(error.into()), }; @@ -1025,14 +1031,14 @@ impl CultNetRudpSocketTransportConnection { { self.remote_addr = Some(remote_addr); } else { - return Ok(None); + return Ok(true); } } } else { if self.mode == CultNetRudpSocketMode::Server && packet.packet_type != CultNetRudpPacketType::Connect { - return Ok(None); + return Ok(true); } self.remote_addr = Some(remote_addr); } @@ -1042,7 +1048,7 @@ impl CultNetRudpSocketTransportConnection { self.session.reset_peer_state(); let accept = self.session.accept_connect(&packet, now_ms(), Vec::new())?; self.send_packet(&accept)?; - return Ok(None); + return Ok(true); } let result = self.session.receive(&packet, now_ms())?; @@ -1054,9 +1060,10 @@ impl CultNetRudpSocketTransportConnection { } if result.disconnected { self.disconnect_reason = Some(result.disconnect_reason); - return Ok(None); + return Ok(true); } + let delivered_any = !result.delivered.is_empty(); for frame in result.delivered { self.delivered_frames.push_back(CultNetTransportFrame { channel_id: frame.channel_id, @@ -1064,12 +1071,11 @@ impl CultNetRudpSocketTransportConnection { }); self.stats.frames_received += 1; } - let frame = self.delivered_frames.pop_front(); - if packet.packet_type == CultNetRudpPacketType::Accept || frame.is_some() { + if packet.packet_type == CultNetRudpPacketType::Accept || delivered_any { let ack = self.session.create_ack(); self.send_packet(&ack)?; } - Ok(frame) + Ok(true) } pub fn poll_resends(&mut self) -> Result<()> {