GameCult/Odin
Watch
9
0
Fork
You've already forked Odin
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Pump RUDP acknowledgements under media backpressure

Browse source
This commit is contained in:
Meta Vens 2026-07-17 00:33:07 +01:00
commit b86eb8b6c8
3 changed files with 111 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

15
crates/cultnet-impair/src/main.rs
View file

@ -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));
}
}
}

105
crates/muninn-daemon/src/main.rs
View file

@ -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<QueuedMuninnMediaSendPayload>,
video: VecDeque<QueuedMuninnMediaSendPayload>,
active_audio: VecDeque<QueuedMuninnMediaSendPayload>,
audio: VecDeque<VecDeque<QueuedMuninnMediaSendPayload>>,
active_video: VecDeque<QueuedMuninnMediaSendPayload>,
video: VecDeque<VecDeque<QueuedMuninnMediaSendPayload>>,
}
impl PendingMuninnMediaSendQueues {
fn push(&mut self, payload: QueuedMuninnMediaSendPayload) {
match payload.kind {
self.push_group(vec![payload]);
}
fn push_group(&mut self, payloads: Vec<QueuedMuninnMediaSendPayload>) {
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<QueuedMuninnMediaSendPayload> {
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<QueuedMuninnMediaSendPayload>,
queued: &mut VecDeque<VecDeque<QueuedMuninnMediaSendPayload>>,
) -> Option<QueuedMuninnMediaSendPayload> {
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::<usize>()
}
fn video_len(&self) -> usize {
self.video.len()
self.active_video.len() + self.video.iter().map(VecDeque::len).sum::<usize>()
}
}
@ -2638,9 +2672,7 @@ fn receive_pending_media_payloads(
) -> Result<bool> {
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<bool> {
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::<Vec<_>>()
};
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);

22
vendor/cultnet-rs/src/rudp.rs vendored
View file

@ -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<bool> {
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<()> {