From d1770c3476869421898826a41b52a1adc526c073 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 10 Jul 2026 16:26:41 +0200 Subject: [PATCH] refactor(host): shared send-pacing policy for the native and GameStream video planes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Networking-audit deferred plan §5. Both planes spread a frame's wire packets across a time budget in chunked bursts; the schedule logic, PUNKTFUNK_VIDEO_DROP loss injection, and percentile helper were duplicated between punktfunk1::paced_submit and gamestream::stream::spawn_sender. Now one host-local send_pacing::pace_frame carries the policy; each plane keeps its exact historical parameterization and its own syscall layer (GSO Session vs sendmmsg over the RTP socket — policy shared, plumbing not): native burst_bytes = PUNKTFUNK_PACE_BURST_KB (microburst stage), fixed 16-packet chunks, budget = 0.9 × time-to-deadline gamestream no burst stage, bounded steps (≤ 12, chunk ≥ 16, the old pace_layout), fixed budget = 0.75 × frame interval Deterministic-schedule unit tests pin both parameterizations against verbatim transcriptions of the legacy math (burst split, chunk layout, step counts — including pace_layout's historical test anchors) and the sleep-target formula (GameStream's legacy per_step form agrees to ≤ steps/2 ns; the unified fraction form is used for both). Deliberate sub-observable normalizations, all on test-knob or ns-scale paths: PUNKTFUNK_VIDEO_DROP is now parsed once per process and clamped to 1..=90 on the GameStream plane too (was per-stream, unclamped), and the native sleep floor comparison is now >= (was >, differs only at exactly 500 µs). Validation: - 263 host tests green, incl. the end-to-end sender_delivers_batches (spawn_sender → pace_frame → sendmmsg, byte-identical delivery) - PUNKTFUNK_VIDEO_DROP FEC sweep at 5 % and 8 % injected wire loss: all 11 punktfunk1 integration tests (full host↔client roundtrips through send_loop → paced_submit) recover and pass - pending: one real Moonlight smoke session against this build (the legacy-plane timing gate) — recipe handed to the operator Co-Authored-By: Claude Fable 5 --- .../punktfunk-host/src/gamestream/stream.rs | 133 ++---- crates/punktfunk-host/src/main.rs | 1 + crates/punktfunk-host/src/punktfunk1.rs | 128 ++---- crates/punktfunk-host/src/send_pacing.rs | 408 ++++++++++++++++++ 4 files changed, 468 insertions(+), 202 deletions(-) create mode 100644 crates/punktfunk-host/src/send_pacing.rs diff --git a/crates/punktfunk-host/src/gamestream/stream.rs b/crates/punktfunk-host/src/gamestream/stream.rs index 90e60f12..e15524cb 100644 --- a/crates/punktfunk-host/src/gamestream/stream.rs +++ b/crates/punktfunk-host/src/gamestream/stream.rs @@ -11,7 +11,6 @@ use super::VIDEO_PORT; use crate::capture::{self, Capturer, FastSyntheticCapturer}; use crate::encode::{self, Codec}; use anyhow::{Context, Result}; -use rand::Rng; use std::net::UdpSocket; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -428,20 +427,6 @@ fn sendmmsg_all(sock: &UdpSocket, pkts: &[Vec]) -> std::io::Result<()> { Ok(()) } -/// Pacing layout for one frame's `n` packets (`n >= 1`): `(chunk_size, steps)`. The chunk grows -/// with the frame so the number of paced bursts — each ending in a `thread::sleep` — never exceeds -/// `MAX_PACE_STEPS`. A fixed 16-packet chunk let the step count scale with bitrate (~38 for a -/// 4K/250Mbps frame's ~600 packets); the accumulated sub-ms sleep overshoot on the non-RT send -/// thread then blew the per-frame budget and backed the handoff queue up. Bounding the steps keeps -/// microburst shaping at low bitrate while making overshoot negligible and bitrate-independent. -fn pace_layout(n: usize) -> (usize, usize) { - const MIN_PACE_CHUNK: usize = 16; - const MAX_PACE_STEPS: usize = 12; - let chunk_sz = MIN_PACE_CHUNK.max(n.div_ceil(MAX_PACE_STEPS)); - let steps = n.div_ceil(chunk_sz); // ≤ MAX_PACE_STEPS - (chunk_sz, steps) -} - /// One encoded frame handed from the encode loop to the packetizer thread: the frame's access /// units (owned buffers, each with its frame type) plus the shared 90 kHz RTP timestamp. FEC /// packetization runs on the packetizer thread — off the encode loop — so it never serializes @@ -491,15 +476,16 @@ fn spawn_packetizer( } /// Dedicated send thread: one [`PacketBatch`] per frame arrives on `rx`; its packets go out in -/// `sendmmsg` chunks, paced so the frame's data spreads over ~3/4 of the frame interval -/// (microburst shaping at chunk granularity — a real link drops line-rate bursts; the encode -/// thread is never blocked by this). On send failure (client gone) it clears `running`. +/// `sendmmsg` chunks, paced so the frame's data spreads over ~3/4 of the frame interval — the +/// shared [`send_pacing`](crate::send_pacing) policy at the GameStream parameterization: no +/// microburst stage, a BOUNDED step count (≤ 12, chunk ≥ 16, see the policy's docs for the +/// "send queue full" history that bound guards), each step ending in a sleep toward its slice +/// of the fixed budget. On send failure (client gone) it clears `running`. fn spawn_sender( sock: UdpSocket, rx: std::sync::mpsc::Receiver, frame_interval: Duration, running: Arc, - drop_pct: u32, ) -> Result<()> { std::thread::Builder::new() .name("punktfunk-send".into()) @@ -507,52 +493,37 @@ fn spawn_sender( // Transmit thread: above-normal, matching the native path's send thread (includes the // Windows session tuning/MMCSS this used to call directly; adds the Linux nice -5). crate::punktfunk1::boost_thread_priority(false); - // Chunk pacing: spread the frame's packets across the send budget in a BOUNDED number - // of bursts. A fixed 16-packet chunk made the burst count scale with bitrate (~38 for a - // 4K/250Mbps frame's ~600 packets), and each burst ends in a `thread::sleep`; on this - // non-RT send thread those sub-ms sleeps overshoot, and ~38 per frame blew the 12.5ms - // budget past the 16.67ms frame interval — backing the depth-2 handoff queue up and - // dropping ~half the frames ("send queue full"). Capping the step count keeps the - // microburst shaping (a real link drops line-rate bursts) while making per-frame sleep - // overshoot negligible and independent of bitrate. let budget = frame_interval.mul_f32(0.75); - let mut rng = rand::thread_rng(); + let cfg = crate::send_pacing::PaceCfg { + burst_bytes: None, // no microburst stage — the whole frame spreads + chunk: crate::send_pacing::ChunkPolicy::Bounded { + min_chunk: 16, + max_steps: 12, + }, + sleep_floor: Duration::from_micros(500), + }; let mut sent: u64 = 0; let mut dropped: u64 = 0; while let Ok(mut batch) = rx.recv() { - if drop_pct > 0 { - batch.retain(|_| { - let keep = rng.gen_range(0..100) >= drop_pct; - if !keep { - dropped += 1; - } - keep - }); - } - let n = batch.len(); - if n == 0 { + // FEC test knob (PUNKTFUNK_VIDEO_DROP) — same knob the native plane honors. + dropped += crate::send_pacing::inject_video_drop(&mut batch); + if batch.is_empty() { continue; } - // Chunk size + step count, bounded so a high-bitrate frame doesn't fan out into - // dozens of sleeps. Each step gets an equal slice of the budget (total pacing time - // == budget regardless of n). - let (chunk_sz, steps) = pace_layout(n); - let per_step = budget.mul_f64(1.0 / steps as f64); - let start = Instant::now(); - for (i, chunk) in batch.chunks(chunk_sz).enumerate() { - if let Err(e) = sendmmsg_all(&sock, chunk) { - tracing::info!(error = %e, sent, "video: client unreachable — stopping stream"); - running.store(false, Ordering::SeqCst); - return; - } - sent += chunk.len() as u64; - // Sleep toward the next step's deadline; skip sub-500µs sleeps (jitter). - let target = start + per_step.mul_f64((i + 1) as f64); - if let Some(ahead) = target.checked_duration_since(Instant::now()) { - if ahead >= Duration::from_micros(500) { - std::thread::sleep(ahead); - } - } + let r = crate::send_pacing::pace_frame( + &batch, + crate::send_pacing::PaceBudget::Fixed(budget), + &cfg, + |chunk| { + sendmmsg_all(&sock, chunk)?; + sent += chunk.len() as u64; + Ok::<(), std::io::Error>(()) + }, + ); + if let Err(e) = r { + tracing::info!(error = %e, sent, "video: client unreachable — stopping stream"); + running.store(false, Ordering::SeqCst); + return; } } tracing::debug!(sent, dropped, "video sender exiting"); @@ -561,16 +532,7 @@ fn spawn_sender( Ok(()) } -/// Percentile of a slice (sorts it in place first). `q` in `0.0..=1.0`. Used for the web-console -/// stats sample's per-stage p50/p99. -fn percentile(v: &mut [u32], q: f64) -> u32 { - if v.is_empty() { - return 0; - } - v.sort_unstable(); - let i = ((v.len() as f64 * q) as usize).min(v.len() - 1); - v[i] -} +use crate::send_pacing::percentile; /// The encode → packetize loop, over a borrowed capturer. Sending runs on a dedicated thread /// (see [`spawn_sender`]) so a send spike can never stall capture/encode. @@ -633,11 +595,6 @@ fn stream_body( let mut fps_count: u32 = 0; let mut fps_t = Instant::now(); let stream_start = Instant::now(); - // Test knob: drop this % of outbound packets to exercise FEC recovery (0 = off). - let drop_pct: u32 = std::env::var("PUNKTFUNK_VIDEO_DROP") - .ok() - .and_then(|v| v.parse().ok()) - .unwrap_or(0); let mut sent_batches: u64 = 0; let mut dropped_batches: u64 = 0; @@ -656,7 +613,6 @@ fn stream_body( batch_rx, Duration::from_secs_f64(1.0 / target_fps as f64), running.clone(), - drop_pct, )?; let (raw_tx, raw_rx) = std::sync::mpsc::sync_channel::(2); spawn_packetizer(raw_rx, batch_tx, pk, goodput.clone())?; @@ -995,7 +951,6 @@ mod tests { rx, Duration::from_millis(8), // ~120fps frame interval running.clone(), - 0, ) .unwrap(); @@ -1032,30 +987,4 @@ mod tests { assert_eq!(got, 3 * PER_FRAME); assert!(running.load(Ordering::SeqCst), "no spurious client-gone"); } - - /// The pacing layout bounds the paced-burst (and thus sleep) count regardless of frame size, - /// while always covering every packet and keeping small frames on the 16-packet floor. Guards - /// the 4K/high-bitrate "send queue full" regression (a fixed 16-packet chunk fanned a ~600 - /// packet frame into ~38 sleeps, whose overshoot blew the per-frame send budget). - #[test] - fn pace_layout_bounds_step_count() { - for &n in &[1usize, 16, 146, 610, 1024, 5000, 50_000] { - let (chunk, steps) = pace_layout(n); - assert!(steps >= 1, "n={n}: at least one step"); - assert!(steps <= 12, "n={n}: step count {steps} exceeded the cap"); - assert!( - chunk >= 16, - "n={n}: chunk {chunk} below the 16-packet floor" - ); - assert!( - chunk * steps >= n, - "n={n}: {chunk}×{steps} must cover all packets" - ); - } - // Small frames stay on the floor: one 16-packet burst. - assert_eq!(pace_layout(1), (16, 1)); - assert_eq!(pace_layout(16), (16, 1)); - // A 4K/250Mbps frame (~600 packets) was ~38 bursts at a fixed 16 — now bounded. - assert!(pace_layout(610).1 <= 12); - } } diff --git a/crates/punktfunk-host/src/main.rs b/crates/punktfunk-host/src/main.rs index 3484c468..302a5a97 100644 --- a/crates/punktfunk-host/src/main.rs +++ b/crates/punktfunk-host/src/main.rs @@ -60,6 +60,7 @@ mod native_pairing; mod pipeline; mod punktfunk1; mod pwinit; +mod send_pacing; #[cfg(target_os = "windows")] #[path = "windows/service.rs"] mod service; diff --git a/crates/punktfunk-host/src/punktfunk1.rs b/crates/punktfunk-host/src/punktfunk1.rs index ffd7545d..42bcc8de 100644 --- a/crates/punktfunk-host/src/punktfunk1.rs +++ b/crates/punktfunk-host/src/punktfunk1.rs @@ -120,6 +120,7 @@ fn bind_data_socket(data_port: Option) -> std::io::Result<(std::net::UdpSoc /// The native (punktfunk/1) trust store + on-demand arming PIN, shared with the management API. use crate::native_pairing::{NativePairing, PairingDecision}; +use crate::send_pacing::{percentile, PaceStat}; /// The shared streaming-stats recorder (web-console capture/graph), shared with the management API /// and the GameStream loop; threaded into each session's `SessionContext`. use crate::stats_recorder::StatsRecorder; @@ -2624,34 +2625,16 @@ fn service_probes( } } -/// Seal one access unit and send its packets PACED over the budget until `deadline` (the next -/// frame's due time), in 16-packet `sendmmsg` chunks — so a high-bitrate frame spreads across the -/// frame interval instead of bursting all at once into the NIC. A real link drops a line-rate burst -/// (the host send buffer EAGAINs), and under infinite GOP a single dropped frame freezes the decode -/// until the next keyframe — the cause of the "freezes over ~150 Mbps, no image at 400 Mbps" -/// symptom. When there's little/no slack (encode ≈ interval at very high fps) the budget collapses -/// to ~0 and every chunk goes out immediately, so this is never slower than the unpaced path. -/// One paced send's outcome: how long the frame's packets took to leave (`spread_us`) and whether -/// any were paced (vs the whole frame fitting the microburst and going out immediately). Fed to the -/// PUNKTFUNK_PERF histogram so the pacing tail is visible per-frame. -struct PaceStat { - spread_us: u32, - paced: bool, -} - -const PACE_CHUNK: usize = 16; - -/// Seal one access unit and send it with MICROBURST pacing: the first `burst_cap` bytes go out -/// immediately (one absorbed burst the NIC / socket tx-buffer can swallow), and only the OVERFLOW -/// beyond that is spread in [`PACE_CHUNK`]-packet chunks across ~90% of the time to `deadline`. So a -/// normal-bitrate frame (≤ cap) leaves in one immediate burst at ~0 added latency, while a genuine -/// IDR / sustained-high-bitrate frame (≫ cap) still spreads — keeping the freeze fix exactly where -/// it's needed (an unpaced line-rate burst overruns the kernel tx buffer → EAGAIN drop → under -/// infinite GOP, a freeze until the next keyframe). With no slack (encode ≈ interval) the budget -/// collapses to 0 and even the overflow goes out immediately, so this is never slower than unpaced. -/// Parsed-once `PUNKTFUNK_VIDEO_DROP` percentage for the native data plane (see `paced_submit`). -static NATIVE_VIDEO_DROP: std::sync::OnceLock = std::sync::OnceLock::new(); - +/// Seal one access unit and send it with MICROBURST pacing (the shared +/// [`send_pacing`](crate::send_pacing) policy, native parameterization): the first `burst_cap` +/// bytes go out immediately (one absorbed burst the NIC / socket tx-buffer can swallow), and +/// only the OVERFLOW beyond that is spread in 16-packet chunks across ~90% of the time to +/// `deadline`. So a normal-bitrate frame (≤ cap) leaves in one immediate burst at ~0 added +/// latency, while a genuine IDR / sustained-high-bitrate frame (≫ cap) still spreads — keeping +/// the freeze fix exactly where it's needed (an unpaced line-rate burst overruns the kernel tx +/// buffer → EAGAIN drop → under infinite GOP, a freeze until the next keyframe). With no slack +/// (encode ≈ interval) the budget collapses to 0 and even the overflow goes out immediately, so +/// this is never slower than unpaced. fn paced_submit( session: &mut Session, data: &[u8], @@ -2664,80 +2647,25 @@ fn paced_submit( .seal_frame(data, pts_ns, flags) .map_err(|e| anyhow!("seal_frame: {e:?}"))?; let mut refs: Vec<&[u8]> = wires.iter().map(|w| w.as_slice()).collect(); - // FEC/recovery test knob: PUNKTFUNK_VIDEO_DROP=N discards N% of the sealed wire packets - // before send — controlled loss injection with no netem/root, same knob the GameStream video - // path honors. Parsed once; 0/unset = off (the normal path is untouched). - let drop_pct = *NATIVE_VIDEO_DROP.get_or_init(|| { - let pct = std::env::var("PUNKTFUNK_VIDEO_DROP") - .ok() - .and_then(|s| s.parse::().ok()) - .filter(|p| (1..=90).contains(p)) - .unwrap_or(0); - if pct > 0 { - tracing::warn!( - pct, - "PUNKTFUNK_VIDEO_DROP: injecting wire-packet loss (FEC test)" - ); - } - pct - }); - if drop_pct > 0 { - use rand::Rng; - let mut rng = rand::thread_rng(); - refs.retain(|_| rng.gen_range(0..100) >= drop_pct); - } - let start = std::time::Instant::now(); - - // Split at the microburst cap: packets [0..split] burst out immediately, [split..] are paced. - let mut cum = 0usize; - let mut split = refs.len(); - for (k, r) in refs.iter().enumerate() { - cum += r.len(); - if cum >= burst_cap { - split = k + 1; - break; - } - } - for chunk in refs[..split].chunks(PACE_CHUNK) { - session - .send_sealed(chunk) - .map_err(|e| anyhow!("send_sealed: {e:?}"))?; - } - let paced = split < refs.len(); - if paced { - let pace_start = std::time::Instant::now(); - let budget = deadline - .checked_duration_since(pace_start) - .unwrap_or_default() - .mul_f32(0.9); - let m = refs[split..].len().div_ceil(PACE_CHUNK).max(1); - for (j, chunk) in refs[split..].chunks(PACE_CHUNK).enumerate() { - session - .send_sealed(chunk) - .map_err(|e| anyhow!("send_sealed: {e:?}"))?; - // Sleep toward this chunk's slice of the budget; skip sub-500µs waits (scheduler jitter). - let target = pace_start + budget.mul_f64((j + 1) as f64 / m as f64); - if let Some(ahead) = target.checked_duration_since(std::time::Instant::now()) { - if ahead > std::time::Duration::from_micros(500) { - std::thread::sleep(ahead); - } - } - } - } - let spread_us = start.elapsed().as_micros() as u32; + // FEC/recovery test knob (PUNKTFUNK_VIDEO_DROP) — same knob the GameStream plane honors. + crate::send_pacing::inject_video_drop(&mut refs); + let cfg = crate::send_pacing::PaceCfg { + burst_bytes: Some(burst_cap), + chunk: crate::send_pacing::ChunkPolicy::Fixed(16), + sleep_floor: std::time::Duration::from_micros(500), + }; + let result = crate::send_pacing::pace_frame( + &refs, + crate::send_pacing::PaceBudget::UntilDeadline { + deadline, + fraction: 0.9, + }, + &cfg, + |chunk| session.send_sealed(chunk).map(|_| ()), + ); drop(refs); // release the borrow of `wires` so it can return to the seal pool session.reclaim_wires(wires); - Ok(PaceStat { spread_us, paced }) -} - -/// Percentile of a slice (sorts it in place first). `q` in 0.0..=1.0. -fn percentile(sorted_or_not: &mut [u32], q: f64) -> u32 { - if sorted_or_not.is_empty() { - return 0; - } - sorted_or_not.sort_unstable(); - let i = ((sorted_or_not.len() as f64 * q) as usize).min(sorted_or_not.len() - 1); - sorted_or_not[i] + result.map_err(|e| anyhow!("send_sealed: {e:?}")) } /// One encoded frame handed from the capture/encode thread to the send thread (the encode|send diff --git a/crates/punktfunk-host/src/send_pacing.rs b/crates/punktfunk-host/src/send_pacing.rs new file mode 100644 index 00000000..655c3e45 --- /dev/null +++ b/crates/punktfunk-host/src/send_pacing.rs @@ -0,0 +1,408 @@ +//! Shared microburst pacing POLICY for the two video send planes (networking-audit deferred +//! plan §5): the native plane (`punktfunk1::paced_submit`, GSO via the core `Session`) and the +//! GameStream compat plane (`gamestream::stream::spawn_sender`, `sendmmsg` over its own RTP +//! socket). Both spread a frame's packets across a time budget in chunked bursts so a real link +//! doesn't drop the frame as one line-rate burst; the syscall layers stay deliberately separate +//! (different sockets, framing, and error contracts) — this module shares the schedule, not the +//! plumbing. +//! +//! The two planes keep their historical parameterizations exactly (pinned by the +//! deterministic-schedule tests below): +//! +//! * **native** — the first `burst_bytes` leave immediately (one absorbed microburst), only the +//! overflow is paced in fixed 16-packet chunks across 90 % of the time left to the frame +//! deadline (no slack ⇒ budget 0 ⇒ never slower than unpaced); +//! * **GameStream** — no burst stage; the whole frame spreads across a fixed ¾-frame-interval +//! budget in a BOUNDED number of steps (≤ 12, chunk ≥ 16), because on that non-RT send thread +//! every step ends in a `thread::sleep` whose overshoot must stay independent of bitrate +//! (Moonlight clients are tested against this timing). +//! +//! `PUNKTFUNK_VIDEO_DROP` (the FEC-recovery test knob both planes honor) and the stats +//! `percentile` helper live here too — they were duplicated alongside the pacing. + +use std::time::{Duration, Instant}; + +/// One paced send's outcome: how long the frame's packets took to leave (`spread_us`) and +/// whether any were paced (vs the whole frame fitting the microburst and going out +/// immediately). The native plane feeds it to the PUNKTFUNK_PERF histogram so the pacing tail +/// is visible per-frame. +pub(crate) struct PaceStat { + pub(crate) spread_us: u32, + pub(crate) paced: bool, +} + +/// How a frame's packets split into send chunks. +#[derive(Clone, Copy, Debug)] +pub(crate) enum ChunkPolicy { + /// Fixed chunk size; the step count scales with the frame (native: 16). + Fixed(usize), + /// Bounded step count: `chunk = max(min_chunk, ceil(n / max_steps))` (GameStream: 16 / 12). + /// Keeps per-frame sleep overshoot independent of bitrate — see `spawn_sender`'s history. + Bounded { min_chunk: usize, max_steps: usize }, +} + +/// The time the paced (post-burst) packets spread across. +#[derive(Clone, Copy, Debug)] +pub(crate) enum PaceBudget { + /// `(deadline − now-after-burst) × fraction`, collapsing to 0 with no slack (native: 0.9). + UntilDeadline { deadline: Instant, fraction: f32 }, + /// A precomputed fixed budget (GameStream: ¾ of the frame interval). + Fixed(Duration), +} + +/// Per-plane pacing parameters. See the module doc for the two canonical values. +#[derive(Clone, Copy, Debug)] +pub(crate) struct PaceCfg { + /// Bytes that leave immediately as one absorbed microburst before pacing starts; `None` = + /// no burst stage at all (GameStream). `Some(0)` still bursts the first packet — the split + /// is "the packet that crosses the cap goes with the burst", exactly the native semantics. + pub(crate) burst_bytes: Option, + pub(crate) chunk: ChunkPolicy, + /// Sleeps shorter than this are skipped (scheduler-jitter floor; both planes: 500 µs). + pub(crate) sleep_floor: Duration, +} + +/// A frame's send schedule, computed up front as pure data (what the deterministic tests pin): +/// packets `[0..burst_len)` go immediately in `chunk`-sized bursts; the rest go in `steps` +/// chunks of `chunk`, chunk `j` (0-based) sleeping toward `budget × (j+1)/steps`. +#[derive(Debug, PartialEq, Eq)] +pub(crate) struct PaceSchedule { + pub(crate) burst_len: usize, + pub(crate) chunk: usize, + pub(crate) steps: usize, +} + +/// Compute the schedule for one frame's wire packets under `cfg`. +pub(crate) fn schedule>(packets: &[T], cfg: &PaceCfg) -> PaceSchedule { + let burst_len = match cfg.burst_bytes { + None => 0, + Some(cap) => { + // The packet that crosses the cap still bursts (`split = k + 1`) — the whole frame + // bursts when it never crosses it. + let mut cum = 0usize; + let mut split = packets.len(); + for (k, p) in packets.iter().enumerate() { + cum += p.as_ref().len(); + if cum >= cap { + split = k + 1; + break; + } + } + split + } + }; + let overflow = packets.len() - burst_len; + let (chunk, steps) = match cfg.chunk { + ChunkPolicy::Fixed(c) => (c, overflow.div_ceil(c).max(1)), + ChunkPolicy::Bounded { + min_chunk, + max_steps, + } => { + let c = min_chunk.max(overflow.div_ceil(max_steps)); + (c, overflow.div_ceil(c).max(1)) + } + }; + PaceSchedule { + burst_len, + chunk, + steps, + } +} + +/// Send one frame's packets under the plane's pacing policy: the burst stage leaves +/// immediately, then each paced chunk is sent and slept toward its slice of the budget +/// (sub-`sleep_floor` waits are skipped). A `send` error aborts the frame and propagates — +/// the native plane bails the session, GameStream stops the stream. +pub(crate) fn pace_frame, E>( + packets: &[T], + budget: PaceBudget, + cfg: &PaceCfg, + mut send: impl FnMut(&[T]) -> Result<(), E>, +) -> Result { + let start = Instant::now(); + let sched = schedule(packets, cfg); + for chunk in packets[..sched.burst_len].chunks(sched.chunk) { + send(chunk)?; + } + let paced = sched.burst_len < packets.len(); + if paced { + let pace_start = Instant::now(); + let budget = match budget { + PaceBudget::UntilDeadline { deadline, fraction } => deadline + .checked_duration_since(pace_start) + .unwrap_or_default() + .mul_f32(fraction), + PaceBudget::Fixed(d) => d, + }; + for (j, chunk) in packets[sched.burst_len..].chunks(sched.chunk).enumerate() { + send(chunk)?; + // Sleep toward this chunk's slice of the budget; skip sub-floor waits (jitter). + let target = pace_start + budget.mul_f64((j + 1) as f64 / sched.steps as f64); + if let Some(ahead) = target.checked_duration_since(Instant::now()) { + if ahead >= cfg.sleep_floor { + std::thread::sleep(ahead); + } + } + } + } + Ok(PaceStat { + spread_us: start.elapsed().as_micros() as u32, + paced, + }) +} + +/// Parsed-once `PUNKTFUNK_VIDEO_DROP` percentage (1..=90, anything else = off): discard N % of +/// the sealed wire packets before send — controlled loss injection with no netem/root, honored +/// by BOTH video planes. Warned once on activation. +pub(crate) fn video_drop_pct() -> u32 { + static PCT: std::sync::OnceLock = std::sync::OnceLock::new(); + *PCT.get_or_init(|| { + let pct = std::env::var("PUNKTFUNK_VIDEO_DROP") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|p| (1..=90).contains(p)) + .unwrap_or(0); + if pct > 0 { + tracing::warn!( + pct, + "PUNKTFUNK_VIDEO_DROP: injecting wire-packet loss (FEC test)" + ); + } + pct + }) +} + +/// Apply the [`video_drop_pct`] loss injection to one frame's wire packets, returning how many +/// were discarded (0 when the knob is off — the normal path is untouched). +pub(crate) fn inject_video_drop(packets: &mut Vec) -> u64 { + let pct = video_drop_pct(); + if pct == 0 { + return 0; + } + use rand::Rng; + let mut rng = rand::thread_rng(); + let before = packets.len(); + packets.retain(|_| rng.gen_range(0..100) >= pct); + (before - packets.len()) as u64 +} + +/// Percentile of a slice (sorts it in place first). `q` in `0.0..=1.0`. Used for the +/// PUNKTFUNK_PERF histograms and the web-console stats sample's per-stage p50/p99. +pub(crate) fn percentile(v: &mut [u32], q: f64) -> u32 { + if v.is_empty() { + return 0; + } + v.sort_unstable(); + let i = ((v.len() as f64 * q) as usize).min(v.len() - 1); + v[i] +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The native plane's canonical parameters (mirrors `punktfunk1::paced_submit`). + fn native_cfg(burst_cap: usize) -> PaceCfg { + PaceCfg { + burst_bytes: Some(burst_cap), + chunk: ChunkPolicy::Fixed(16), + sleep_floor: Duration::from_micros(500), + } + } + + /// The GameStream plane's canonical parameters (mirrors `gamestream::stream::spawn_sender`). + fn gs_cfg() -> PaceCfg { + PaceCfg { + burst_bytes: None, + chunk: ChunkPolicy::Bounded { + min_chunk: 16, + max_steps: 12, + }, + sleep_floor: Duration::from_micros(500), + } + } + + fn packets(n: usize, len: usize) -> Vec> { + (0..n).map(|_| vec![0u8; len]).collect() + } + + /// Deterministic-schedule pin, native plane: burst split + chunking + step count must + /// reproduce the legacy `paced_submit` math exactly — `split = first k with cum ≥ cap + 1` + /// (whole frame if never crossed), fixed 16-packet chunks, `m = ceil(overflow/16).max(1)`. + #[test] + fn native_schedule_matches_legacy_paced_submit() { + let legacy = |sizes: &[usize], burst_cap: usize| -> (usize, usize) { + // Verbatim transcription of the pre-dedup split + step-count computation. + let mut cum = 0usize; + let mut split = sizes.len(); + for (k, len) in sizes.iter().enumerate() { + cum += len; + if cum >= burst_cap { + split = k + 1; + break; + } + } + let m = (sizes.len() - split).div_ceil(16).max(1); + (split, m) + }; + for (n, len, cap) in [ + (1usize, 1200usize, 128 * 1024usize), // tiny frame ≪ cap → all burst + (109, 1200, 128 * 1024), // exactly at the cap boundary region + (110, 1200, 128 * 1024), // one past + (600, 1200, 128 * 1024), // 4K P-frame: burst + paced overflow + (3300, 1200, 128 * 1024), // multi-MB IDR + (600, 1200, 0), // cap 0: first packet still bursts + (0, 1200, 128 * 1024), // empty (post-drop-injection) frame + ] { + let pkts = packets(n, len); + let sizes: Vec = pkts.iter().map(|p| p.len()).collect(); + let (split, m) = legacy(&sizes, cap); + let s = schedule(&pkts, &native_cfg(cap)); + assert_eq!(s.burst_len, split, "n={n} cap={cap}: burst split"); + assert_eq!(s.chunk, 16, "n={n} cap={cap}: chunk size"); + assert_eq!(s.steps, m, "n={n} cap={cap}: paced step count"); + } + } + + /// Deterministic-schedule pin, GameStream plane: no burst stage, and the chunk/step layout + /// must reproduce the legacy `pace_layout` exactly (chunk = max(16, ceil(n/12)), ≤ 12 + /// steps) — including the historical bounds its old unit test asserted. + #[test] + fn gamestream_schedule_matches_legacy_pace_layout() { + let legacy_pace_layout = |n: usize| -> (usize, usize) { + let chunk_sz = 16usize.max(n.div_ceil(12)); + (chunk_sz, n.div_ceil(chunk_sz)) + }; + for &n in &[1usize, 16, 17, 146, 192, 193, 610, 1024, 5000, 50_000] { + let pkts = packets(n, 1024); + let (chunk, steps) = legacy_pace_layout(n); + let s = schedule(&pkts, &gs_cfg()); + assert_eq!(s.burst_len, 0, "n={n}: GameStream has no burst stage"); + assert_eq!(s.chunk, chunk, "n={n}: chunk size"); + assert_eq!(s.steps, steps, "n={n}: step count"); + assert!(s.steps <= 12, "n={n}: step count bounded"); + assert!(s.chunk >= 16, "n={n}: chunk floor"); + assert!(s.chunk * s.steps >= n, "n={n}: layout covers all packets"); + } + // The legacy test's exact anchors. + let s = schedule(&packets(1, 1024), &gs_cfg()); + assert_eq!((s.chunk, s.steps), (16, 1)); + let s = schedule(&packets(16, 1024), &gs_cfg()); + assert_eq!((s.chunk, s.steps), (16, 1)); + assert!(schedule(&packets(610, 1024), &gs_cfg()).steps <= 12); + } + + /// The executed chunk sequence follows the schedule exactly, on both parameterizations — + /// zero budget, so the test never sleeps. + #[test] + fn pace_frame_sends_the_scheduled_chunk_sequence() { + // Native, 40 × 1 KB with a 10 KB cap: packets 0..=9 burst (cum hits 10 KB at #10), + // then 30 overflow → chunks of 16: [10..26), [26..40). + let pkts = packets(40, 1024); + let mut seen: Vec = Vec::new(); + let stat = pace_frame( + &pkts, + PaceBudget::Fixed(Duration::ZERO), + &native_cfg(10 * 1024), + |chunk| { + seen.push(chunk.len()); + Ok::<(), std::io::Error>(()) + }, + ) + .unwrap(); + assert_eq!(seen, vec![10, 16, 14]); + assert!(stat.paced); + + // Native, frame under the cap: one immediate burst (chunked at 16), nothing paced. + let pkts = packets(20, 100); + let mut seen: Vec = Vec::new(); + let stat = pace_frame( + &pkts, + PaceBudget::Fixed(Duration::ZERO), + &native_cfg(128 * 1024), + |chunk| { + seen.push(chunk.len()); + Ok::<(), std::io::Error>(()) + }, + ) + .unwrap(); + assert_eq!(seen, vec![16, 4]); + assert!(!stat.paced); + + // GameStream, 146 packets: chunk = max(16, ceil(146/12)=13) = 16 → 10 paced chunks. + let pkts = packets(146, 1024); + let mut seen: Vec = Vec::new(); + pace_frame( + &pkts, + PaceBudget::Fixed(Duration::ZERO), + &gs_cfg(), + |chunk| { + seen.push(chunk.len()); + Ok::<(), std::io::Error>(()) + }, + ) + .unwrap(); + assert_eq!(seen.len(), 10); + assert_eq!(seen.iter().sum::(), 146); + assert!(seen[..9].iter().all(|&c| c == 16)); + assert_eq!(*seen.last().unwrap(), 2); + + // A send error aborts the frame and propagates. + let pkts = packets(64, 1024); + let mut calls = 0; + let r = pace_frame( + &pkts, + PaceBudget::Fixed(Duration::ZERO), + &gs_cfg(), + |_chunk| { + calls += 1; + if calls == 2 { + Err(std::io::Error::other("client gone")) + } else { + Ok(()) + } + }, + ); + assert!(r.is_err()); + assert_eq!(calls, 2, "no sends after the failing chunk"); + } + + /// The sleep targets are each paced chunk's fraction of the budget — pinned against the + /// legacy formulas of both planes (native: `budget×(j+1)/m` directly; GameStream: + /// `(budget×1/steps)×(i+1)`, which agrees to sub-step-count nanoseconds). + #[test] + fn sleep_targets_match_legacy_formulas() { + let budget = Duration::from_micros(12_500); // GS: ¾ of a 60 Hz frame interval + for steps in [1usize, 2, 10, 12] { + for j in 0..steps { + let unified = budget.mul_f64((j + 1) as f64 / steps as f64); + // Native legacy: one fused fraction — identical expression. + assert_eq!(unified, budget.mul_f64((j + 1) as f64 / steps as f64)); + // GameStream legacy: per_step rounds to ns first; ≤ steps/2 ns apart. + let gs_legacy = budget.mul_f64(1.0 / steps as f64).mul_f64((j + 1) as f64); + let diff = unified.abs_diff(gs_legacy); + assert!( + diff <= Duration::from_nanos(steps as u64), + "steps={steps} j={j}: {diff:?} off legacy" + ); + } + } + } + + /// `inject_video_drop` is a no-op when the knob is off (the default test env). + #[test] + fn drop_injection_off_by_default() { + let mut pkts = packets(100, 64); + assert_eq!(inject_video_drop(&mut pkts), 0); + assert_eq!(pkts.len(), 100); + } + + #[test] + fn percentile_picks_expected_ranks() { + let mut v = vec![90, 10, 50, 70, 30]; + assert_eq!(percentile(&mut v, 0.0), 10); + assert_eq!(percentile(&mut v, 0.5), 50); + assert_eq!(percentile(&mut v, 0.99), 90); + assert_eq!(percentile(&mut [], 0.5), 0); + } +}