perf(latency): tier-0 attribution + tier-1 send-path levers from the latency plan
ci / web (push) Successful in 49s
ci / docs-site (push) Successful in 54s
decky / build-publish (push) Successful in 18s
docker / build-push (--build-arg FEDORA_VERSION=44, ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm) (push) Successful in 10s
docker / build-push (., web/Dockerfile, punktfunk-web) (push) Successful in 9s
docker / build-push (ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Successful in 7s
docker / build-push (ci, ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Successful in 9s
docker / build-push (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Successful in 8s
apple / swift (push) Successful in 1m23s
ci / rust (push) Failing after 4m15s
arch / build-publish (push) Failing after 4m35s
deb / build-publish (push) Failing after 4m4s
docker / deploy-docs (push) Successful in 24s
ci / bench (push) Successful in 5m35s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Failing after 3m45s
windows-host / package (push) Failing after 9m8s
release / apple (push) Successful in 5m49s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Failing after 5m25s
flatpak / build-publish (push) Failing after 8m3s
windows-msix / package (arm64, C:\Users\Public\ffmpeg-arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Successful in 3m58s
android / android (push) Successful in 14m9s
apple / screenshots (push) Successful in 6m28s
windows-msix / package (x64, C:\Users\Public\ffmpeg, , x86_64-pc-windows-msvc, C:\t) (push) Successful in 7m0s
windows / build (aarch64-pc-windows-msvc) (push) Successful in 9m28s
windows / build (x86_64-pc-windows-msvc) (push) Successful in 12m26s
ci / web (push) Successful in 49s
ci / docs-site (push) Successful in 54s
decky / build-publish (push) Successful in 18s
docker / build-push (--build-arg FEDORA_VERSION=44, ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm) (push) Successful in 10s
docker / build-push (., web/Dockerfile, punktfunk-web) (push) Successful in 9s
docker / build-push (ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Successful in 7s
docker / build-push (ci, ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Successful in 9s
docker / build-push (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Successful in 8s
apple / swift (push) Successful in 1m23s
ci / rust (push) Failing after 4m15s
arch / build-publish (push) Failing after 4m35s
deb / build-publish (push) Failing after 4m4s
docker / deploy-docs (push) Successful in 24s
ci / bench (push) Successful in 5m35s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Failing after 3m45s
windows-host / package (push) Failing after 9m8s
release / apple (push) Successful in 5m49s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Failing after 5m25s
flatpak / build-publish (push) Failing after 8m3s
windows-msix / package (arm64, C:\Users\Public\ffmpeg-arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Successful in 3m58s
android / android (push) Successful in 14m9s
apple / screenshots (push) Successful in 6m28s
windows-msix / package (x64, C:\Users\Public\ffmpeg, , x86_64-pc-windows-msvc, C:\t) (push) Successful in 7m0s
windows / build (aarch64-pc-windows-msvc) (push) Successful in 9m28s
windows / build (x86_64-pc-windows-msvc) (push) Successful in 12m26s
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 <noreply@anthropic.com>
This commit is contained in:
@@ -1241,6 +1241,12 @@ async fn session(args: Args) -> Result<()> {
|
||||
std::collections::VecDeque::new();
|
||||
let mut host_us_v: Vec<u64> = Vec::new();
|
||||
let mut net_us_v: Vec<u64> = 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<u64> = Vec::new();
|
||||
let mut enc_us_v: Vec<u64> = Vec::new();
|
||||
let mut xfer_us_v: Vec<u64> = Vec::new();
|
||||
let mut pace_us_v: Vec<u64> = 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");
|
||||
}
|
||||
|
||||
@@ -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<u64> = Vec::with_capacity(256);
|
||||
let mut net_us_win: Vec<u64> = Vec::with_capacity(256);
|
||||
// T0.1 host-stage windows (extended 0xCF only; empty against an older host).
|
||||
let mut queue_us_win: Vec<u64> = Vec::with_capacity(256);
|
||||
let mut enc_us_win: Vec<u64> = Vec::with_capacity(256);
|
||||
let mut xfer_us_win: Vec<u64> = Vec::with_capacity(256);
|
||||
let mut pace_us_win: Vec<u64> = 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();
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -1171,28 +1171,50 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result<Option<Outcome>
|
||||
};
|
||||
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.
|
||||
|
||||
@@ -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<present_timing::PresentTimer>,
|
||||
/// 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<present_timing::PresentedSample> {
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<mpsc::Sender<Job>>,
|
||||
/// Jobs enqueued but not yet finished — the drain barrier for swapchain teardown.
|
||||
pending: Arc<AtomicUsize>,
|
||||
results: Arc<Mutex<Vec<PresentedSample>>>,
|
||||
join: Option<std::thread::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl PresentTimer {
|
||||
pub(crate) fn spawn(wait_d: ash::khr::present_wait::Device) -> Self {
|
||||
let (tx, rx) = mpsc::channel::<Job>();
|
||||
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<PresentedSample> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<u8>,
|
||||
/// 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<Vec<u8>>,
|
||||
/// 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<Vec<Vec<u8>>>,
|
||||
}
|
||||
|
||||
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(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<u8> = (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<PacketHeader> = 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
|
||||
|
||||
@@ -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<HostStages>,
|
||||
}
|
||||
|
||||
/// 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<u8> {
|
||||
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<HostTiming> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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<usize>,
|
||||
pace_rate_bps: u64,
|
||||
) -> Result<PaceStat> {
|
||||
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(),
|
||||
|
||||
@@ -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<T: AsRef<[u8]>, 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<T: AsRef<[u8]>, 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<usize> = 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<usize> = 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<usize> = 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() {
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user