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

Bound the media sender: one deadline owner, whole-group drops, a drivable seam

Browse source 
The sender's queues were unbounded (mpsc::channel plus two VecDeques), expired
payloads one at a time so half a frame could leave after the other half died,
gave repairs a fresh deadline from the clock, spun on a 1 ms sleep, and could
not be driven without PowerShell, ffmpeg and WASAPI. The latency budget also had
two owners: a 2 s producer default that the receiver's 250 ms age never matched.

MediaSendCore (new media_send.rs) owns the queues, hub, feedback intake, repair
cache and budget, and progress counters. run_rudp_mux_once now spawns encoders,
feeds readers into the core and publishes; it decides no deadline.

- The unit is a group: one access unit with its parity, or one audio packet,
  stamped produced_at + the request's latency budget when it leaves the
  packetizer. Expiry is per group; a group is sent whole or not at all.
- Queues are capped per kind (audio 256, video 512 groups), drop the oldest
  group and count it (groups_dropped_audio, groups_dropped_video). The reader
  channel is a bounded sync_channel.
- The Annex B packetizer returns one group per access unit and no longer errors
  on pending overflow: an unfinished access unit larger than the ceiling is
  dropped and counted.
- Repairs inherit the deadline of their frame and are not sent after it. A
  blocked send polls resends and stops at the group deadline; the sleep is gone.
- MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS is deleted. The default budget
  is 250 ms (Q7), used only when a request names none and offered as the
  advertisement's default_latency_budget_ms. The profile loses its three copies
  of the budget; the telemetry JSON carries one latency_budget_ms.
- A test harness drives the real readers and core over loopback through a
  seeded drop relay to a CultNet RUDP client, from a committed H.264 clip
  (testdata/clip.h264, ffmpeg command recorded beside it) and a PCM ramp.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 20:04:17 +02:00
commit e66c2565fb
6 changed files with 2064 additions and 1128 deletions
Download patch file Download diff file Expand all files Collapse all files

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

File diff suppressed because it is too large Load diff

143
crates/muninn-daemon/src/media_packetizer.rs
View file

