unom/punktfunk
1
0
Fork 0
Code Issues Pull Requests Actions 8 Packages Projects Releases 20 Wiki Activity

refactor(host): shared send-pacing policy for the native and GameStream video planes

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Enrico Bühler
2026-07-10 16:26:41 +02:00
parent baa04d2d24
commit d1770c3476
4 changed files with 468 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/punktfunk-host/src/gamestream/stream.rs
+28 -99
View File
@@ -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<u8>]) -> 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<PacketBatch>,
frame_interval: Duration,
running: Arc<AtomicBool>,
drop_pct: u32,
) -> Result<()> {
std::thread::Builder::new()
.name("punktfunk-send".into())
@@ -507,53 +493,38 @@ 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) {
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;
}
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);
}
}
}
}
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::<RawFrame>(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);
}
}
crates/punktfunk-host/src/main.rs
+1
View File
@@ -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;
crates/punktfunk-host/src/punktfunk1.rs
+27 -99
View File
@@ -120,6 +120,7 @@ fn bind_data_socket(data_port: Option<u16>) -> 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<u32> = 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::<u32>().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)"
// 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(|_| ()),
);
}
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;
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
crates/punktfunk-host/src/send_pacing.rs
+408
View File
@@ -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<usize>,
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<T: AsRef<[u8]>>(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<T: AsRef<[u8]>, E>(
packets: &[T],
budget: PaceBudget,
cfg: &PaceCfg,
mut send: impl FnMut(&[T]) -> Result<(), E>,
) -> Result<PaceStat, E> {
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<u32> = std::sync::OnceLock::new();
*PCT.get_or_init(|| {
let pct = std::env::var("PUNKTFUNK_VIDEO_DROP")
.ok()
.and_then(|s| s.parse::<u32>().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<T>(packets: &mut Vec<T>) -> 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<Vec<u8>> {
(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<usize> = 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<usize> = 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<usize> = 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<usize> = 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::<usize>(), 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);
}
}