From aedee2a4e34c9a25ff520495c9704af568dc334d Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 17 Jul 2026 19:12:36 +0200 Subject: [PATCH] perf(latency): tier-0 attribution + tier-1 send-path levers from the latency plan design/latency-reduction-2026-07.md T0.1/T0.2/T1.2/T1.3: - T1.2 rate-capped front-loaded pacing: the paced overflow's budget is now min(0.9x slack, overflow wire time at ~3x the live encoder bitrate) (PUNKTFUNK_PACE_FACTOR, 0 = legacy deadline-only spread). A 300 KB-1 MB frame's tail leaves in ~2-5 ms instead of smearing across ~15 ms at 60 fps; GameStream schedule byte-identical (pins unchanged). - T1.3 data-first wire order: packetize emits every block's data shards before any parity (per-block parity pools keep all blocks' parity alive for the second pass), so lossless completion stops waiting behind the parity tail. EOF = last emitted packet; receiver already order-agnostic. - T0.1 staged 0xCF: HostTiming gains an append-extensible per-stage tail (queue/encode/pace us; seal+channel-wait derived as residual) - no cap bit needed, old peers read the 13-byte prefix. Joined client-side into Stats::host_{queue,encode,xfer,pace}_ms, the OSD detailed tier, and the probe's report. - T0.2 true on-glass present timing: VK_KHR_present_id/present_wait enabled when supported; a PresentTimer waiter thread resolves each present id to real visibility, replacing the submit-time display stamp (which undercounts by up to a refresh and hides a silent-FIFO standing queue). Validated on .21: core 185 + host 185 tests, pf-presenter 19, clippy -D warnings across all five touched crates; loss-harness recovery curve unchanged; C ABI harness round-trips. Co-Authored-By: Claude Fable 5 --- clients/probe/src/main.rs | 31 ++++ crates/pf-client-core/src/session.rs | 46 ++++++ crates/pf-presenter/src/run.rs | 64 +++++++-- crates/pf-presenter/src/vk/mod.rs | 39 ++++++ crates/pf-presenter/src/vk/present.rs | 28 ++-- crates/pf-presenter/src/vk/present_timing.rs | 132 ++++++++++++++++++ crates/pf-presenter/src/vk/reconfig.rs | 8 ++ crates/pf-presenter/src/vk/setup.rs | 41 +++++- crates/punktfunk-core/src/packet/packetize.rs | 115 ++++++++++----- crates/punktfunk-core/src/packet/tests.rs | 88 +++++++++++- crates/punktfunk-core/src/quic/datagram.rs | 45 +++++- crates/punktfunk-core/src/quic/tests.rs | 28 ++++ crates/punktfunk-host/src/native/stream.rs | 67 +++++++-- crates/punktfunk-host/src/send_pacing.rs | 131 +++++++++++++++-- include/punktfunk_core.h | 10 ++ 15 files changed, 779 insertions(+), 94 deletions(-) create mode 100644 crates/pf-presenter/src/vk/present_timing.rs diff --git a/clients/probe/src/main.rs b/clients/probe/src/main.rs index c894afb8..37c553f8 100644 --- a/clients/probe/src/main.rs +++ b/clients/probe/src/main.rs @@ -1241,6 +1241,12 @@ async fn session(args: Args) -> Result<()> { std::collections::VecDeque::new(); let mut host_us_v: Vec = Vec::new(); let mut net_us_v: Vec = Vec::new(); + // T0.1 host-stage split (extended 0xCF): queue/encode/pace + the derived seal/xfer + // residual. Empty against a host that predates the stage tail. + let mut queue_us_v: Vec = Vec::new(); + let mut enc_us_v: Vec = Vec::new(); + let mut xfer_us_v: Vec = Vec::new(); + let mut pace_us_v: Vec = Vec::new(); let mut last_rx = std::time::Instant::now(); let started = std::time::Instant::now(); // Stream-duration cap: `--seconds N`, else the 120s default. Ending the loop here reaches the @@ -1316,6 +1322,14 @@ async fn session(args: Args) -> Result<()> { let (_, hostnet_us) = pending_split.remove(i).unwrap(); host_us_v.push(t.host_us as u64); net_us_v.push(hostnet_us.saturating_sub(t.host_us as u64)); + if let Some(s) = t.stages { + queue_us_v.push(s.queue_us as u64); + enc_us_v.push(s.encode_us as u64); + pace_us_v.push(s.pace_us as u64); + xfer_us_v.push((t.host_us as u64).saturating_sub( + s.queue_us as u64 + s.encode_us as u64 + s.pace_us as u64, + )); + } } } if expected > 0 { @@ -1399,6 +1413,23 @@ async fn session(args: Args) -> Result<()> { "host/network latency split (host = capture→sent on the host; network = wire + \ reassembly)" ); + if !queue_us_v.is_empty() { + // The T0.1 per-stage host attribution: queue (capture→submit) → encode + // (submit→bitstream) → xfer (seal/FEC + send-channel wait, derived residual) + // → pace (the microburst spread). The four tile host_us per frame. + tracing::info!( + stage_samples = queue_us_v.len(), + queue_p50_us = pcts(&mut queue_us_v, 0.50), + queue_p95_us = pcts(&mut queue_us_v, 0.95), + encode_p50_us = pcts(&mut enc_us_v, 0.50), + encode_p95_us = pcts(&mut enc_us_v, 0.95), + xfer_p50_us = pcts(&mut xfer_us_v, 0.50), + xfer_p95_us = pcts(&mut xfer_us_v, 0.95), + pace_p50_us = pcts(&mut pace_us_v, 0.50), + pace_p95_us = pcts(&mut pace_us_v, 0.95), + "host stage split (queue → encode → xfer → pace tile the host figure)" + ); + } } else { tracing::info!("no host timing datagrams (0xCF) — old host; host+network unsplit"); } diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index b6fa23e9..8d2a1e4a 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -86,6 +86,17 @@ pub struct Stats { /// `host + network`. An old host never emits 0xCF, so this stays false and the /// combined stage renders unchanged. pub split: bool, + /// p50 host STAGE split (latency plan T0.1), valid only when `staged`: capture→submit + /// queue age, encoder submit→bitstream, seal/FEC + send-channel wait (the residual + /// `host − queue − encode − pace`), and the paced-send spread. Together they tile + /// `host_ms`, giving per-stage attribution without a host-side log in hand. + pub host_queue_ms: f32, + pub host_encode_ms: f32, + pub host_xfer_ms: f32, + pub host_pace_ms: f32, + /// The window had extended (staged) 0xCF timings — a host older than the stage tail + /// sends the 13-byte form and the OSD keeps the plain `host` figure. + pub staged: bool, /// p50 `decode` stage: received → decode COMPLETE, single-clock client-local (ms). /// Hardware paths measure GPU completion via the frame's timeline fence (an async /// decoder's submission returning in ~0.1 ms is not "decoded"); software measures @@ -361,6 +372,11 @@ fn pump( std::collections::VecDeque::with_capacity(PENDING_SPLIT_CAP); let mut host_us_win: Vec = Vec::with_capacity(256); let mut net_us_win: Vec = Vec::with_capacity(256); + // T0.1 host-stage windows (extended 0xCF only; empty against an older host). + let mut queue_us_win: Vec = Vec::with_capacity(256); + let mut enc_us_win: Vec = Vec::with_capacity(256); + let mut xfer_us_win: Vec = Vec::with_capacity(256); + let mut pace_us_win: Vec = Vec::with_capacity(256); // What actually decoded the last frame — a VAAPI failure demotes mid-session, so // this is read off each frame's image variant rather than fixed at startup. let mut dec_path: &'static str = ""; @@ -658,6 +674,18 @@ fn pump( let (_, hn_us) = pending_split.remove(i).unwrap(); host_us_win.push(t.host_us as u64); net_us_win.push(hn_us.saturating_sub(t.host_us as u64)); + // Extended 0xCF (T0.1): per-stage host split; the seal/FEC + channel-wait + // residual is derived so the four stages tile host_us exactly. + if let Some(s) = t.stages { + queue_us_win.push(s.queue_us as u64); + enc_us_win.push(s.encode_us as u64); + pace_us_win.push(s.pace_us as u64); + xfer_us_win.push( + (t.host_us as u64).saturating_sub( + s.queue_us as u64 + s.encode_us as u64 + s.pace_us as u64, + ), + ); + } } } @@ -691,6 +719,11 @@ fn pump( let split = !host_us_win.is_empty(); let (host_p50, _) = window_percentiles(&mut host_us_win); let (net_p50, _) = window_percentiles(&mut net_us_win); + let staged = !queue_us_win.is_empty(); + let (queue_p50, _) = window_percentiles(&mut queue_us_win); + let (enc_p50, _) = window_percentiles(&mut enc_us_win); + let (xfer_p50, _) = window_percentiles(&mut xfer_us_win); + let (pace_p50, _) = window_percentiles(&mut pace_us_win); let lost = dropped.saturating_sub(window_dropped) as u32; window_dropped = dropped; tracing::debug!( @@ -698,6 +731,10 @@ fn pump( hostnet_p50_us = hn_p50, host_p50_us = host_p50, net_p50_us = net_p50, + queue_p50_us = queue_p50, + encode_p50_us = enc_p50, + xfer_p50_us = xfer_p50, + pace_p50_us = pace_p50, decode_p50_us = dec_p50, lost, total_frames, @@ -710,6 +747,11 @@ fn pump( host_ms: host_p50 as f32 / 1000.0, net_ms: net_p50 as f32 / 1000.0, split, + host_queue_ms: queue_p50 as f32 / 1000.0, + host_encode_ms: enc_p50 as f32 / 1000.0, + host_xfer_ms: xfer_p50 as f32 / 1000.0, + host_pace_ms: pace_p50 as f32 / 1000.0, + staged, decode_ms: dec_p50 as f32 / 1000.0, lost, lost_pct: if lost > 0 { @@ -726,6 +768,10 @@ fn pump( decode_us.clear(); host_us_win.clear(); net_us_win.clear(); + queue_us_win.clear(); + enc_us_win.clear(); + xfer_us_win.clear(); + pace_us_win.clear(); } }; diff --git a/crates/pf-presenter/src/run.rs b/crates/pf-presenter/src/run.rs index d11d7576..02b8578b 100644 --- a/crates/pf-presenter/src/run.rs +++ b/crates/pf-presenter/src/run.rs @@ -1171,28 +1171,50 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result }; if did_present { presented_video = true; - let displayed_ns = session::now_ns(); if opts.json_status && !st.ready_announced { st.ready_announced = true; println!("{{\"ready\":true}}"); } - // The `displayed` stamp (same clamp rules as the pump's windows). - let clock_offset_ns = st - .clock_offset - .as_ref() - .map_or(0, |o| o.load(Ordering::Relaxed)); - let e2e = (displayed_ns as i128 + clock_offset_ns as i128 - pts_ns as i128) - .max(0) as u64; - if e2e > 0 && e2e < 10_000_000_000 { - st.win_e2e_us.push(e2e / 1000); + if presenter.present_timing_active() { + // T0.2: hand the frame's stamps to the present-wait waiter — the + // e2e/display samples arrive via `take_presented_samples` with a + // TRUE on-glass stamp instead of the submit-time one below. + presenter.note_presented(pts_ns, decoded_ns); + } else { + let displayed_ns = session::now_ns(); + // The `displayed` stamp (same clamp rules as the pump's windows). + let clock_offset_ns = st + .clock_offset + .as_ref() + .map_or(0, |o| o.load(Ordering::Relaxed)); + let e2e = (displayed_ns as i128 + clock_offset_ns as i128 - pts_ns as i128) + .max(0) as u64; + if e2e > 0 && e2e < 10_000_000_000 { + st.win_e2e_us.push(e2e / 1000); + } + st.win_disp_us + .push(displayed_ns.saturating_sub(decoded_ns) / 1000); } - st.win_disp_us - .push(displayed_ns.saturating_sub(decoded_ns) / 1000); } } // Fold the presenter window into the shared stats line once per second. if st.win_start.elapsed() >= Duration::from_secs(1) { + // On-glass samples the present-wait waiter completed this window (empty + // when timing is inactive — the legacy submit-time pushes fill in then). + let clock_offset_ns = st + .clock_offset + .as_ref() + .map_or(0, |o| o.load(Ordering::Relaxed)); + for s in presenter.take_presented_samples() { + let e2e = (s.displayed_ns as i128 + clock_offset_ns as i128 - s.pts_ns as i128) + .max(0) as u64; + if e2e > 0 && e2e < 10_000_000_000 { + st.win_e2e_us.push(e2e / 1000); + } + st.win_disp_us + .push(s.displayed_ns.saturating_sub(s.decoded_ns) / 1000); + } let (e2e_p50, e2e_p95) = session::window_percentiles(&mut st.win_e2e_us); let (disp_p50, _) = session::window_percentiles(&mut st.win_disp_us); st.presented = PresentedWindow { @@ -1635,6 +1657,14 @@ fn stats_text( " · decode {:.1} · display {:.1} ms", s.decode_ms, p.display_ms )); + // Extended 0xCF host-stage split (T0.1): its own line so the per-stage attribution + // (queue → encode → seal/xfer → pace) reads as the host pipeline in order. + if s.staged { + text.push_str(&format!( + "\nhost: queue {:.1} · encode {:.1} · xfer {:.1} · pace {:.1} ms", + s.host_queue_ms, s.host_encode_ms, s.host_xfer_ms, s.host_pace_ms + )); + } } if s.lost > 0 { text.push_str(&format!("\nlost {} ({:.1}%)", s.lost, s.lost_pct)); @@ -1830,6 +1860,11 @@ mod tests { host_ms: 1.2, net_ms: 0.9, split: true, + host_queue_ms: 0.3, + host_encode_ms: 0.5, + host_xfer_ms: 0.1, + host_pace_ms: 0.3, + staged: true, decode_ms: 1.8, lost: 3, lost_pct: 0.4, @@ -1866,7 +1901,12 @@ mod tests { let detailed = text(StatsVerbosity::Detailed); assert!(detailed.contains("vulkan · HDR→SDR")); assert!(detailed.contains("host 1.2 · net 0.9 · decode 1.8 · display 1.1 ms")); + assert!(detailed.contains("host: queue 0.3 · encode 0.5 · xfer 0.1 · pace 0.3 ms")); assert!(detailed.contains("lost 3 (0.4%)")); + assert!( + !normal.contains("queue"), + "host-stage split is Detailed-only" + ); } /// Compact omits the latency term until the presenter's first e2e window lands. diff --git a/crates/pf-presenter/src/vk/mod.rs b/crates/pf-presenter/src/vk/mod.rs index c5a05890..546b5eb7 100644 --- a/crates/pf-presenter/src/vk/mod.rs +++ b/crates/pf-presenter/src/vk/mod.rs @@ -28,6 +28,7 @@ use pf_client_core::video::{CpuFrame, VkVideoFrame}; mod gpu; mod overlay_pipe; mod present; +mod present_timing; mod reconfig; mod resources; mod setup; @@ -187,6 +188,16 @@ pub struct Presenter { /// The submit fence has a submission pending (wait before recording again — also /// what makes the single staging buffer safe to overwrite). submitted: bool, + /// `VK_KHR_present_wait` on-glass timing (latency plan T0.2) — `None` when the + /// device lacks the present-id/present-wait pair; the run loop then keeps its + /// submit-time display stamp. + present_timer: Option, + /// Monotonic present id (global counter — strictly increasing per swapchain, which + /// is all the spec asks). 0 = nothing presented with an id yet. + next_present_id: u64, + /// The last successful id-carrying present, awaiting its [`Presenter::note_presented`] + /// claim from the run loop (which owns the frame's pts/decode stamps). + last_presented: Option<(vk::SwapchainKHR, u64)>, } impl Presenter { @@ -220,6 +231,30 @@ impl Presenter { unsafe { self.device.device_wait_idle() }.ok(); } + /// True when `VK_KHR_present_wait` drives the display stamp — the run loop then + /// defers its e2e/display windows to [`Presenter::take_presented_samples`] instead + /// of stamping at `present()` return. + pub(crate) fn present_timing_active(&self) -> bool { + self.present_timer.is_some() + } + + /// Claim the just-submitted present for on-glass timing. Call right after a + /// `present()` that returned `true`, with that frame's capture + decode stamps + /// (the presenter itself never sees them). No-op when timing is inactive. + pub(crate) fn note_presented(&mut self, pts_ns: u64, decoded_ns: u64) { + if let (Some(t), Some((sc, id))) = (&self.present_timer, self.last_presented.take()) { + t.enqueue(sc, id, pts_ns, decoded_ns); + } + } + + /// Take the window's completed on-glass samples (empty when timing is inactive). + pub(crate) fn take_presented_samples(&self) -> Vec { + self.present_timer + .as_ref() + .map(|t| t.take_samples()) + .unwrap_or_default() + } + /// The device handles the console-UI overlay renders on (§6.1). Valid for the /// presenter's lifetime; the run loop drops the overlay first. pub fn shared_device(&self) -> SharedDevice { @@ -237,6 +272,10 @@ impl Presenter { impl Drop for Presenter { fn drop(&mut self) { + // The present-wait waiter references the swapchain — stop it (its Drop joins + // after in-flight waits complete, bounded by their 250 ms cap) BEFORE the + // swapchain teardown below. + self.present_timer.take(); unsafe { { // Insurance against a straggling submitter (the run loop joins the diff --git a/crates/pf-presenter/src/vk/present.rs b/crates/pf-presenter/src/vk/present.rs index 5b54f737..710daaa9 100644 --- a/crates/pf-presenter/src/vk/present.rs +++ b/crates/pf-presenter/src/vk/present.rs @@ -569,20 +569,32 @@ impl Presenter { let swapchains = [self.swapchain]; let indices = [index]; let present_sems = [render_sem]; + // On-glass timing (T0.2): attach a monotonically increasing present id the + // PresentTimer's `vkWaitForPresentKHR` resolves to real visibility. + let ids = [self.next_present_id + 1]; + let mut pid_info = vk::PresentIdKHR::default().present_ids(&ids); + let mut present_info = vk::PresentInfoKHR::default() + .wait_semaphores(&present_sems) + .swapchains(&swapchains) + .image_indices(&indices); + if self.present_timer.is_some() { + self.next_present_id += 1; + present_info = present_info.push_next(&mut pid_info); + } // Same queue external-sync rule as the submit above. Scoped tightly: the // OUT_OF_DATE arm re-enters the lock via recreate_swapchain's queue drain. let present_res = { let _q = self.queue_lock.guard(); - self.swap_d.queue_present( - self.queue, - &vk::PresentInfoKHR::default() - .wait_semaphores(&present_sems) - .swapchains(&swapchains) - .image_indices(&indices), - ) + self.swap_d.queue_present(self.queue, &present_info) }; match present_res { - Ok(_) => Ok(true), + Ok(_) => { + // A failed present's id may never signal — claimable only on Ok. + if self.present_timer.is_some() { + self.last_presented = Some((self.swapchain, self.next_present_id)); + } + Ok(true) + } Err(vk::Result::ERROR_OUT_OF_DATE_KHR) => { self.recreate_swapchain(window)?; Ok(false) diff --git a/crates/pf-presenter/src/vk/present_timing.rs b/crates/pf-presenter/src/vk/present_timing.rs new file mode 100644 index 00000000..ac8cbc4a --- /dev/null +++ b/crates/pf-presenter/src/vk/present_timing.rs @@ -0,0 +1,132 @@ +//! True on-glass present timing via `VK_KHR_present_wait` (latency plan T0.2). +//! +//! The render loop's `displayed` stamp is taken when `vkQueuePresentKHR` *returns* — CPU +//! submit time, excluding the presentation engine's queue and the vblank latch, so the +//! HUD's `display` stage under-reports by up to a refresh (and hides a silent-FIFO +//! standing queue entirely). When the device offers `VK_KHR_present_id` + +//! `VK_KHR_present_wait`, each present carries a monotonically increasing id and a +//! dedicated waiter thread blocks in `vkWaitForPresentKHR` — which completes when the +//! image is actually visible — stamping the REAL on-glass time off the render loop. +//! +//! Lifecycle: `vkWaitForPresentKHR` requires the swapchain to stay alive for the call's +//! duration, so [`PresentTimer::drain`] must run before any `vkDestroySwapchainKHR` +//! (recreate and teardown both do). Waits carry a 250 ms cap: presentation ids complete +//! in submission order (a MAILBOX-replaced image's id completes with the present that +//! replaced it), so a wait only outlives that cap when the pipeline is already wedged — +//! the timeout keeps the drain bounded rather than wedging a resize with it. + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; + +use ash::vk; + +/// One presented frame's identity + the true on-glass stamp the waiter filled in. +pub(crate) struct PresentedSample { + /// The frame's capture stamp (host clock) — the e2e anchor. + pub pts_ns: u64, + /// Decode-complete stamp (client clock) — the display-stage anchor. + pub decoded_ns: u64, + /// `vkWaitForPresentKHR` completion = the image is visible (client clock). + pub displayed_ns: u64, +} + +struct Job { + swapchain: vk::SwapchainKHR, + present_id: u64, + pts_ns: u64, + decoded_ns: u64, +} + +/// The waiter: a channel-fed thread turning (swapchain, present-id) pairs into +/// [`PresentedSample`]s. One frame in flight upstream keeps the queue depth ~1. +pub(crate) struct PresentTimer { + tx: Option>, + /// Jobs enqueued but not yet finished — the drain barrier for swapchain teardown. + pending: Arc, + results: Arc>>, + join: Option>, +} + +impl PresentTimer { + pub(crate) fn spawn(wait_d: ash::khr::present_wait::Device) -> Self { + let (tx, rx) = mpsc::channel::(); + let pending = Arc::new(AtomicUsize::new(0)); + let results = Arc::new(Mutex::new(Vec::with_capacity(256))); + let (pending_t, results_t) = (pending.clone(), results.clone()); + let join = std::thread::Builder::new() + .name("pf-present-wait".into()) + .spawn(move || { + while let Ok(job) = rx.recv() { + // 250 ms cap — see the module doc's lifecycle note. + let r = unsafe { + wait_d.wait_for_present(job.swapchain, job.present_id, 250_000_000) + }; + if r.is_ok() { + let displayed_ns = pf_client_core::session::now_ns(); + results_t.lock().unwrap().push(PresentedSample { + pts_ns: job.pts_ns, + decoded_ns: job.decoded_ns, + displayed_ns, + }); + } + // SUBOPTIMAL/TIMEOUT/DEVICE_LOST: no sample; the frame still showed + // (or the loop is about to find out) — never poison the window. + pending_t.fetch_sub(1, Ordering::AcqRel); + } + }) + .expect("spawn pf-present-wait"); + PresentTimer { + tx: Some(tx), + pending, + results, + join: Some(join), + } + } + + /// Hand a successfully submitted present to the waiter. + pub(crate) fn enqueue( + &self, + swapchain: vk::SwapchainKHR, + present_id: u64, + pts_ns: u64, + decoded_ns: u64, + ) { + if let Some(tx) = &self.tx { + self.pending.fetch_add(1, Ordering::AcqRel); + if tx + .send(Job { + swapchain, + present_id, + pts_ns, + decoded_ns, + }) + .is_err() + { + self.pending.fetch_sub(1, Ordering::AcqRel); + } + } + } + + /// Block until no wait references any swapchain — REQUIRED before + /// `vkDestroySwapchainKHR`. Bounded by the waiter's own 250 ms wait cap. + pub(crate) fn drain(&self) { + while self.pending.load(Ordering::Acquire) > 0 { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + } + + /// Take the window's completed samples (called at the 1 s stats fold). + pub(crate) fn take_samples(&self) -> Vec { + std::mem::take(&mut *self.results.lock().unwrap()) + } +} + +impl Drop for PresentTimer { + fn drop(&mut self) { + // Close the channel, let in-flight waits finish (bounded), then join. + self.tx.take(); + if let Some(j) = self.join.take() { + let _ = j.join(); + } + } +} diff --git a/crates/pf-presenter/src/vk/reconfig.rs b/crates/pf-presenter/src/vk/reconfig.rs index 5b587ac6..f1b4b807 100644 --- a/crates/pf-presenter/src/vk/reconfig.rs +++ b/crates/pf-presenter/src/vk/reconfig.rs @@ -71,6 +71,14 @@ impl Presenter { // The old swapchain and everything tied to its images dies NOW: the fence // quiesce covered our own command buffers, the queue drain above covered the // presentation engine's semaphore waits — nothing can still reference them. + // The present-wait waiter is the one remaining referent: `vkWaitForPresentKHR` + // requires the swapchain alive for the call, so drain it first (bounded by the + // waiter's 250 ms cap; ids complete in order so this is normally instant). + if let Some(t) = &self.present_timer { + t.drain(); + } + // An unclaimed last present belonged to the dying swapchain — drop the claim. + self.last_presented = None; let (overlay_views, overlay_framebuffers) = self.overlay_pipe.take_targets(); unsafe { for fb in overlay_framebuffers { diff --git a/crates/pf-presenter/src/vk/setup.rs b/crates/pf-presenter/src/vk/setup.rs index 536bb51b..fb28a7be 100644 --- a/crates/pf-presenter/src/vk/setup.rs +++ b/crates/pf-presenter/src/vk/setup.rs @@ -129,17 +129,30 @@ impl Presenter { let dev_props = unsafe { instance.get_physical_device_properties(pdev) }; let dev_is_13 = vk::api_version_major(dev_props.api_version) > 1 || vk::api_version_minor(dev_props.api_version) >= 3; + let mut have_pid = vk::PhysicalDevicePresentIdFeaturesKHR::default(); + let mut have_pwait = vk::PhysicalDevicePresentWaitFeaturesKHR::default(); let mut have_f11 = vk::PhysicalDeviceVulkan11Features::default(); let mut have_f12 = vk::PhysicalDeviceVulkan12Features::default(); let mut have_f13 = vk::PhysicalDeviceVulkan13Features::default(); + // Present-id/present-wait (on-glass timing, latency plan T0.2): query the feature + // structs only when the device lists both extensions. + let present_wait_exts = + has(ash::khr::present_id::NAME) && has(ash::khr::present_wait::NAME); let mut have_f2 = vk::PhysicalDeviceFeatures2::default() .push_next(&mut have_f11) .push_next(&mut have_f12) .push_next(&mut have_f13); + if present_wait_exts { + have_f2 = have_f2.push_next(&mut have_pid).push_next(&mut have_pwait); + } unsafe { instance.get_physical_device_features2(pdev, &mut have_f2) }; - // Copy the one base-features fact out NOW: `have_f2` mutably borrows the 11/12/13 - // structs through its pNext chain, so any later use of it would pin those borrows. + // Copy the one base-features fact out NOW: `have_f2` mutably borrows the chained + // structs through its pNext chain, so any later use of it would pin those borrows — + // every read of a chained struct below must come after this, have_f2's last use. let have_shader_int16 = have_f2.features.shader_int16; + let present_wait_ok = present_wait_exts + && have_pid.present_id == vk::TRUE + && have_pwait.present_wait == vk::TRUE; let features_ok = have_f11.sampler_ycbcr_conversion == vk::TRUE && have_f12.timeline_semaphore == vk::TRUE && have_f13.synchronization2 == vk::TRUE; @@ -224,6 +237,15 @@ impl Presenter { ); } + // Present-id/present-wait: enable when fully supported — the presenter then runs + // the on-glass PresentTimer; otherwise the display stamp stays submit-time. + if present_wait_ok { + dev_exts.push(ash::khr::present_id::NAME.as_ptr()); + dev_exts.push(ash::khr::present_wait::NAME.as_ptr()); + } + let mut en_pid = vk::PhysicalDevicePresentIdFeaturesKHR::default().present_id(true); + let mut en_pwait = vk::PhysicalDevicePresentWaitFeaturesKHR::default().present_wait(true); + // Enable only the features the video path needs, and only where supported // (harmless when the path is off; reported to FFmpeg via device_features). let mut en_f11 = vk::PhysicalDeviceVulkan11Features::default() @@ -240,6 +262,9 @@ impl Presenter { .push_next(&mut en_f11) .push_next(&mut en_f12) .push_next(&mut en_f13); + if present_wait_ok { + en_f2 = en_f2.push_next(&mut en_pid).push_next(&mut en_pwait); + } en_f2.features.shader_int16 = if pyrowave_ok { vk::TRUE } else { vk::FALSE }; let priorities = [1.0f32]; @@ -265,6 +290,15 @@ impl Presenter { } .context("vkCreateDevice")?; let swap_d = ash::khr::swapchain::Device::new(&instance, &device); + let present_timer = present_wait_ok.then(|| { + super::present_timing::PresentTimer::spawn(ash::khr::present_wait::Device::new( + &instance, &device, + )) + }); + tracing::info!( + present_wait = present_wait_ok, + "on-glass present timing (VK_KHR_present_wait)" + ); let hdr_metadata_d = has_hdr_metadata.then(|| ash::ext::hdr_metadata::Device::new(&instance, &device)); let queue = unsafe { device.get_device_queue(qfi, 0) }; @@ -441,6 +475,9 @@ impl Presenter { staging: None, video: None, submitted: false, + present_timer, + next_present_id: 0, + last_presented: None, }; p.recreate_swapchain(window)?; Ok(p) diff --git a/crates/punktfunk-core/src/packet/packetize.rs b/crates/punktfunk-core/src/packet/packetize.rs index b614844c..0112b527 100644 --- a/crates/punktfunk-core/src/packet/packetize.rs +++ b/crates/punktfunk-core/src/packet/packetize.rs @@ -32,10 +32,13 @@ pub struct Packetizer { /// Every other data shard is a `shard_payload`-sized slice straight into the frame buffer — /// blocks are consecutive shard ranges, so only the frame's last shard can be partial. tail: Vec, - /// Reusable parity buffers for [`ErasureCoder::encode_into`] (plan Phase 1.4): grows once - /// to the session's high-water recovery count, then every block's parity is generated - /// into it with zero allocations. - recovery: Vec>, + /// Reusable PER-BLOCK parity buffers for [`ErasureCoder::encode_into`] (plan Phase 1.4): + /// one pool per block index, each growing once to its high-water recovery count, then + /// every frame's parity is generated into them with zero allocations. Per-block (not one + /// shared pool) because the data-first wire order (latency plan T1.3) emits every block's + /// DATA shards before any block's parity — all blocks' parity must stay alive until the + /// frame's second emission pass. + recovery: Vec>>, } impl Packetizer { @@ -103,6 +106,15 @@ impl Packetizer { /// ([`Session::seal_frame`](crate::session::Session::seal_frame)). An `emit` error aborts /// the frame mid-way (packet numbering has already advanced — callers treat it as fatal). /// + /// Wire order is DATA-FIRST (latency plan T1.3): every block's data shards in block order, + /// then every block's parity shards in block order. In the lossless case a frame completes + /// the moment its last DATA shard arrives, so the completion-gating packet no longer sits + /// behind the parity tail of the paced spread (~fec% of the spread saved). The receiver is + /// order-agnostic (headers are self-describing; the reassembler completes each block on + /// `data + recovery ≥ k`), so this is not a wire-format change. `FLAG_SOF` stays on the + /// first emitted packet (block 0, shard 0); `FLAG_EOF` marks the last emitted packet — + /// the final parity shard, or the final data shard of a FEC-free frame. + /// /// `frame_index`: `Some(i)` stamps the AU with the caller's index — the punktfunk/1 encode /// loop numbers video AUs itself so the encoder's RFI bookkeeping (LTR marks, DPB timestamps) /// is 1:1 with what the client sees, surviving encoder rebuilds/resets that restart internal @@ -152,7 +164,6 @@ impl Packetizer { self.tail[..rem].copy_from_slice(&frame[full_shards * payload..]); } let tail = &self.tail; - let recovery_pool = &mut self.recovery; let shard_at = |s: usize| -> &[u8] { if s < full_shards { &frame[s * payload..(s + 1) * payload] @@ -160,41 +171,30 @@ impl Packetizer { tail.as_slice() } }; + // Per-block shard geometry (deterministic — recomputed in both passes). + let block_data_count = |b: usize| ((b + 1) * max_block).min(total_data) - b * max_block; + // One parity pool per block, reused across frames (steady-state zero-alloc). + if self.recovery.len() < block_count { + self.recovery.resize_with(block_count, Vec::new); + } + + // Total parity across the frame decides where FLAG_EOF lands (the last emitted packet). + let mut total_recovery = 0usize; for b in 0..block_count { - let first = b * max_block; - let last = ((b + 1) * max_block).min(total_data); - let block_data_count = last - first; - - // This block's data shards: references into `frame` (plus the staged tail). - let data_shards: Vec<&[u8]> = (first..last).map(shard_at).collect(); - - let recovery_count = self.fec.recovery_for(block_data_count); - coder.encode_into(&data_shards, recovery_count, recovery_pool)?; - let recovery = &*recovery_pool; - let total_shards = block_data_count + recovery_count; - if total_shards > u16::MAX as usize { + let k = block_data_count(b); + let m = self.fec.recovery_for(k); + if k + m > u16::MAX as usize { return Err(PunktfunkError::Unsupported("block shard count exceeds u16")); } + total_recovery += m; + } - for shard_index in 0..total_shards { - let body: &[u8] = if shard_index < block_data_count { - data_shards[shard_index] - } else { - &recovery[shard_index - block_data_count] - }; - - let seq = self.next_seq; - self.next_seq = self.next_seq.wrapping_add(1); - - let mut flags = FLAG_PIC; - if b == 0 && shard_index == 0 { - flags |= FLAG_SOF; - } - if b + 1 == block_count && shard_index + 1 == total_shards { - flags |= FLAG_EOF; - } - + let mut emit_one = + |next_seq: &mut u32, b: usize, shard_index: usize, body: &[u8], flags: u8| { + let seq = *next_seq; + *next_seq = next_seq.wrapping_add(1); + let k = block_data_count(b); let hdr = PacketHeader { pts_ns, frame_index, @@ -203,8 +203,8 @@ impl Packetizer { user_flags, block_index: b as u16, block_count: block_count as u16, - data_shards: block_data_count as u16, - recovery_shards: recovery_count as u16, + data_shards: k as u16, + recovery_shards: self.fec.recovery_for(k) as u16, shard_index: shard_index as u16, shard_bytes: payload as u16, magic: PUNKTFUNK_MAGIC, @@ -212,9 +212,48 @@ impl Packetizer { fec_scheme: coder.scheme() as u8, flags, }; - emit(&hdr, body)?; + emit(&hdr, body) + }; + let mut next_seq = self.next_seq; + + // Pass 1 — per block: generate parity into the block's pool, emit the DATA shards. + for b in 0..block_count { + let first = b * max_block; + let k = block_data_count(b); + + // This block's data shards: references into `frame` (plus the staged tail). + let data_shards: Vec<&[u8]> = (first..first + k).map(shard_at).collect(); + let recovery_count = self.fec.recovery_for(k); + coder.encode_into(&data_shards, recovery_count, &mut self.recovery[b])?; + + for (shard_index, body) in data_shards.iter().enumerate() { + let mut flags = FLAG_PIC; + if b == 0 && shard_index == 0 { + flags |= FLAG_SOF; + } + if total_recovery == 0 && b + 1 == block_count && shard_index + 1 == k { + flags |= FLAG_EOF; + } + emit_one(&mut next_seq, b, shard_index, body, flags)?; } } + + // Pass 2 — per block: emit the parity shards (the frame's tail on the wire). + let mut parity_left = total_recovery; + for b in 0..block_count { + let k = block_data_count(b); + let recovery_count = self.fec.recovery_for(k); + for r in 0..recovery_count { + parity_left -= 1; + let mut flags = FLAG_PIC; + if parity_left == 0 { + flags |= FLAG_EOF; + } + let body: &[u8] = &self.recovery[b][r]; + emit_one(&mut next_seq, b, k + r, body, flags)?; + } + } + self.next_seq = next_seq; Ok(()) } } diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 1f90bc57..2736aac6 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -370,16 +370,94 @@ fn e2e_roundtrip( /// 100 bytes / 16 = 7 shards → blocks of (4 data + 2 rec) and (3 data + 2 rec). #[test] fn e2e_multiblock_loss_reorder_dup_gf16() { - // Packet order: blk0 = idx 0..6 (4 data + 2 rec), blk1 = idx 6..11 (3 data + 2 rec). + // Data-first wire order (T1.3): blk0 data = idx 0..4, blk1 data = idx 4..7, + // blk0 rec = idx 7..9, blk1 rec = idx 9..11. // Kill 2 data in block 0 and 1 data in block 1 — all within the 50% budget. - e2e_roundtrip(FecScheme::Gf16, 100, 50, &[0, 2, 7], false); - e2e_roundtrip(FecScheme::Gf16, 100, 50, &[0, 2, 7], true); + e2e_roundtrip(FecScheme::Gf16, 100, 50, &[0, 2, 5], false); + e2e_roundtrip(FecScheme::Gf16, 100, 50, &[0, 2, 5], true); } #[test] fn e2e_multiblock_loss_reorder_dup_gf8() { - e2e_roundtrip(FecScheme::Gf8, 100, 50, &[1, 3, 8], false); - e2e_roundtrip(FecScheme::Gf8, 100, 50, &[1, 3, 8], true); + e2e_roundtrip(FecScheme::Gf8, 100, 50, &[1, 3, 6], false); + e2e_roundtrip(FecScheme::Gf8, 100, 50, &[1, 3, 6], true); +} + +/// T1.3 pin: the wire order is DATA-FIRST — every block's data shards in block order, then +/// every block's parity in block order — so the lossless-completion-gating packet (the last +/// data shard) never sits behind parity in the paced spread. SOF on the first emitted packet, +/// EOF on the last (a parity shard whenever the frame carries FEC). +#[test] +fn packetize_emits_all_data_before_any_parity() { + use zerocopy::FromBytes; + let cfg = e2e_config(FecScheme::Gf16, 50); + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + // 100 B / 16 → 7 data shards → blocks (4 data + 2 rec) + (3 data + 2 rec). + let src: Vec = (0..100).map(|i| (i * 31 + 3) as u8).collect(); + let pkts = pk.packetize(&src, 1, 0, coder.as_ref()).unwrap(); + assert_eq!(pkts.len(), 11); + let hdrs: Vec = pkts + .iter() + .map(|p| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap()) + .collect(); + // (block_index, shard_index) in emission order. + let layout: Vec<(u16, u16)> = hdrs + .iter() + .map(|h| (h.block_index, h.shard_index)) + .collect(); + assert_eq!( + layout, + vec![ + (0, 0), + (0, 1), + (0, 2), + (0, 3), // blk0 data + (1, 0), + (1, 1), + (1, 2), // blk1 data + (0, 4), + (0, 5), // blk0 parity + (1, 3), + (1, 4), // blk1 parity + ], + "data-first wire order" + ); + // A shard is parity iff shard_index >= data_shards; no parity may precede any data. + let first_parity = hdrs + .iter() + .position(|h| h.shard_index >= h.data_shards) + .unwrap(); + assert!( + hdrs[first_parity..] + .iter() + .all(|h| h.shard_index >= h.data_shards), + "no data shard after the first parity shard" + ); + // Stream seqs stay strictly sequential in emission order (the nonce contract). + for (i, w) in hdrs.windows(2).enumerate() { + assert_eq!(w[1].stream_seq, w[0].stream_seq + 1, "seq gap at {i}"); + } + assert_eq!(hdrs[0].flags & FLAG_SOF, FLAG_SOF, "SOF on first packet"); + assert_eq!( + hdrs.last().unwrap().flags & FLAG_EOF, + FLAG_EOF, + "EOF on last (parity) packet" + ); + assert_eq!( + hdrs.iter().filter(|h| h.flags & FLAG_EOF != 0).count(), + 1, + "exactly one EOF" + ); + + // FEC-free frame: EOF falls on the last data shard instead. + let cfg0 = e2e_config(FecScheme::Gf16, 0); + let mut pk0 = Packetizer::new(&cfg0); + let pkts0 = pk0.packetize(&src, 2, 0, coder.as_ref()).unwrap(); + assert_eq!(pkts0.len(), 7, "no parity at 0% FEC"); + let last = PacketHeader::read_from_bytes(&pkts0.last().unwrap()[..HEADER_LEN]).unwrap(); + assert_eq!(last.flags & FLAG_EOF, FLAG_EOF, "EOF on last data shard"); + assert!(last.shard_index < last.data_shards, "last packet is data"); } /// Zero losses, in order: the pure fast path (no codec call, recovered == 0) must still diff --git a/crates/punktfunk-core/src/quic/datagram.rs b/crates/punktfunk-core/src/quic/datagram.rs index 2d094b95..a1bbcad3 100644 --- a/crates/punktfunk-core/src/quic/datagram.rs +++ b/crates/punktfunk-core/src/quic/datagram.rs @@ -545,28 +545,63 @@ pub struct HostTiming { /// Host capture→sent duration, µs (saturated at `u32::MAX` ≈ 71 min — far past the 10 s /// client-side sanity clamp anyway). pub host_us: u32, + /// Per-stage split of `host_us` (latency plan T0.1). `None` from a host that predates the + /// extended datagram — the 0xCF wire is APPEND-extensible (decode reads the 13-byte prefix + /// and takes the stage tail only when present), so no capability bit is needed in either + /// direction: old client + new host reads the prefix, new client + old host gets `None`. + pub stages: Option, } -/// Wire length of a [`HOST_TIMING_MAGIC`] datagram: tag + u64 pts + u32 µs = 13 bytes. -const HOST_TIMING_LEN: usize = 1 + 8 + 4; +/// The extended 0xCF's per-stage split of [`HostTiming::host_us`], all µs against the same +/// capture anchor. The stages tile the host pipeline as +/// `host_us = queue + encode + (seal/FEC + channel-wait = the residual) + pace`, so the client +/// derives the residual as `host_us − queue_us − encode_us − pace_us` — no fifth field needed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct HostStages { + /// Capture delivery → encoder submit (the capture ring / channel-queue age; 0 for + /// re-encoded hold frames, which never waited). + pub queue_us: u32, + /// Encoder submit → bitstream ready (scheduling wait + ASIC time). + pub encode_us: u32, + /// Paced send: first byte handed to the socket → last packet sent (the microburst spread). + pub pace_us: u32, +} -/// Encode a [`HostTiming`] into a [`HOST_TIMING_MAGIC`] datagram. +/// Wire length of a legacy [`HOST_TIMING_MAGIC`] datagram: tag + u64 pts + u32 µs = 13 bytes. +const HOST_TIMING_LEN: usize = 1 + 8 + 4; +/// Wire length with the [`HostStages`] tail appended: + 3 × u32 = 25 bytes. +const HOST_TIMING_STAGES_LEN: usize = HOST_TIMING_LEN + 12; + +/// Encode a [`HostTiming`] into a [`HOST_TIMING_MAGIC`] datagram (extended form when `stages` +/// is set — an older client parses the prefix and ignores the tail). pub fn encode_host_timing_datagram(t: &HostTiming) -> Vec { - let mut b = Vec::with_capacity(HOST_TIMING_LEN); + let mut b = Vec::with_capacity(HOST_TIMING_STAGES_LEN); b.push(HOST_TIMING_MAGIC); b.extend_from_slice(&t.pts_ns.to_le_bytes()); b.extend_from_slice(&t.host_us.to_le_bytes()); + if let Some(s) = &t.stages { + b.extend_from_slice(&s.queue_us.to_le_bytes()); + b.extend_from_slice(&s.encode_us.to_le_bytes()); + b.extend_from_slice(&s.pace_us.to_le_bytes()); + } b } /// Parse a [`HOST_TIMING_MAGIC`] datagram → [`HostTiming`]. `None` on bad tag or a short buffer -/// (the fixed length bounds every read before it happens). +/// (the fixed lengths bound every read before it happens). A datagram carrying only the 13-byte +/// prefix (an older host) yields `stages: None`. pub fn decode_host_timing_datagram(b: &[u8]) -> Option { if b.len() < HOST_TIMING_LEN || b[0] != HOST_TIMING_MAGIC { return None; } + let stages = (b.len() >= HOST_TIMING_STAGES_LEN).then(|| HostStages { + queue_us: u32::from_le_bytes(b[13..17].try_into().unwrap()), + encode_us: u32::from_le_bytes(b[17..21].try_into().unwrap()), + pace_us: u32::from_le_bytes(b[21..25].try_into().unwrap()), + }); Some(HostTiming { pts_ns: u64::from_le_bytes(b[1..9].try_into().unwrap()), host_us: u32::from_le_bytes(b[9..13].try_into().unwrap()), + stages, }) } diff --git a/crates/punktfunk-core/src/quic/tests.rs b/crates/punktfunk-core/src/quic/tests.rs index da3c74e9..502e0e0d 100644 --- a/crates/punktfunk-core/src/quic/tests.rs +++ b/crates/punktfunk-core/src/quic/tests.rs @@ -266,6 +266,7 @@ fn host_timing_datagram_roundtrip_and_truncation() { let t = HostTiming { pts_ns: 1_751_500_000_123_456_789, // a realistic 2026 CLOCK_REALTIME capture stamp host_us: 4_321, + stages: None, }; let d = encode_host_timing_datagram(&t); assert_eq!(d[0], HOST_TIMING_MAGIC); @@ -278,6 +279,33 @@ fn host_timing_datagram_roundtrip_and_truncation() { let mut bad = d.clone(); bad[0] = HDR_META_MAGIC; assert_eq!(decode_host_timing_datagram(&bad), None); + + // Extended form (T0.1): the stage tail roundtrips; a truncated tail (an old host's 13-byte + // datagram, or anything short of the full 25) degrades to `stages: None`, never a partial + // read; the prefix fields stay identical in both forms (the append-extensibility contract). + let ts = HostTiming { + stages: Some(HostStages { + queue_us: 900, + encode_us: 3_100, + pace_us: 2_500, + }), + ..t + }; + let ds = encode_host_timing_datagram(&ts); + assert_eq!(ds.len(), 25); + assert_eq!( + &ds[..13], + &d[..13], + "prefix is byte-identical to the legacy form" + ); + assert_eq!(decode_host_timing_datagram(&ds), Some(ts)); + for n in 13..ds.len() { + assert_eq!( + decode_host_timing_datagram(&ds[..n]), + Some(t), + "partial stage tail ({n} B) must degrade to the legacy decode" + ); + } } #[test] diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 1c1dfbd3..c6029167 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -60,6 +60,7 @@ pub(super) fn synthetic_stream( let t = punktfunk_core::quic::HostTiming { pts_ns, host_us: (now_ns().saturating_sub(pts_ns) / 1000).min(u32::MAX as u64) as u32, + stages: None, // synthetic loop: no capture/encode stages to split }; let _ = tc.send_datagram(punktfunk_core::quic::encode_host_timing_datagram(&t).into()); } @@ -195,18 +196,26 @@ fn service_probes( /// 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 across ~90% of the time to `deadline` in ADAPTIVE -/// chunks — 16 packets at today's rates, coarsening to at most 64 (the GSO-segment cap) once -/// the rate would otherwise skip every sub-floor sleep, so ≥1 Gbps frames still pace instead -/// of collapsing into an unpaced blast (plan Phase 1.2). `burst_cap` `None` = auto: -/// `max(128 KB, this AU's wire bytes / 4)`, so the burst stays a bounded fraction of a -/// high-rate frame instead of swallowing it whole (plan Phase 1.3); `Some` = -/// PUNKTFUNK_PACE_BURST_KB pinned an absolute cap. 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. +/// only the OVERFLOW beyond that is spread across `min(~90% of the time to deadline, the time +/// the overflow needs at pace_rate_bps)` in ADAPTIVE chunks — 16 packets at today's rates, +/// coarsening to at most 64 (the GSO-segment cap) once the rate would otherwise skip every +/// sub-floor sleep, so ≥1 Gbps frames still pace instead of collapsing into an unpaced blast +/// (plan Phase 1.2). `burst_cap` `None` = auto: `max(128 KB, this AU's wire bytes / 4)`, so +/// the burst stays a bounded fraction of a high-rate frame instead of swallowing it whole +/// (plan Phase 1.3); `Some` = PUNKTFUNK_PACE_BURST_KB pinned an absolute cap. 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. +/// +/// `pace_rate_bps` (latency plan T1.2) bounds the spread from above: the deadline term alone +/// smears a big frame's tail across the whole remaining interval (~15 ms at 60 fps) even when +/// the link could drain it in 2–3 ms. The caller passes ~3× the live encoder bitrate — a rate +/// the link is proven to carry sustained, so the bounded excursion keeps the anti-freeze +/// property while the tail leaves as soon as the link plausibly allows. `0` = uncapped +/// (legacy smoothness-only spread, and the fallback when the bitrate isn't known yet). #[allow(clippy::too_many_arguments)] fn paced_submit( session: &mut Session, @@ -216,6 +225,7 @@ fn paced_submit( frame_index: u32, deadline: std::time::Instant, burst_cap: Option, + pace_rate_bps: u64, ) -> Result { let wires = session .seal_frame_at(data, pts_ns, flags, frame_index) @@ -224,11 +234,22 @@ fn paced_submit( // FEC/recovery test knob (PUNKTFUNK_VIDEO_DROP) — same knob the GameStream plane honors. crate::send_pacing::inject_video_drop(&mut refs); let wire_bytes: usize = refs.iter().map(|p| p.len()).sum(); + let burst_bytes = burst_cap.unwrap_or_else(|| (wire_bytes / 4).max(128 * 1024)); let cfg = crate::send_pacing::PaceCfg { - burst_bytes: Some(burst_cap.unwrap_or_else(|| (wire_bytes / 4).max(128 * 1024))), + burst_bytes: Some(burst_bytes), chunk: crate::send_pacing::ChunkPolicy::Adaptive { base: 16, max: 64 }, sleep_floor: std::time::Duration::from_micros(500), }; + // T1.2 rate cap: the overflow's wire time at `pace_rate_bps`. Only the bytes past the + // burst pace at all, so only they bound the budget. + let overflow_bytes = wire_bytes.saturating_sub(burst_bytes) as u64; + let cap = if pace_rate_bps > 0 && overflow_bytes > 0 { + std::time::Duration::from_nanos( + (overflow_bytes * 8).saturating_mul(1_000_000_000) / pace_rate_bps, + ) + } else { + std::time::Duration::MAX + }; // Time the socket handoff per chunk and fold it into the session's SealPerf split — the // sleeps between chunks stay excluded, so sock_ns is pure send_gso/sendmmsg time. let mut sock_ns = 0u64; @@ -237,6 +258,7 @@ fn paced_submit( crate::send_pacing::PaceBudget::UntilDeadline { deadline, fraction: 0.9, + cap, }, &cfg, |chunk| { @@ -352,6 +374,15 @@ fn send_loop( probe_seq: bool, ) { boost_thread_priority(false); // transmit thread: above-normal (Apollo's encoder-thread level) + // T1.2 front-loaded pacing: the paced overflow drains at `factor ×` the live encoder + // bitrate instead of stretching to the frame deadline. 3× default (the link carries 1× + // sustained, so a bounded 3× excursion is safe — WebRTC's pacer uses 2.5×); + // `PUNKTFUNK_PACE_FACTOR=0` restores the legacy deadline-only spread. + let pace_factor: f64 = std::env::var("PUNKTFUNK_PACE_FACTOR") + .ok() + .and_then(|s| s.parse().ok()) + .filter(|f: &f64| f.is_finite() && *f >= 0.0) + .unwrap_or(3.0); let mut last_perf = std::time::Instant::now(); let mut last_bytes = 0u64; let mut last_send_dropped = 0u64; @@ -391,6 +422,8 @@ fn send_loop( msg.frame_index, msg.deadline, burst_cap, + // Live ABR-tracked encoder bitrate → pace rate; 0 (not yet known) = uncapped. + (stats.bitrate_kbps.load(Ordering::Relaxed) as f64 * 1000.0 * pace_factor) as u64, ) { Ok(stat) => { // First VIDEO packets are on the wire — complete the bring-up trace (P0.1; @@ -411,6 +444,14 @@ fn send_loop( let t = punktfunk_core::quic::HostTiming { pts_ns: msg.capture_ns, host_us, + // T0.1 stage split: queue + encode ride the FrameMsg (always + // measured), pace is this send's spread. The client derives + // seal/FEC + channel-wait as the residual against host_us. + stages: Some(punktfunk_core::quic::HostStages { + queue_us: msg.queue_us, + encode_us: msg.encode_us, + pace_us: stat.spread_us, + }), }; let _ = tc.send_datagram( punktfunk_core::quic::encode_host_timing_datagram(&t).into(), diff --git a/crates/punktfunk-host/src/send_pacing.rs b/crates/punktfunk-host/src/send_pacing.rs index 78b245ce..eca69999 100644 --- a/crates/punktfunk-host/src/send_pacing.rs +++ b/crates/punktfunk-host/src/send_pacing.rs @@ -10,11 +10,14 @@ //! deterministic-schedule tests below): //! //! * **native** — the first `burst_bytes` leave immediately (one absorbed microburst), only the -//! overflow is paced across 90 % of the time left to the frame deadline in ADAPTIVE chunks: -//! 16 packets at today's rates, coarsening just enough that the per-chunk interval clears the -//! sleep floor (≤ 64, the GSO-segment cap) once the rate would otherwise skip every sleep — -//! so ≥1 Gbps frames still pace instead of blasting (no slack ⇒ budget 0 ⇒ never slower than -//! unpaced); +//! overflow is paced across `min(90 % of the time left to the frame deadline, the time the +//! overflow needs at ~3× the live stream bitrate)` in ADAPTIVE chunks: 16 packets at today's +//! rates, coarsening just enough that the per-chunk interval clears the sleep floor (≤ 64, +//! the GSO-segment cap) once the rate would otherwise skip every sleep — so ≥1 Gbps frames +//! still pace instead of blasting (no slack ⇒ budget 0 ⇒ never slower than unpaced). The +//! rate cap (latency plan T1.2) front-loads the spread: the link demonstrably carries 1× the +//! stream rate sustained, so a bounded 3× excursion is safe and a large frame's tail stops +//! waiting out the whole interval; //! * **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 @@ -54,8 +57,17 @@ pub(crate) enum ChunkPolicy { /// 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 }, + /// `min((deadline − now-after-burst) × fraction, cap)`, collapsing to 0 with no slack + /// (native: fraction 0.9). `cap` bounds the spread to the time the overflow actually needs + /// at a rate the link is proven to carry (latency plan T1.2): the deadline term alone + /// smears a large frame across the whole remaining interval even when the link could drain + /// it in a fraction of that — `Duration::MAX` = uncapped (the legacy smoothness-only + /// schedule). + UntilDeadline { + deadline: Instant, + fraction: f32, + cap: Duration, + }, /// A precomputed fixed budget (GameStream: ¾ of the frame interval). Fixed(Duration), } @@ -157,10 +169,15 @@ pub(crate) fn pace_frame, E>( // (it overshoots the post-burst budget by the burst's few µs — harmless, sub-floor sleeps // are skipped anyway). let budget_est = match budget { - PaceBudget::UntilDeadline { deadline, fraction } => deadline + PaceBudget::UntilDeadline { + deadline, + fraction, + cap, + } => deadline .checked_duration_since(start) .unwrap_or_default() - .mul_f32(fraction), + .mul_f32(fraction) + .min(cap), PaceBudget::Fixed(d) => d, }; let sched = schedule(packets, cfg, budget_est); @@ -171,10 +188,15 @@ pub(crate) fn pace_frame, E>( if paced { let pace_start = Instant::now(); let budget = match budget { - PaceBudget::UntilDeadline { deadline, fraction } => deadline + PaceBudget::UntilDeadline { + deadline, + fraction, + cap, + } => deadline .checked_duration_since(pace_start) .unwrap_or_default() - .mul_f32(fraction), + .mul_f32(fraction) + .min(cap), PaceBudget::Fixed(d) => d, }; for (j, chunk) in packets[sched.burst_len..].chunks(sched.chunk).enumerate() { @@ -489,6 +511,93 @@ mod tests { } } + /// The T1.2 rate cap bounds an `UntilDeadline` budget from above: with ample deadline + /// slack the cap decides the spread (and therefore the adaptive chunk sizing); a + /// `Duration::MAX` cap reproduces the legacy deadline-only schedule exactly. + #[test] + fn until_deadline_cap_bounds_the_budget() { + let cfg = PaceCfg { + burst_bytes: Some(12_000), + chunk: ChunkPolicy::Adaptive { base: 16, max: 64 }, + sleep_floor: Duration::from_micros(500), + }; + // 210 × 1200 B: 10 burst, 200 overflow (the adaptive test's canonical frame). + let pkts = packets(210, 1200); + + // Zero cap + far deadline: the budget collapses to 0 → blast schedule (max chunks, + // no sleeps) even though the deadline alone would have spread ~90 ms. + let mut seen: Vec = Vec::new(); + let stat = pace_frame( + &pkts, + PaceBudget::UntilDeadline { + deadline: Instant::now() + Duration::from_millis(100), + fraction: 0.9, + cap: Duration::ZERO, + }, + &cfg, + |chunk| { + seen.push(chunk.len()); + Ok::<(), std::io::Error>(()) + }, + ) + .unwrap(); + assert_eq!(seen, vec![10, 64, 64, 64, 8], "zero cap = blast schedule"); + assert!(stat.paced); + assert!( + stat.spread_us < 50_000, + "zero cap must not sleep toward the deadline" + ); + + // A 2.5 ms cap under a ~90 ms deadline budget: the cap sizes the chunks + // (c ≥ 200 × 500 µs / 2.5 ms = 40) and the frame drains in ~2.5 ms, not ~90. + let mut seen: Vec = Vec::new(); + let stat = pace_frame( + &pkts, + PaceBudget::UntilDeadline { + deadline: Instant::now() + Duration::from_millis(100), + fraction: 0.9, + cap: Duration::from_micros(2_500), + }, + &cfg, + |chunk| { + seen.push(chunk.len()); + Ok::<(), std::io::Error>(()) + }, + ) + .unwrap(); + assert_eq!( + seen, + vec![10, 40, 40, 40, 40, 40], + "cap drives chunk sizing" + ); + assert!( + stat.spread_us < 50_000, + "capped spread must be ~2.5 ms, nowhere near the 90 ms deadline budget" + ); + + // MAX cap = legacy: no-slack deadline still collapses to the blast path. + let mut seen: Vec = Vec::new(); + pace_frame( + &pkts, + PaceBudget::UntilDeadline { + deadline: Instant::now(), + fraction: 0.9, + cap: Duration::MAX, + }, + &cfg, + |chunk| { + seen.push(chunk.len()); + Ok::<(), std::io::Error>(()) + }, + ) + .unwrap(); + assert_eq!( + seen, + vec![10, 64, 64, 64, 8], + "MAX cap = legacy no-slack blast" + ); + } + /// `inject_video_drop` is a no-op when the knob is off (the default test env). #[test] fn drop_injection_off_by_default() { diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index f06e7704..bf4c2c69 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -926,6 +926,12 @@ // The client's wire (protocol) version does not match the host's — one side needs updating. #define WIRE_VERSION_CLOSE_CODE 103 +// Minimum supported multiplier (renders under native, upscaled on present). +#define MIN_SCALE 0.5 + +// Maximum supported multiplier (supersamples, clamped to the codec ceiling per axis). +#define MAX_SCALE 4.0 + // Stable C ABI status codes. `Ok` is 0; all errors are negative so callers can // test `rc < 0`. Do not renumber existing variants — only append. enum PunktfunkStatus @@ -1370,6 +1376,10 @@ typedef struct { +// The multipliers a picker offers. `1.0` (Native) is the default; the rest are the round stops +// users reason about. Shared so every client's list stays identical. +#define PRESETS { 0.5, 0.67, 0.75, 1.0, 1.25, 1.5, 2.0, 3.0, 4.0, } + #ifdef __cplusplus extern "C" { #endif // __cplusplus