@ -213,11 +213,17 @@ pub struct MuninnMediaSendPayload {
pub payload: Vec<u8>,
}
/// Everything one video access unit puts on the wire: its chunks and, until
/// the FEC cut replaces it, their parity. A group is sent whole or not at
/// all, because parity for a frame that never completes is wasted bandwidth.
pub type MuninnMediaPayloadGroup = Vec<MuninnMediaSendPayload>;
pub struct VideoAnnexBStreamSendState {
config: VideoAnnexBStreamSendConfig,
pending: Vec<u8>,
next_frame_id: u64,
next_pts_ticks: i64,
dropped_access_units: u64,
}
impl VideoAnnexBStreamSendState {
@ -260,26 +266,35 @@ impl VideoAnnexBStreamSendState {
next_pts_ticks: config.first_pts_ticks,
config,
pending: Vec::new(),
dropped_access_units: 0,
})
}
pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result<Vec<MuninnMediaSendPayload>> {
/// One group per completed access unit. Whatever is still pending after
/// that is a single unfinished access unit; when it outgrows
/// `max_pending_bytes` it is dropped and counted, never an error that
/// would restart the stream.
pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result<Vec<MuninnMediaPayloadGroup>> {
if !bytes.is_empty() {
self.pending.extend_from_slice(bytes);
}
let groups = self.emit_available(stored_at, false)?;
if self.pending.len() > self.config.max_pending_bytes {
return Err(anyhow!(
"Annex B stream sender pending buffer exceeded {} bytes without a complete frame",
self.config.max_pending_bytes
));
self.pending.clear();
self.dropped_access_units += 1;
}
self.emit_available(stored_at, false)
Ok(groups)
}
pub fn finish(&mut self, stored_at: &str) -> Result<Vec<MuninnMediaSendPayload>> {
pub fn finish(&mut self, stored_at: &str) -> Result<Vec<MuninnMediaPayloadGroup>> {
self.emit_available(stored_at, true)
}
/// Access units dropped for outgrowing `max_pending_bytes`, cumulative.
pub fn dropped_access_units(&self) -> u64 {
self.dropped_access_units
}
pub fn pending_bytes(&self) -> usize {
self.pending.len()
}
@ -292,10 +307,17 @@ impl VideoAnnexBStreamSendState {
&mut self,
stored_at: &str,
flush_last: bool,
) -> Result<Vec<MuninnMediaSendPayload>> {
) -> Result<Vec<MuninnMediaPayloadGroup>> {
if stored_at.is_empty() {
return Err(anyhow!("stored_at must be non-empty"));
}
// After a drop the tail of the dropped access unit arrives with no
// start code in front of it; skip to the next one.
if let Some((start, _)) = find_annex_b_start_code(&self.pending, 0)
&& start > 0
{
self.pending.drain(..start);
}
if self.pending.is_empty() {
return Ok(Vec::new());
}
@ -319,7 +341,7 @@ impl VideoAnnexBStreamSendState {
return Ok(Vec::new());
}
let mut send_payloads = Vec::new();
let mut groups = Vec::with_capacity(emit_count);
let mut emitted_bytes = 0_usize;
for access_unit in access_units.iter().take(emit_count) {
emitted_bytes = emitted_bytes
@ -345,7 +367,7 @@ impl VideoAnnexBStreamSendState {
access_unit,
)?;
let wire_records = video_wire_records_with_parity(&records)?;
send_payloads.extend(wire_payloads_to_media_send(encode_media_wire_records(
groups.push(wire_payloads_to_media_send(encode_media_wire_records(
&wire_records,
stored_at,
&self.config.source_runtime_id,
@ -355,7 +377,7 @@ impl VideoAnnexBStreamSendState {
}
self.pending.drain(..emitted_bytes);
Ok(send_payloads)
Ok(groups)
}
fn advance_frame_clock(&mut self) -> Result<()> {
@ -1525,8 +1547,8 @@ mod tests {
stream.extend_from_slice(&start_code());
stream.extend_from_slice(&[second_slice_header, 0x80]);
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?;
payloads.extend(sender.finish("2026-06-18T00:00:00Z")?);
let mut payloads = flat(sender.push("2026-06-18T00:00:00Z", &stream)?);
payloads.extend(flat(sender.finish("2026-06-18T00:00:00Z")?));
let mut records = Vec::new();
for payload in &payloads {
if let GameCultMediaWireRecord::Video(record) =
@ -1595,8 +1617,8 @@ mod tests {
stream.extend_from_slice(&start_code());
stream.extend_from_slice(&[0x41, 0x80]);
let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?;
let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?;
payloads.extend(sender.finish("2026-06-18T00:00:00Z")?);
let mut payloads = flat(sender.push("2026-06-18T00:00:00Z", &stream)?);
payloads.extend(flat(sender.finish("2026-06-18T00:00:00Z")?));
assert!(payloads.len() >= 2);
let wire_records = payloads
.iter()
@ -1636,9 +1658,10 @@ mod tests {
let second_push = sender.push("2026-06-18T00:00:00Z", &second_frame)?;
assert_eq!(second_push.len(), 1);
assert_eq!(second_push[0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL);
assert_eq!(second_push[0].len(), 1);
assert_eq!(second_push[0][0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL);
let GameCultMediaWireRecord::Video(first) =
decode_media_wire_record(&second_push[0].payload)?
decode_media_wire_record(&second_push[0][0].payload)?
else {
panic!("expected video media record");
};
@ -1651,7 +1674,8 @@ mod tests {
let tail = sender.finish("2026-06-18T00:00:00Z")?;
assert_eq!(tail.len(), 1);
let GameCultMediaWireRecord::Video(second) = decode_media_wire_record(&tail[0].payload)?
assert_eq!(tail[0].len(), 1);
let GameCultMediaWireRecord::Video(second) = decode_media_wire_record(&tail[0][0].payload)?
else {
panic!("expected video media record");
};
@ -1680,21 +1704,84 @@ mod tests {
);
}
fn flat(groups: Vec<MuninnMediaPayloadGroup>) -> Vec<MuninnMediaSendPayload> {
groups.into_iter().flatten().collect()
}
#[test]
fn annex_b_stream_send_state_rejects_unbounded_pending_bytes() -> Result<()> {
fn annex_b_stream_send_state_drops_and_counts_an_oversize_access_unit_instead_of_erroring()
-> Result<()> {
let mut config = stream_send_config();
config.max_pending_bytes = 4;
config.max_pending_bytes = 8;
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let error = sender
.push("2026-06-18T00:00:00Z", &[0x47, 0x40, 0x00, 0x10, 0x00])
.unwrap_err();
// One access unit that never completes and is larger than the ceiling.
let mut oversize = start_code().to_vec();
oversize.extend_from_slice(&[0x65, 0x80, 1, 2, 3, 4, 5, 6, 7, 8]);
let groups = sender.push("2026-06-18T00:00:00Z", &oversize)?;
assert!(
error
.to_string()
.contains("pending buffer exceeded 4 bytes")
);
assert!(groups.is_empty());
assert_eq!(sender.dropped_access_units(), 1);
assert_eq!(sender.pending_bytes(), 0);
// The tail of the dropped unit has no start code and is skipped; the
// stream then carries on with the next access units.
let mut next = vec![9, 9, 9];
next.extend_from_slice(&start_code());
next.extend_from_slice(&[0x65, 0x80]);
next.extend_from_slice(&start_code());
next.extend_from_slice(&[0x41, 0x80]);
let groups = sender.push("2026-06-18T00:00:00Z", &next)?;
assert_eq!(groups.len(), 1);
assert_eq!(sender.dropped_access_units(), 1);
let GameCultMediaWireRecord::Video(record) =
decode_media_wire_record(&groups[0][0].payload)?
else {
panic!("expected video media record");
};
assert!(record.keyframe);
Ok(())
}
#[test]
fn annex_b_stream_send_state_returns_one_group_per_access_unit_with_all_its_chunks_and_parity()
-> Result<()> {
// Three chunks of 16 bytes plus a short fourth: one access unit.
let mut stream = start_code().to_vec();
stream.extend_from_slice(&[0x65, 0x80]);
stream.resize(52, 0xAB);
stream.extend_from_slice(&start_code());
stream.extend_from_slice(&[0x41, 0x80]);
let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?;
let mut groups = sender.push("2026-06-18T00:00:00Z", &stream)?;
groups.extend(sender.finish("2026-06-18T00:00:00Z")?);
assert_eq!(groups.len(), 2);
let mut frame_ids_per_group = Vec::new();
for group in &groups {
let mut ids = std::collections::BTreeSet::new();
let mut data = 0;
let mut parity = 0;
for payload in group {
match decode_media_wire_record(&payload.payload)? {
GameCultMediaWireRecord::Video(record) => {
ids.insert(record.frame_id);
data += 1;
}
GameCultMediaWireRecord::VideoParity(record) => {
ids.insert(record.frame_id);
parity += 1;
}
other => panic!("unexpected record {other:?}"),
}
}
assert_eq!(ids.len(), 1, "a group carries exactly one frame");
frame_ids_per_group.push((ids.into_iter().next().unwrap(), data, parity));
}
assert_eq!(frame_ids_per_group[0], (9, 4, 4));
assert_eq!(frame_ids_per_group[1], (10, 1, 0));
Ok(())
}

960
crates/muninn-daemon/src/media_send.rs Normal file
View file

@ -0,0 +1,960 @@
//! The media send path: the one place that decides when a frame is too old.
//!
//! **Owner.** `MediaSendCore` owns the bounded group queues, the hub, the
//! receiver-feedback intake, the repair cache and its budget, and the progress
//! counters. `run_rudp_mux_once` spawns the encoders and publishes; it feeds
//! this module bytes and never decides a deadline.
//!
//! **One deadline owner.** `MediaSendPolicy::latency_budget` comes from the
//! request's `latency_budget_ms` and is the only source of a deadline. A group
//! is stamped `produced_at + latency_budget` when it leaves the packetizer; a
//! repair inherits the deadline of the frame it repairs; the hub's reliable
//! expiry and the record deadline ticks derive from the same budget.
//!
//! **Bounds.** Between the encoder and the socket every queue has a count
//! bound (`MEDIA_GROUP_CHANNEL_BOUND`, the per-kind group caps) and a time
//! bound (the group deadline). Crossing either drops whole groups, oldest
//! first, and counts them. The video cap is 512 groups of at most one access
//! unit each, and an access unit is at most `max_pending_bytes` (848 * 4096
//! bytes at the default packet size), so the count cap alone implies a worst
//! case of about 1.7 GiB; in practice the deadline empties the queue long
//! before that.
use crate::media_packetizer::{
AudioPcmStreamSendConfig, AudioPcmStreamSendState, GameCultMediaWireRecord,
MUNINN_MEDIA_RUDP_CHANNEL, MuninnMediaPayloadGroup, MuninnMediaSendPayload,
VideoAnnexBStreamSendConfig, VideoAnnexBStreamSendState, decode_media_wire_record,
video_chunk_feedback_key,
};
use anyhow::{Context, Result};
use cultnet_rs::{
CultNetRudpServerEvent, CultNetRudpServerHub, CultNetRudpServerSessionContext,
CultNetTransportFrame, GameCultMediaReceiverFeedbackRecord,
};
use std::collections::{HashMap, VecDeque};
use std::io::{ErrorKind, Read};
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};
pub const MEDIA_SEND_AUDIO_GROUP_CAP: usize = 256;
pub const MEDIA_SEND_VIDEO_GROUP_CAP: usize = 512;
/// Groups the reader threads may have handed over and the send loop not yet
/// taken. Small on purpose: the caps above are where a backlog lives.
pub const MEDIA_GROUP_CHANNEL_BOUND: usize = 8;
const REPAIR_CACHE_CHUNKS: usize = 16_384;
pub const REPAIR_BURST_CHUNKS: usize = 2_048;
pub const REPAIR_INITIAL_CHUNKS_PER_SECOND: usize = 4_096;
const REPAIR_MIN_CHUNKS_PER_SECOND: usize = 8;
const REPAIR_MAX_CHUNKS_PER_SECOND: usize = 16_384;
const REPAIR_ADD_CHUNKS_PER_SECOND: usize = 2_048;
const REPAIR_RECOVERY_INTERVAL_MS: u64 = 2_000;
const REPAIR_MAX_FEEDBACK_PER_POLL: usize = 32;
const REPAIR_MAX_CHUNKS_PER_POLL: usize = 256;
/// A receiver that has said nothing for this long (Ratatoskr pings every
/// second or so) is dropped so its payloads stop being sent into the void.
const RECEIVER_TIMEOUT_MS: u64 = 6_000;
pub const SEND_PACE_EVERY_PAYLOADS: usize = 4;
pub const SEND_PACE_SLEEP_US: u64 = 250;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MediaKind {
Video,
Audio,
}
/// What the request decided about sending. Nothing else mints a deadline.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MediaSendPolicy {
pub latency_budget: Duration,
pub video_pace_every_payloads: usize,
pub video_pace_sleep: Duration,
}
impl MediaSendPolicy {
pub fn from_request(latency_budget_ms: u32) -> Self {
Self {
latency_budget: Duration::from_millis(u64::from(latency_budget_ms.max(1))),
video_pace_every_payloads: SEND_PACE_EVERY_PAYLOADS,
video_pace_sleep: Duration::from_micros(SEND_PACE_SLEEP_US),
}
}
pub fn latency_budget_ms(&self) -> u64 {
self.latency_budget.as_millis() as u64
}
/// The video record deadline, in the 90 kHz video timebase.
pub fn video_deadline_delay_ticks(&self) -> i64 {
i64::try_from(self.latency_budget_ms().saturating_mul(90)).unwrap_or(i64::MAX)
}
/// The audio record deadline, in samples at `sample_rate`.
pub fn audio_deadline_delay_ticks(&self, sample_rate: u32) -> i64 {
if sample_rate == 0 {
return 1_024;
}
i64::try_from(
self.latency_budget_ms()
.saturating_mul(u64::from(sample_rate))
/ 1_000,
)
.unwrap_or(i64::MAX)
.max(1_024)
}
}
/// One access unit with its parity, or one audio packet. It is sent whole or
/// not at all, and it dies at `deadline`.
pub struct QueuedMediaGroup {
pub kind: MediaKind,
pub deadline: Instant,
pub payloads: MuninnMediaPayloadGroup,
}
impl QueuedMediaGroup {
pub fn new(
kind: MediaKind,
produced_at: Instant,
policy: &MediaSendPolicy,
payloads: MuninnMediaPayloadGroup,
) -> Self {
Self {
kind,
deadline: produced_at + policy.latency_budget,
payloads,
}
}
/// Sendable at the deadline itself, expired one instant after.
pub fn expired(&self, now: Instant) -> bool {
now > self.deadline
}
}
/// What a reader thread hands the send loop.
pub enum MediaIntake {
Group(QueuedMediaGroup),
/// Video access units the packetizer dropped for outgrowing its ceiling.
VideoDropped(u64),
}
pub type MediaIntakeSender = mpsc::SyncSender<Result<MediaIntake>>;
pub type MediaIntakeReceiver = mpsc::Receiver<Result<MediaIntake>>;
pub fn media_intake_channel() -> (MediaIntakeSender, MediaIntakeReceiver) {
mpsc::sync_channel(MEDIA_GROUP_CHANNEL_BOUND)
}
#[derive(Default)]
struct PendingMediaGroups {
audio: VecDeque<QueuedMediaGroup>,
video: VecDeque<QueuedMediaGroup>,
}
impl PendingMediaGroups {
/// `true` when the queue was at its cap and its oldest group was dropped
/// to make room.
fn push(&mut self, group: QueuedMediaGroup) -> bool {
let (queue, cap) = match group.kind {
MediaKind::Audio => (&mut self.audio, MEDIA_SEND_AUDIO_GROUP_CAP),
MediaKind::Video => (&mut self.video, MEDIA_SEND_VIDEO_GROUP_CAP),
};
let dropped = queue.len() >= cap;
if dropped {
queue.pop_front();
}
queue.push_back(group);
dropped
}
/// Audio leaves before video: a late audio packet is a click, a late
/// frame is one dropped frame.
fn pop_next(&mut self) -> Option<QueuedMediaGroup> {
self.audio.pop_front().or_else(|| self.video.pop_front())
}
fn is_empty(&self) -> bool {
self.audio.is_empty() && self.video.is_empty()
}
}
/// Counted where it happens. `groups_dropped_*` is a cap crossing;
/// `groups_expired` is a group whose deadline passed with none of it sent;
/// `groups_cut_short` is one whose deadline passed with part of it already on
/// the wire, which is the only way a partial frame can leave.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct MediaSendStats {
pub groups_sent: u64,
pub payloads_sent: u64,
pub groups_dropped_audio: u64,
pub groups_dropped_video: u64,
pub groups_expired: u64,
pub groups_cut_short: u64,
}
impl MediaSendStats {
pub fn groups_lost(&self) -> u64 {
self.groups_dropped_audio
+ self.groups_dropped_video
+ self.groups_expired
+ self.groups_cut_short
}
/// Accounts for one group the send loop has finished with.
fn record_group(&mut self, outcome: &GroupSend, group_payloads: usize) {
match outcome {
GroupSend::Complete => {
self.groups_sent += 1;
self.payloads_sent += group_payloads as u64;
}
GroupSend::DeadlinePassed { payloads_out: 0 } => self.groups_expired += 1,
GroupSend::DeadlinePassed { payloads_out } => {
self.payloads_sent += *payloads_out as u64;
self.groups_cut_short += 1;
}
}
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct MuninnRudpReceiverFeedbackStats {
pub feedback_records: u64,
pub requested_keyframes: u64,
pub late_frames: u64,
pub missing_video_chunks: u64,
pub repaired_video_chunks: u64,
pub deferred_repair_chunks: u64,
pub repair_chunks_per_second: usize,
pub highest_decodable_frame_id: Option<u64>,
}
#[derive(Debug)]
pub struct MuninnRudpRepairBudget {
chunks_per_second: usize,
min_chunks_per_second: usize,
max_chunks_per_second: usize,
add_chunks_per_second: usize,
recovery_interval: Duration,
max_available_chunks: usize,
available_chunks: usize,
last_refill_at: Instant,
last_rate_adjust_at: Instant,
last_queue_dropped: u64,
}
impl MuninnRudpRepairBudget {
pub fn new(chunks_per_second: usize, max_available_chunks: usize) -> Self {
let now = Instant::now();
let max_available_chunks = max_available_chunks.max(1);
Self {
chunks_per_second: chunks_per_second.max(1),
min_chunks_per_second: REPAIR_MIN_CHUNKS_PER_SECOND,
max_chunks_per_second: REPAIR_MAX_CHUNKS_PER_SECOND,
add_chunks_per_second: REPAIR_ADD_CHUNKS_PER_SECOND,
recovery_interval: Duration::from_millis(REPAIR_RECOVERY_INTERVAL_MS),
max_available_chunks,
available_chunks: max_available_chunks,
last_refill_at: now,
last_rate_adjust_at: now,
last_queue_dropped: 0,
}
}
pub fn chunks_per_second(&self) -> usize {
self.chunks_per_second
}
pub fn take(&mut self, requested_chunks: usize, now: Instant, queue_dropped: u64) -> usize {
self.adjust_rate(now, queue_dropped, requested_chunks);
self.refill(now);
let allowed = requested_chunks.min(self.available_chunks);
self.available_chunks -= allowed;
allowed
}
fn adjust_rate(&mut self, now: Instant, queue_dropped: u64, requested_chunks: usize) {
if queue_dropped > self.last_queue_dropped {
self.chunks_per_second = (self.chunks_per_second / 2).max(self.min_chunks_per_second);
self.available_chunks = self.available_chunks.min(self.chunks_per_second);
self.last_queue_dropped = queue_dropped;
self.last_rate_adjust_at = now;
return;
}
self.last_queue_dropped = queue_dropped;
if requested_chunks == 0
|| now.saturating_duration_since(self.last_rate_adjust_at) < self.recovery_interval
{
return;
}
if self.chunks_per_second < self.max_chunks_per_second {
self.chunks_per_second = self
.chunks_per_second
.saturating_add(self.add_chunks_per_second)
.min(self.max_chunks_per_second);
self.last_rate_adjust_at = now;
}
}
fn refill(&mut self, now: Instant) {
let elapsed_ms = now
.saturating_duration_since(self.last_refill_at)
.as_millis() as usize;
if elapsed_ms == 0 {
return;
}
let refill_chunks = elapsed_ms.saturating_mul(self.chunks_per_second) / 1_000;
if refill_chunks == 0 {
return;
}
self.available_chunks = self
.available_chunks
.saturating_add(refill_chunks)
.min(self.max_available_chunks);
self.last_refill_at = now;
}
}
#[derive(Debug)]
struct RepairEntry {
payload: MuninnMediaSendPayload,
/// The deadline of the frame group the chunk left in. A repair never
/// gets a deadline of its own.
deadline: Instant,
}
#[derive(Debug)]
pub struct RecentVideoChunkRepairCache {
max_entries: usize,
order: VecDeque<String>,
entries: HashMap<String, RepairEntry>,
}
impl RecentVideoChunkRepairCache {
pub fn new(max_entries: usize) -> Self {
Self {
max_entries,
order: VecDeque::new(),
entries: HashMap::new(),
}
}
pub fn remember(&mut self, payload: &MuninnMediaSendPayload, deadline: Instant) -> Result<()> {
let Some(key) = video_repair_cache_key_from_payload(payload)? else {
return Ok(());
};
let entry = RepairEntry {
payload: payload.clone(),
deadline,
};
if self.entries.contains_key(&key) {
self.entries.insert(key, entry);
return Ok(());
}
self.order.push_back(key.clone());
self.entries.insert(key, entry);
while self.entries.len() > self.max_entries {
let Some(expired) = self.order.pop_front() else {
break;
};
self.entries.remove(&expired);
}
Ok(())
}
/// The chunks the receiver is missing that are still inside their frame's
/// deadline at `now`, each with that deadline.
pub fn repair_payloads_for_feedback(
&self,
feedback: &GameCultMediaReceiverFeedbackRecord,
now: Instant,
) -> Vec<(MuninnMediaSendPayload, Instant)> {
feedback
.missing_video_chunk_keys
.iter()
.filter_map(|chunk_key| {
self.entries.get(&video_repair_cache_key(
&feedback.stream_id,
&feedback.session_id,
chunk_key,
))
})
.filter(|entry| now <= entry.deadline)
.map(|entry| (entry.payload.clone(), entry.deadline))
.collect()
}
}
fn video_repair_cache_key(stream_id: &str, session_id: &str, chunk_key: &str) -> String {
format!("{stream_id}:{session_id}:video:{chunk_key}")
}
fn video_repair_cache_key_from_payload(payload: &MuninnMediaSendPayload) -> Result<Option<String>> {
if payload.channel_id != MUNINN_MEDIA_RUDP_CHANNEL {
return Ok(None);
}
let GameCultMediaWireRecord::Video(video) = decode_media_wire_record(&payload.payload)? else {
return Ok(None);
};
Ok(Some(video_repair_cache_key(
&video.stream_id,
&video.session_id,
&video_chunk_feedback_key(video.frame_id, video.chunk_index),
)))
}
#[derive(Debug)]
pub struct MuninnRudpMediaSendPacer {
payloads_since_pause: usize,
every_payloads: usize,
sleep_for: Duration,
}
impl MuninnRudpMediaSendPacer {
pub fn new(every_payloads: usize, sleep_for: Duration) -> Self {
Self {
payloads_since_pause: 0,
every_payloads: every_payloads.max(1),
sleep_for,
}
}
fn observe_sent_payload(&mut self) {
if self.sleep_for.is_zero() {
return;
}
self.payloads_since_pause = self.payloads_since_pause.saturating_add(1);
if self.payloads_since_pause < self.every_payloads {
return;
}
self.payloads_since_pause = 0;
thread::sleep(self.sleep_for);
}
}
pub fn is_would_block_error(error: &anyhow::Error) -> bool {
error.chain().any(|cause| {
cause
.downcast_ref::<std::io::Error>()
.is_some_and(|io| io.kind() == ErrorKind::WouldBlock)
|| cause.to_string().contains("os error 10035")
})
}
/// Tries `attempt` until it goes through or `deadline` passes. While the
/// socket is full it polls resends and tries again; it never sleeps and never
/// outlives the deadline. `false` means the deadline won.
fn send_until<H>(
host: &mut H,
deadline: Instant,
mut now: impl FnMut() -> Instant,
mut attempt: impl FnMut(&mut H) -> Result<()>,
mut poll: impl FnMut(&mut H) -> Result<()>,
) -> Result<bool> {
loop {
match attempt(host) {
Ok(()) => return Ok(true),
Err(error) if is_would_block_error(&error) => {
if now() > deadline {
return Ok(false);
}
poll(host)?;
thread::yield_now();
}
Err(error) => return Err(error).context("sending typed Muninn media payload"),
}
}
}
fn poll_resends(hub: &mut CultNetRudpServerHub) -> Result<()> {
match hub.poll_resends() {
Ok(()) => Ok(()),
Err(error) if is_would_block_error(&error) => Ok(()),
Err(error) => Err(error).context("polling Muninn RUDP media resends"),
}
}
#[derive(Debug)]
enum GroupSend {
Complete,
/// The deadline passed after `payloads_out` payloads had gone out.
DeadlinePassed { payloads_out: usize },
}
/// Every payload to every receiver, in order, or as many as the deadline
/// allows. A payload with nobody to send it to is not a failure; the progress
/// line says how many receivers there are.
fn send_payloads(
hub: &mut CultNetRudpServerHub,
receivers: &[CultNetRudpServerSessionContext],
payloads: &[MuninnMediaSendPayload],
deadline: Instant,
pacer: &mut MuninnRudpMediaSendPacer,
) -> Result<GroupSend> {
for (payloads_out, payload) in payloads.iter().enumerate() {
for receiver in receivers {
let sent = send_until(
hub,
deadline,
Instant::now,
|hub| hub.send(receiver, payload.channel_id, payload.payload.clone()),
poll_resends,
)?;
if !sent {
return Ok(GroupSend::DeadlinePassed { payloads_out });
}
}
pacer.observe_sent_payload();
}
Ok(GroupSend::Complete)
}
pub struct MediaSendCore {
hub: CultNetRudpServerHub,
queues: PendingMediaGroups,
repair_cache: RecentVideoChunkRepairCache,
repair_budget: MuninnRudpRepairBudget,
video_pacer: MuninnRudpMediaSendPacer,
audio_pacer: MuninnRudpMediaSendPacer,
stats: MediaSendStats,
feedback: MuninnRudpReceiverFeedbackStats,
handled_keyframe_requests: u64,
disconnected: bool,
}
impl MediaSendCore {
pub fn new(hub: CultNetRudpServerHub, policy: &MediaSendPolicy) -> Self {
Self {
hub,
queues: PendingMediaGroups::default(),
repair_cache: RecentVideoChunkRepairCache::new(REPAIR_CACHE_CHUNKS),
repair_budget: MuninnRudpRepairBudget::new(
REPAIR_INITIAL_CHUNKS_PER_SECOND,
REPAIR_BURST_CHUNKS,
),
video_pacer: MuninnRudpMediaSendPacer::new(
policy.video_pace_every_payloads,
policy.video_pace_sleep,
),
audio_pacer: MuninnRudpMediaSendPacer::new(0, Duration::ZERO),
stats: MediaSendStats::default(),
feedback: MuninnRudpReceiverFeedbackStats::default(),
handled_keyframe_requests: 0,
disconnected: false,
}
}
#[cfg(test)]
pub fn stats(&self) -> &MediaSendStats {
&self.stats
}
#[cfg(test)]
pub fn feedback(&self) -> &MuninnRudpReceiverFeedbackStats {
&self.feedback
}
#[cfg(test)]
pub fn hub(&self) -> &CultNetRudpServerHub {
&self.hub
}
/// Nothing queued and every reader gone: the encoders have ended.
pub fn finished(&self) -> bool {
self.disconnected && self.queues.is_empty()
}
/// Takes what the reader threads have produced. Waits up to `wait` for the
/// first group only when nothing is queued to send.
pub fn intake(&mut self, rx: &MediaIntakeReceiver, wait: Duration) -> Result<()> {
if self.disconnected {
return Ok(());
}
if self.queues.is_empty() {
match rx.recv_timeout(wait) {
Ok(item) => self.accept(item?),
Err(mpsc::RecvTimeoutError::Timeout) => return Ok(()),
Err(mpsc::RecvTimeoutError::Disconnected) => {
self.disconnected = true;
return Ok(());
}
}
}
loop {
match rx.try_recv() {
Ok(item) => self.accept(item?),
Err(mpsc::TryRecvError::Empty) => return Ok(()),
Err(mpsc::TryRecvError::Disconnected) => {
self.disconnected = true;
return Ok(());
}
}
}
}
fn accept(&mut self, item: MediaIntake) {
match item {
MediaIntake::Group(group) => self.enqueue(group),
MediaIntake::VideoDropped(count) => {
self.stats.groups_dropped_video += count;
self.log_loss_milestone();
}
}
}
pub fn enqueue(&mut self, group: QueuedMediaGroup) {
let kind = group.kind;
if self.queues.push(group) {
match kind {
MediaKind::Audio => self.stats.groups_dropped_audio += 1,
MediaKind::Video => self.stats.groups_dropped_video += 1,
}
self.log_loss_milestone();
}
}
/// Sends the next group that is still inside its deadline at `now`,
/// discarding expired ones whole on the way. `true` when a group went out
/// (all of it, or the part the deadline allowed).
pub fn send_next(&mut self, now: Instant) -> Result<bool> {
while let Some(group) = self.queues.pop_next() {
if group.expired(now) {
self.stats.groups_expired += 1;
self.log_loss_milestone();
continue;
}
let receivers = self.hub.sessions();
let pacer = match group.kind {
MediaKind::Video => &mut self.video_pacer,
MediaKind::Audio => &mut self.audio_pacer,
};
let outcome = send_payloads(
&mut self.hub,
&receivers,
&group.payloads,
group.deadline,
pacer,
)?;
self.stats.record_group(&outcome, group.payloads.len());
match outcome {
GroupSend::Complete => {
if group.kind == MediaKind::Video {
for payload in &group.payloads {
self.repair_cache.remember(payload, group.deadline)?;
}
}
if self.stats.groups_sent == 1 || self.stats.groups_sent % 900 == 0 {
eprintln!("{}", self.progress_detail());
}
}
GroupSend::DeadlinePassed { .. } => self.log_loss_milestone(),
}
return Ok(true);
}
Ok(false)
}
/// Receiver feedback, repairs, keyframe pressure and resends.
pub fn service_control(&mut self, now: Instant) -> Result<()> {
self.poll_feedback(now)?;
if self.feedback.requested_keyframes > self.handled_keyframe_requests {
self.handled_keyframe_requests = self.feedback.requested_keyframes;
eprintln!(
"Muninn RUDP receiver requested a fresh keyframe; continuing current low-latency encoder session until explicit encoder control exists."
);
}
poll_resends(&mut self.hub)
}
fn poll_feedback(&mut self, now: Instant) -> Result<()> {
for gone in self.hub.remove_timed_out_sessions(RECEIVER_TIMEOUT_MS) {
eprintln!(
"Muninn media receiver {} at {} went silent and was dropped.",
String::from_utf8_lossy(&gone.connect_payload),
gone.remote_addr
);
}
let mut feedback_processed = 0_usize;
loop {
match self.hub.receive_event_once() {
Ok(Some(CultNetRudpServerEvent::Connected { session })) => {
eprintln!(
"Muninn media receiver {} attached from {}.",
String::from_utf8_lossy(&session.connect_payload),
session.remote_addr
);
}
Ok(Some(CultNetRudpServerEvent::Disconnected { session, reason })) => {
eprintln!(
"Muninn media receiver {} at {} disconnected: {}.",
String::from_utf8_lossy(&session.connect_payload),
session.remote_addr,
String::from_utf8_lossy(&reason)
);
}
Ok(Some(CultNetRudpServerEvent::Frame { session, frame })) => {
if feedback_processed >= REPAIR_MAX_FEEDBACK_PER_POLL {
return Ok(());
}
feedback_processed += 1;
let repairs = record_receiver_feedback(
&frame,
&mut self.feedback,
&self.repair_cache,
now,
)?;
let requested = repairs.len();
let allowed = self.repair_budget.take(
requested.min(REPAIR_MAX_CHUNKS_PER_POLL),
now,
self.stats.groups_lost(),
);
self.feedback.repair_chunks_per_second = self.repair_budget.chunks_per_second();
self.feedback.deferred_repair_chunks = self
.feedback
.deferred_repair_chunks
.saturating_add(requested.saturating_sub(allowed) as u64);
// Repairs go back to the receiver that asked, not to everyone,
// and each one keeps the deadline of its own frame.
for (payload, deadline) in repairs.into_iter().take(allowed) {
let outcome = send_payloads(
&mut self.hub,
std::slice::from_ref(&session),
std::slice::from_ref(&payload),
deadline,
&mut self.video_pacer,
)?;
if matches!(outcome, GroupSend::Complete) {
self.feedback.repaired_video_chunks =
self.feedback.repaired_video_chunks.saturating_add(1);
}
}
}
Ok(Some(CultNetRudpServerEvent::Pong { .. })) => {}
Ok(None) => return Ok(()),
Err(error) if is_would_block_error(&error) => return Ok(()),
Err(error) => return Err(error).context("polling Muninn RUDP media feedback"),
}
}
}
fn log_loss_milestone(&self) {
let lost = self.stats.groups_lost();
// Loss can arrive in the hundreds per second; the first and every
// 300th is enough to see it without flooding the log.
if lost == 1 || lost % 300 == 0 {
eprintln!("{}", self.progress_detail());
}
}
pub fn progress_detail(&self) -> String {
format!(
"Muninn RUDP media progress: receivers={} groups_sent={} payloads_sent={} groups_dropped_audio={} groups_dropped_video={} groups_expired={} groups_cut_short={} reliable_expired={} receiver_feedback={} receiver_keyframes={} receiver_late_frames={} receiver_missing_chunks={} receiver_repaired_chunks={} receiver_deferred_repairs={} repair_rate={} receiver_highest_decodable={}; pending_audio={} pending_video={}",
self.hub.sessions().len(),
self.stats.groups_sent,
self.stats.payloads_sent,
self.stats.groups_dropped_audio,
self.stats.groups_dropped_video,
self.stats.groups_expired,
self.stats.groups_cut_short,
self.hub.stats().reliable_packets_expired,
self.feedback.feedback_records,
self.feedback.requested_keyframes,
self.feedback.late_frames,
self.feedback.missing_video_chunks,
self.feedback.repaired_video_chunks,
self.feedback.deferred_repair_chunks,
self.feedback.repair_chunks_per_second,
self.feedback
.highest_decodable_frame_id
.map(|frame_id| frame_id.to_string())
.unwrap_or_else(|| "none".to_string()),
self.queues.audio.len(),
self.queues.video.len(),
)
}
#[cfg(test)]
pub(crate) fn queued_groups(&self, kind: MediaKind) -> usize {
match kind {
MediaKind::Audio => self.queues.audio.len(),
MediaKind::Video => self.queues.video.len(),
}
}
#[cfg(test)]
pub(crate) fn queued_payload_bytes(&self, kind: MediaKind) -> usize {
match kind {
MediaKind::Audio => &self.queues.audio,
MediaKind::Video => &self.queues.video,
}
.iter()
.flat_map(|group| &group.payloads)
.map(|payload| payload.payload.len())
.sum()
}
#[cfg(test)]
pub(crate) fn oldest_deadline(&self, kind: MediaKind) -> Option<Instant> {
match kind {
MediaKind::Audio => self.queues.audio.front(),
MediaKind::Video => self.queues.video.front(),
}
.map(|group| group.deadline)
}
}
fn record_receiver_feedback(
frame: &CultNetTransportFrame,
stats: &mut MuninnRudpReceiverFeedbackStats,
repair_cache: &RecentVideoChunkRepairCache,
now: Instant,
) -> Result<Vec<(MuninnMediaSendPayload, Instant)>> {
if frame.channel_id != MUNINN_MEDIA_RUDP_CHANNEL {
return Ok(Vec::new());
}
let GameCultMediaWireRecord::Feedback(feedback) = decode_media_wire_record(&frame.payload)?
else {
return Ok(Vec::new());
};
let repairs = repair_cache.repair_payloads_for_feedback(&feedback, now);
stats.feedback_records = stats.feedback_records.saturating_add(1);
if feedback.requested_keyframe {
stats.requested_keyframes = stats.requested_keyframes.saturating_add(1);
}
stats.late_frames = stats
.late_frames
.saturating_add(feedback.late_frame_ids.len() as u64);
stats.missing_video_chunks = stats
.missing_video_chunks
.saturating_add(feedback.missing_video_chunk_keys.len() as u64);
if let Some(frame_id) = feedback.highest_decodable_frame_id {
stats.highest_decodable_frame_id = Some(
stats
.highest_decodable_frame_id
.map(|current| current.max(frame_id))
.unwrap_or(frame_id),
);
}
Ok(repairs)
}
/// Reads Annex B from `reader`, packetizes it, and hands one group per access
/// unit to the send loop, stamped with the request's deadline.
pub fn spawn_video_group_reader<R>(
tx: MediaIntakeSender,
mut reader: R,
config: VideoAnnexBStreamSendConfig,
policy: MediaSendPolicy,
) -> thread::JoinHandle<()>
where
R: Read + Send + 'static,
{
thread::spawn(move || {
if let Err(error) = read_video_groups(&tx, &mut reader, config, &policy) {
let _ = tx.send(Err(error));
}
})
}
pub fn spawn_audio_group_reader<R>(
tx: MediaIntakeSender,
mut reader: R,
config: AudioPcmStreamSendConfig,
policy: MediaSendPolicy,
) -> thread::JoinHandle<()>
where
R: Read + Send + 'static,
{
thread::spawn(move || {
if let Err(error) = read_audio_groups(&tx, &mut reader, config, &policy) {
let _ = tx.send(Err(error));
}
})
}
fn send_intake(tx: &MediaIntakeSender, item: MediaIntake) -> Result<()> {
tx.send(Ok(item)).context("queueing typed Muninn media")
}
fn read_video_groups<R: Read>(
tx: &MediaIntakeSender,
reader: &mut R,
config: VideoAnnexBStreamSendConfig,
policy: &MediaSendPolicy,
) -> Result<()> {
let mut sender = VideoAnnexBStreamSendState::new(config)?;
let mut buffer = vec![0_u8; 64 * 1024];
let mut dropped_reported = 0_u64;
loop {
let read = reader
.read(&mut buffer)
.context("reading encoded Annex B video from ffmpeg stdout")?;
let stored_at = crate::timestamp()?;
let (groups, done) = if read == 0 {
(sender.finish(&stored_at)?, true)
} else {
(sender.push(&stored_at, &buffer[..read])?, false)
};
for payloads in groups {
send_intake(
tx,
MediaIntake::Group(QueuedMediaGroup::new(
MediaKind::Video,
Instant::now(),
policy,
payloads,
)),
)?;
}
let dropped = sender.dropped_access_units();
if dropped > dropped_reported {
send_intake(tx, MediaIntake::VideoDropped(dropped - dropped_reported))?;
dropped_reported = dropped;
}
if done {
return Ok(());
}
}
}
fn read_audio_groups<R: Read>(
tx: &MediaIntakeSender,
reader: &mut R,
config: AudioPcmStreamSendConfig,
policy: &MediaSendPolicy,
) -> Result<()> {
let mut sender = AudioPcmStreamSendState::new(config)?;
let mut buffer = vec![0_u8; 16 * 1024];
loop {
let read = reader
.read(&mut buffer)
.context("reading PCM audio from ffmpeg stdout")?;
let stored_at = crate::timestamp()?;
let (payloads, done) = if read == 0 {
(sender.finish(&stored_at)?, true)
} else {
(sender.push(&stored_at, &buffer[..read])?, false)
};
for payload in payloads {
send_intake(
tx,
MediaIntake::Group(QueuedMediaGroup::new(
MediaKind::Audio,
Instant::now(),
policy,
vec![payload],
)),
)?;
}
if done {
return Ok(());
}
}
}
#[cfg(test)]
mod tests;

845
crates/muninn-daemon/src/media_send/tests.rs Normal file
View file

@ -0,0 +1,845 @@
use super::*;
use crate::media_packetizer::{MediaWireProvenance, encode_media_wire_record};
use cultnet_rs::{
CultNetRudpPacketType, CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions,
CultNetTransportDelivery, ReceiverFeedbackOptions, build_receiver_feedback, decode_rudp_packet,
};
use std::collections::{BTreeMap, BTreeSet};
use std::io::Cursor;
use std::net::{SocketAddr, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
const CLIP: &[u8] = include_bytes!("../../testdata/clip.h264");
const STREAM_ID: &str = "muninn.harness.rudp";
const VIDEO_SESSION_ID: &str = "harness:video";
const AUDIO_SESSION_ID: &str = "harness:audio";
const WAIT: Duration = Duration::from_secs(5);
fn wait_until(timeout: Duration, mut done: impl FnMut() -> bool) -> bool {
let start = Instant::now();
while start.elapsed() < timeout {
if done() {
return true;
}
thread::sleep(Duration::from_millis(1));
}
done()
}
fn video_config() -> VideoAnnexBStreamSendConfig {
VideoAnnexBStreamSendConfig {
stream_id: STREAM_ID.to_string(),
session_id: VIDEO_SESSION_ID.to_string(),
codec: "h264".to_string(),
first_frame_id: 0,
first_pts_ticks: 0,
frame_duration_ticks: 3_000,
timebase_num: 1,
timebase_den: 90_000,
deadline_delay_ticks: 22_500,
max_payload_bytes: 848,
max_pending_bytes: 848 * 4096,
source_runtime_id: "harness".to_string(),
source_role: "muninn.rudp.video".to_string(),
}
}
fn audio_config() -> AudioPcmStreamSendConfig {
AudioPcmStreamSendConfig {
stream_id: STREAM_ID.to_string(),
session_id: AUDIO_SESSION_ID.to_string(),
codec: "pcm-f32le-interleaved".to_string(),
first_packet_id: 0,
first_pts_ticks: 0,
packet_duration_ticks: 480,
timebase_num: 1,
timebase_den: 48_000,
deadline_delay_ticks: 24_000,
channels: 2,
bytes_per_sample: 4,
max_pending_bytes: 480 * 2 * 4 * 128,
source_runtime_id: "harness".to_string(),
source_role: "muninn.rudp.audio".to_string(),
}
}
/// Ten audio packets of an ascending f32 ramp: every byte position is
/// distinct, so a reordered or dropped packet cannot pass for the ramp.
fn pcm_ramp() -> Vec<u8> {
let samples = 480 * 2 * 10;
(0..samples)
.flat_map(|index| (index as f32 / samples as f32).to_le_bytes())
.collect()
}
/// The clip's access units as the packetizer sends them: one group each.
fn clip_groups() -> Vec<MuninnMediaPayloadGroup> {
let mut sender = VideoAnnexBStreamSendState::new(video_config()).unwrap();
let mut groups = sender.push("unix-0", CLIP).unwrap();
groups.extend(sender.finish("unix-0").unwrap());
groups
}
fn group_at(
kind: MediaKind,
produced_at: Instant,
policy: &MediaSendPolicy,
payloads: MuninnMediaPayloadGroup,
) -> QueuedMediaGroup {
QueuedMediaGroup::new(kind, produced_at, policy, payloads)
}
fn marker_payload(marker: u8) -> MuninnMediaSendPayload {
MuninnMediaSendPayload {
channel_id: MUNINN_MEDIA_RUDP_CHANNEL,
payload: vec![marker],
}
}
/// `(frame_id, is_parity, chunk or parity index)` of a video wire record.
fn frame_of(payload: &[u8]) -> Option<(u64, bool, u16)> {
match decode_media_wire_record(payload).ok()? {
GameCultMediaWireRecord::Video(record) => Some((record.frame_id, false, record.chunk_index)),
GameCultMediaWireRecord::VideoParity(record) => {
Some((record.frame_id, true, record.parity_index))
}
_ => None,
}
}
fn test_hub(policy: &MediaSendPolicy) -> CultNetRudpServerHub {
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
socket.set_nonblocking(true).unwrap();
CultNetRudpServerHub::new(crate::muninn_media_rudp_hub_options(
socket,
&crate::muninn_rudp_media_profile(),
policy,
))
.unwrap()
}
/// The request's budget and no pacing sleeps: the harness is about what the
/// core decides, not about how fast a Windows sender is allowed to go.
fn quiet_policy(latency_budget_ms: u32) -> MediaSendPolicy {
let mut policy = MediaSendPolicy::from_request(latency_budget_ms);
policy.video_pace_sleep = Duration::ZERO;
policy
}
/// One datagram the hub sent toward the receiver, as the relay saw it.
#[derive(Clone, Debug)]
struct Sent {
packet_type: CultNetRudpPacketType,
channel: String,
payload: Vec<u8>,
dropped: bool,
}
/// A UDP relay between the receiver and the hub. The hub's data packets are
/// dropped by a seeded schedule. Everything is logged before the drop, so
/// tests assert on what the sender put on the wire.
struct DropRelay {
addr: SocketAddr,
log: Arc<Mutex<Vec<Sent>>>,
stop: Arc<AtomicBool>,
thread: Option<thread::JoinHandle<()>>,
}
impl DropRelay {
fn start(hub_addr: SocketAddr, drop_percent: u64, seed: u64) -> Self {
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
socket
.set_read_timeout(Some(Duration::from_millis(1)))
.unwrap();
let addr = socket.local_addr().unwrap();
let log = Arc::new(Mutex::new(Vec::new()));
let stop = Arc::new(AtomicBool::new(false));
let thread = {
let log = Arc::clone(&log);
let stop = Arc::clone(&stop);
thread::spawn(move || {
let mut client: Option<SocketAddr> = None;
let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1;
let mut buffer = vec![0_u8; 65_535];
while !stop.load(Ordering::Relaxed) {
let Ok((received, from)) = socket.recv_from(&mut buffer) else {
continue;
};
if from != hub_addr {
client = Some(from);
let _ = socket.send_to(&buffer[..received], hub_addr);
continue;
}
let Some(client) = client else { continue };
let Ok(packet) = decode_rudp_packet(&buffer[..received]) else {
continue;
};
let is_data = packet.packet_type == CultNetRudpPacketType::Data;
state ^= state << 13;
state ^= state >> 7;
state ^= state << 17;
let dropped = is_data && state % 100 < drop_percent;
log.lock().unwrap().push(Sent {
packet_type: packet.packet_type,
channel: packet.channel_id,
payload: packet.payload,
dropped,
});
if !dropped {
let _ = socket.send_to(&buffer[..received], client);
}
}
})
};
Self {
addr,
log,
stop,
thread: Some(thread),
}
}
fn data(&self) -> Vec<Sent> {
self.log
.lock()
.unwrap()
.iter()
.filter(|sent| sent.packet_type == CultNetRudpPacketType::Data)
.cloned()
.collect()
}
fn video_frames_on_wire(&self) -> Vec<(u64, bool, u16)> {
self.data()
.iter()
.filter(|sent| sent.channel == MUNINN_MEDIA_RUDP_CHANNEL)
.filter_map(|sent| frame_of(&sent.payload))
.collect()
}
}
impl Drop for DropRelay {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
if let Some(thread) = self.thread.take() {
let _ = thread.join();
}
}
}
struct Rig {
core: MediaSendCore,
client: CultNetRudpSocketTransportConnection,
relay: DropRelay,
policy: MediaSendPolicy,
}
fn rig(latency_budget_ms: u32, drop_percent: u64, seed: u64) -> Rig {
let policy = quiet_policy(latency_budget_ms);
let hub = test_hub(&policy);
let relay = DropRelay::start(hub.local_addr().unwrap(), drop_percent, seed);
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
socket.set_nonblocking(true).unwrap();
let mut options = CultNetRudpSocketTransportOptions::client(
"harness-receiver",
socket,
relay.addr,
crate::MUNINN_MEDIA_RUDP_CONNECTION_ID,
);
options.media_delivery = Some(CultNetTransportDelivery::Unreliable);
options.max_fragment_bytes = Some(crate::MUNINN_RUDP_MEDIA_MAX_FRAGMENT_BYTES as u32);
let mut client = CultNetRudpSocketTransportConnection::new(options).unwrap();
let mut core = MediaSendCore::new(hub, &policy);
client.connect(b"harness-receiver".to_vec()).unwrap();
assert!(
wait_until(WAIT, || {
core.service_control(Instant::now()).unwrap();
let _ = client.receive_once().unwrap();
client.connected() && core.hub().sessions().len() == 1
}),
"receiver never attached through the relay"
);
Rig {
core,
client,
relay,
policy,
}
}
// ---- policy, groups and queues -----------------------------------------------------
#[test]
fn a_group_is_expired_only_after_its_deadline_and_never_at_it() {
let policy = quiet_policy(250);
let produced_at = Instant::now();
let group = group_at(MediaKind::Video, produced_at, &policy, Vec::new());
assert!(!group.expired(group.deadline));
assert!(!group.expired(group.deadline - Duration::from_millis(1)));
assert!(group.expired(group.deadline + Duration::from_millis(1)));
}
#[test]
fn a_zero_request_budget_still_leaves_the_stream_a_millisecond() {
assert_eq!(
MediaSendPolicy::from_request(0).latency_budget,
Duration::from_millis(1)
);
}
#[test]
fn the_queue_caps_groups_per_kind_and_drops_the_oldest_first() {
let policy = quiet_policy(250);
let mut core = MediaSendCore::new(test_hub(&policy), &policy);
let t0 = Instant::now();
let over = 37;
for index in 0..(MEDIA_SEND_VIDEO_GROUP_CAP + over) {
core.enqueue(group_at(
MediaKind::Video,
t0 + Duration::from_millis(index as u64),
&policy,
vec![marker_payload(1)],
));
}
for index in 0..(MEDIA_SEND_AUDIO_GROUP_CAP + 5) {
core.enqueue(group_at(
MediaKind::Audio,
t0 + Duration::from_millis(index as u64),
&policy,
vec![marker_payload(2)],
));
}
assert_eq!(
core.queued_groups(MediaKind::Video),
MEDIA_SEND_VIDEO_GROUP_CAP
);
assert_eq!(
core.queued_groups(MediaKind::Audio),
MEDIA_SEND_AUDIO_GROUP_CAP
);
// One byte per group: resident bytes are bounded by the cap times the group size.
assert_eq!(
core.queued_payload_bytes(MediaKind::Video),
MEDIA_SEND_VIDEO_GROUP_CAP
);
assert_eq!(
core.queued_payload_bytes(MediaKind::Audio),
MEDIA_SEND_AUDIO_GROUP_CAP
);
assert_eq!(core.stats().groups_dropped_video, over as u64);
assert_eq!(core.stats().groups_dropped_audio, 5);
// The survivors are the newest: the oldest deadline left is the 38th group's.
assert_eq!(
core.oldest_deadline(MediaKind::Video),
Some(t0 + Duration::from_millis(over as u64) + policy.latency_budget)
);
assert_eq!(
core.oldest_deadline(MediaKind::Audio),
Some(t0 + Duration::from_millis(5) + policy.latency_budget)
);
}
#[test]
fn the_intake_channel_holds_the_reader_at_its_bound_instead_of_growing() {
let policy = quiet_policy(5_000);
let (tx, rx) = media_intake_channel();
let frames = clip_groups().len();
assert!(frames > MEDIA_GROUP_CHANNEL_BOUND + 1);
let reader = spawn_video_group_reader(tx, Cursor::new(CLIP.to_vec()), video_config(), policy);
// Nothing drains, so the reader parks with the channel full.
thread::sleep(Duration::from_millis(300));
assert!(
!reader.is_finished(),
"an unbounded channel let the reader run to the end"
);
let mut received = 0;
wait_until(WAIT, || {
while rx.try_recv().is_ok() {
received += 1;
}
received == frames
});
assert_eq!(received, frames);
assert!(wait_until(WAIT, || reader.is_finished()));
}
#[test]
fn a_blocked_send_gives_up_at_the_group_deadline_and_not_before() {
let start = Instant::now();
let deadline = start + Duration::from_millis(3);
let ticks = std::cell::Cell::new(0_u64);
let clock = || {
ticks.set(ticks.get() + 1);
start + Duration::from_millis(ticks.get())
};
let mut polls = 0_u32;
// Blocked forever: the deadline ends it, after polling resends in between.
let sent = send_until(
&mut polls,
deadline,
clock,
|_| Err(std::io::Error::from(ErrorKind::WouldBlock).into()),
|polls| {
*polls += 1;
Ok(())
},
)
.unwrap();
assert!(!sent);
assert_eq!(polls, 3, "one poll per blocked attempt inside the deadline");
// Blocked twice, then through: sent, without waiting for the deadline.
ticks.set(0);
let mut attempts = 0_u32;
let sent = send_until(
&mut attempts,
start + Duration::from_millis(100),
clock,
|attempts| {
*attempts += 1;
if *attempts < 3 {
Err(std::io::Error::from(ErrorKind::WouldBlock).into())
} else {
Ok(())
}
},
|_| Ok(()),
)
.unwrap();
assert!(sent);
assert_eq!(attempts, 3);
}
#[test]
fn a_send_error_that_is_not_backpressure_is_an_error_not_a_dropped_group() {
let error = send_until(
&mut (),
Instant::now() + Duration::from_secs(1),
Instant::now,
|_| Err(anyhow::anyhow!("socket closed")),
|_| Ok(()),
)
.unwrap_err();
assert!(format!("{error:#}").contains("socket closed"));
}
// ---- moved from the mux: repair budget, feedback intake, repair cache ---------------
#[test]
fn rudp_repair_budget_backs_off_on_media_drops_and_recovers_when_stable() {
let start = Instant::now();
let mut budget = MuninnRudpRepairBudget {
chunks_per_second: 64,
min_chunks_per_second: 8,
max_chunks_per_second: 128,
add_chunks_per_second: 8,
recovery_interval: Duration::from_secs(2),
max_available_chunks: 4,
available_chunks: 4,
last_refill_at: start,
last_rate_adjust_at: start,
last_queue_dropped: 0,
};
assert_eq!(budget.take(4, start, 0), 4);
assert_eq!(budget.chunks_per_second(), 64);
assert_eq!(budget.take(3, start + Duration::from_millis(10), 1), 0);
assert_eq!(budget.chunks_per_second(), 32);
assert_eq!(budget.take(3, start + Duration::from_secs(1), 1), 3);
assert_eq!(budget.chunks_per_second(), 32);
assert_eq!(budget.take(3, start + Duration::from_secs(3), 1), 3);
assert_eq!(budget.chunks_per_second(), 40);
}
#[test]
fn default_rudp_repair_budget_has_lan_stream_headroom() {
let mut budget =
MuninnRudpRepairBudget::new(REPAIR_INITIAL_CHUNKS_PER_SECOND, REPAIR_BURST_CHUNKS);
let start = budget.last_refill_at;
assert_eq!(budget.chunks_per_second(), 4_096);
assert_eq!(budget.take(2_048, start, 0), 2_048);
assert_eq!(budget.take(8_192, start + Duration::from_secs(1), 0), 2_048);
assert_eq!(budget.take(8_192, start + Duration::from_secs(3), 0), 2_048);
assert_eq!(budget.chunks_per_second(), 6_144);
assert_eq!(budget.take(512, start + Duration::from_secs(4), 1), 512);
assert_eq!(budget.chunks_per_second(), 3_072);
}
fn feedback_for(
missing_chunk_keys: Vec<String>,
requested_keyframe: bool,
) -> GameCultMediaReceiverFeedbackRecord {
build_receiver_feedback(ReceiverFeedbackOptions {
stream_id: STREAM_ID,
session_id: VIDEO_SESSION_ID,
receiver_id: "harness-receiver",
highest_decodable_frame_id: Some(41),
missing_frame_ids: Vec::new(),
missing_video_chunk_keys: missing_chunk_keys,
late_frame_ids: vec![42, 43],
requested_keyframe,
jitter_us: 500,
decode_queue_us: 2_000,
observed_at: "unix:1000",
})
.unwrap()
}
fn feedback_wire(feedback: GameCultMediaReceiverFeedbackRecord) -> Vec<u8> {
encode_media_wire_record(
&GameCultMediaWireRecord::Feedback(feedback),
MediaWireProvenance {
stored_at: "unix:1000",
runtime_id: "harness",
role: "harness.receiver",
producer: "harness",
},
)
.unwrap()
}
#[test]
fn rudp_media_receiver_feedback_updates_sender_pressure_stats() {
let frame = CultNetTransportFrame {
channel_id: MUNINN_MEDIA_RUDP_CHANNEL.to_string(),
payload: feedback_wire(feedback_for(vec!["42:1".into(), "42:3".into()], true)),
};
let mut stats = MuninnRudpReceiverFeedbackStats::default();
let repair_cache = RecentVideoChunkRepairCache::new(16);
let repairs =
record_receiver_feedback(&frame, &mut stats, &repair_cache, Instant::now()).unwrap();
assert_eq!(stats.feedback_records, 1);
assert_eq!(stats.requested_keyframes, 1);
assert_eq!(stats.late_frames, 2);
assert_eq!(stats.missing_video_chunks, 2);
assert_eq!(stats.repaired_video_chunks, 0);
assert_eq!(stats.highest_decodable_frame_id, Some(41));
assert!(repairs.is_empty());
}
#[test]
fn repair_cache_returns_recent_missing_video_chunks_with_their_frames_deadline() {
let groups = clip_groups();
let payload = groups[0][0].clone();
let (frame_id, _, chunk_index) = frame_of(&payload.payload).unwrap();
let feedback = feedback_for(
vec![
video_chunk_feedback_key(frame_id, chunk_index),
"999:0".to_string(),
],
false,
);
let deadline = Instant::now() + Duration::from_millis(250);
let mut cache = RecentVideoChunkRepairCache::new(16);
cache.remember(&payload, deadline).unwrap();
let repairs = cache.repair_payloads_for_feedback(&feedback, deadline);
assert_eq!(repairs, vec![(payload, deadline)]);
}
// ---- the wire: whole groups, repairs inside their frame's deadline ------------------
#[test]
fn a_group_past_its_deadline_puts_nothing_on_the_wire_and_a_live_one_puts_all_of_it() {
let mut rig = rig(250, 0, 1);
let groups = clip_groups();
let stale = groups[0].clone();
let live = groups
.iter()
.skip(1)
.find(|group| group.len() > 2)
.expect("the clip has a later multi-chunk frame")
.clone();
let live_frame = frame_of(&live[0].payload).unwrap().0;
let t0 = Instant::now();
let budget = rig.policy.latency_budget;
rig.core
.enqueue(group_at(MediaKind::Video, t0, &rig.policy, stale));
rig.core.enqueue(group_at(
MediaKind::Video,
t0 + 5 * budget,
&rig.policy,
live.clone(),
));
// Two budgets after the stale group was made, inside the live group's.
assert!(rig.core.send_next(t0 + 2 * budget).unwrap());
assert_eq!(rig.core.stats().groups_expired, 1);
assert_eq!(rig.core.stats().groups_sent, 1);
assert!(wait_until(WAIT, || {
rig.relay.video_frames_on_wire().len() == live.len()
}));
thread::sleep(Duration::from_millis(100));
let on_wire = rig.relay.video_frames_on_wire();
assert_eq!(
on_wire.len(),
live.len(),
"every payload of the live group, and no other"
);
assert!(on_wire.iter().all(|(frame, _, _)| *frame == live_frame));
assert!(rig.core.progress_detail().contains("groups_sent=1"));
assert!(rig.core.progress_detail().contains("groups_expired=1"));
}
#[test]
fn audio_leaves_before_video_on_the_wire() {
let mut rig = rig(250, 0, 1);
let t0 = Instant::now();
let video = clip_groups().remove(1);
let mut audio_sender = AudioPcmStreamSendState::new(audio_config()).unwrap();
let audio = audio_sender.push("unix-0", &pcm_ramp()).unwrap().remove(0);
rig.core
.enqueue(group_at(MediaKind::Video, t0, &rig.policy, video));
rig.core
.enqueue(group_at(MediaKind::Audio, t0, &rig.policy, vec![audio]));
assert!(rig.core.send_next(t0).unwrap());
assert!(rig.core.send_next(t0).unwrap());
assert!(wait_until(WAIT, || !rig.relay.data().is_empty()));
assert_eq!(rig.relay.data()[0].channel, "audio");
}
/// Sends the clip's first multi-chunk frame at `t0` and returns its deadline,
/// its frame id and how many payloads it put on the wire.
fn send_repairable_frame(rig: &mut Rig, t0: Instant) -> (Instant, u64, usize) {
let group = clip_groups()
.into_iter()
.find(|group| group.len() > 2)
.unwrap();
let (frame_id, _, _) = frame_of(&group[0].payload).unwrap();
let queued = group_at(MediaKind::Video, t0, &rig.policy, group.clone());
let deadline = queued.deadline;
rig.core.enqueue(queued);
assert!(rig.core.send_next(t0).unwrap());
assert!(wait_until(WAIT, || rig.relay.video_frames_on_wire().len()
== group.len()));
(deadline, frame_id, group.len())
}
fn ask_for_chunk(rig: &mut Rig, frame_id: u64, at: Instant, records_seen: u64) {
rig.client
.send(
MUNINN_MEDIA_RUDP_CHANNEL,
feedback_wire(feedback_for(
vec![video_chunk_feedback_key(frame_id, 0)],
false,
)),
)
.unwrap();
assert!(
wait_until(WAIT, || {
rig.core.service_control(at).unwrap();
rig.core.feedback().feedback_records == records_seen
}),
"the hub never received the feedback"
);
}
#[test]
fn a_repair_is_sent_inside_its_frames_deadline_and_not_after_it() {
// Taken before the handshake, so real time is well past it by the time
// anything is sent: a repair that took its deadline from the clock
// instead of its frame would still be alive at the probe below.
let t0 = Instant::now();
let mut rig = rig(250, 0, 1);
let (deadline, frame_id, frame_payloads) = send_repairable_frame(&mut rig, t0);
let epsilon = Duration::from_millis(2);
ask_for_chunk(&mut rig, frame_id, deadline - epsilon, 1);
assert!(
wait_until(WAIT, || rig.relay.video_frames_on_wire().len()
== frame_payloads + 1),
"a repair asked for just before the deadline is sent"
);
assert_eq!(rig.core.feedback().repaired_video_chunks, 1);
ask_for_chunk(&mut rig, frame_id, deadline + epsilon, 2);
thread::sleep(Duration::from_millis(150));
assert_eq!(
rig.relay.video_frames_on_wire().len(),
frame_payloads + 1,
"a repair asked for just after the deadline is not sent"
);
assert_eq!(rig.core.feedback().repaired_video_chunks, 1);
assert_eq!(rig.core.feedback().missing_video_chunks, 2);
}
// ---- the harness end to end: the fixture clip and the PCM ramp ---------------------
#[derive(Default)]
struct Received {
chunks: BTreeMap<u64, BTreeSet<u16>>,
chunk_counts: BTreeMap<u64, u16>,
audio: BTreeMap<u64, Vec<u8>>,
}
impl Received {
fn take(&mut self, rig: &mut Rig) {
while let Some(frame) = rig.client.receive_once().unwrap() {
match decode_media_wire_record(&frame.payload).unwrap() {
GameCultMediaWireRecord::Video(record) => {
self.chunks
.entry(record.frame_id)
.or_default()
.insert(record.chunk_index);
self.chunk_counts
.insert(record.frame_id, record.chunk_count);
}
GameCultMediaWireRecord::Audio(record) => {
self.audio.insert(record.packet_id, record.payload);
}
_ => {}
}
}
}
fn complete_frames(&self) -> BTreeSet<u64> {
self.chunks
.iter()
.filter(|(frame, chunks)| chunks.len() == self.chunk_counts[frame] as usize)
.map(|(frame, _)| *frame)
.collect()
}
}
/// Feeds the fixture through the real readers and the send loop, over the
/// relay, and returns what the receiver assembled.
fn run_clip(rig: &mut Rig, with_audio: bool) -> Received {
let (tx, rx) = media_intake_channel();
let video = spawn_video_group_reader(
tx.clone(),
Cursor::new(CLIP.to_vec()),
video_config(),
rig.policy.clone(),
);
let audio = with_audio.then(|| {
spawn_audio_group_reader(
tx.clone(),
Cursor::new(pcm_ramp()),
audio_config(),
rig.policy.clone(),
)
});
drop(tx);
let mut received = Received::default();
let start = Instant::now();
while !rig.core.finished() && start.elapsed() < WAIT {
rig.core.intake(&rx, Duration::from_millis(2)).unwrap();
rig.core.send_next(Instant::now()).unwrap();
rig.core.service_control(Instant::now()).unwrap();
received.take(rig);
}
assert!(rig.core.finished(), "the send loop never finished the clip");
let settle = Instant::now();
while settle.elapsed() < Duration::from_millis(300) {
rig.core.service_control(Instant::now()).unwrap();
received.take(rig);
thread::sleep(Duration::from_millis(1));
}
video.join().unwrap();
if let Some(audio) = audio {
audio.join().unwrap();
}
received
}
#[test]
fn the_clip_and_the_pcm_ramp_arrive_whole_through_a_lossless_relay() {
// A budget far above the run time: nothing here may expire.
let mut rig = rig(5_000, 0, 1);
let frames = clip_groups().len() as u64;
let received = run_clip(&mut rig, true);
assert_eq!(
received.complete_frames(),
(0..frames).collect::<BTreeSet<_>>()
);
assert_eq!(rig.core.stats().groups_lost(), 0);
let heard: Vec<u8> = received.audio.values().flatten().copied().collect();
assert_eq!(received.audio.len(), 10);
assert_eq!(heard, pcm_ramp(), "the ramp arrived complete and in order");
}
#[test]
fn the_seeded_relay_drops_the_same_datagrams_every_run_and_the_sender_carries_on() {
let run = |seed| {
let mut rig = rig(5_000, 30, seed);
let received = run_clip(&mut rig, false);
let dropped: Vec<bool> = rig.relay.data().iter().map(|sent| sent.dropped).collect();
(received.complete_frames(), dropped, rig.core.stats().clone())
};
let (complete_a, dropped_a, stats_a) = run(7);
let (complete_b, dropped_b, _) = run(7);
let (_, dropped_other, _) = run(8);
let frames = clip_groups().len();
assert_eq!(dropped_a, dropped_b, "same seed, same drop schedule");
assert_ne!(
dropped_a, dropped_other,
"a different seed drops different datagrams"
);
assert_eq!(complete_a, complete_b);
assert!(dropped_a.iter().any(|dropped| *dropped));
assert!(
complete_a.len() < frames,
"30% loss without parity or repair leaves incomplete frames"
);
assert_eq!(stats_a.groups_sent, frames as u64);
assert_eq!(
stats_a.groups_lost(),
0,
"loss on the wire is not a sender-side drop"
);
}
#[test]
fn a_group_is_counted_by_what_the_deadline_let_reach_the_wire() {
let mut stats = MediaSendStats::default();
stats.record_group(&GroupSend::Complete, 6);
assert_eq!((stats.groups_sent, stats.payloads_sent), (1, 6));
// Deadline passed before the first payload: nothing left, so nothing was cut.
stats.record_group(&GroupSend::DeadlinePassed { payloads_out: 0 }, 6);
assert_eq!((stats.groups_expired, stats.groups_cut_short), (1, 0));
assert_eq!(stats.payloads_sent, 6);
// Deadline passed with part of the group already out: a partial frame.
stats.record_group(&GroupSend::DeadlinePassed { payloads_out: 2 }, 6);
assert_eq!((stats.groups_expired, stats.groups_cut_short), (1, 1));
assert_eq!(stats.payloads_sent, 8);
assert_eq!(stats.groups_lost(), 2);
}
#[test]
fn an_oversize_access_unit_reaches_the_sender_as_a_dropped_video_group_not_an_error() {
let policy = quiet_policy(250);
let mut core = MediaSendCore::new(test_hub(&policy), &policy);
let (tx, rx) = media_intake_channel();
let mut config = video_config();
config.max_pending_bytes = 8;
let mut oversize = vec![0, 0, 0, 1, 0x65, 0x80];
oversize.extend_from_slice(&[7; 20]);
let reader = spawn_video_group_reader(tx, Cursor::new(oversize), config, policy);
assert!(wait_until(WAIT, || reader.is_finished()));
core.intake(&rx, Duration::from_millis(50)).unwrap();
core.intake(&rx, Duration::from_millis(50)).unwrap();
assert_eq!(core.stats().groups_dropped_video, 1);
assert!(core.finished(), "the stream ended cleanly, it was not restarted by an error");
}

BIN
crates/muninn-daemon/testdata/clip.h264 vendored Normal file
View file

Binary file not shown.

7
crates/muninn-daemon/testdata/clip.h264.provenance.txt vendored Normal file
View file

@ -0,0 +1,7 @@
clip.h264: 15 frames of the ffmpeg testsrc2 pattern, 160x120 at 30 fps, H.264 Constrained
Baseline in Annex B with access unit delimiters, one IDR every 8 frames, no B-frames.
Generated once and committed so the media send harness needs no ffmpeg.
Command (ffmpeg 8.1.1 full build, libx264, 2026-09-29):
ffmpeg -f lavfi -i "testsrc2=size=160x120:rate=30:duration=0.5" -c:v libx264 -profile:v baseline -preset veryslow -crf 28 -bf 0 -g 8 -x264-params "threads=1:aud=1:repeat-headers=1:keyint=8:min-keyint=8:scenecut=0" -f h264 clip.h264