Merge pull request 'PyroWave on Linux: the GPU-priority knob never fired, a dmabuf timeout condemned the host, and the jumbo grow was dead code' (#132) from worktree-wave2-pyrowave into main
apple / swift (push) Successful in 1m35s
android / android (push) Canceled after 3m13s
apple / screenshots (push) Canceled after 0s
arch / build-publish (push) Canceled after 3m9s
ci / rust (push) Canceled after 2m42s
ci / rust-arm64 (push) Canceled after 1m10s
ci / web (push) Canceled after 3s
ci / docs-site (push) Canceled after 0s
ci / bun-nix (push) Canceled after 0s
deb / build-publish (push) Canceled after 2s
deb / build-publish-host (push) Canceled after 0s
deb / build-publish-client-arm64 (push) Canceled after 0s
docker / builders (--build-arg FEDORA_VERSION=44, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm, -f44) (push) Canceled after 0s
docker / builders (ci/android-ci.Dockerfile, punktfunk-android-ci) (push) Canceled after 0s
docker / builders (ci/arch-ci.Dockerfile, punktfunk-arch-ci) (push) Canceled after 0s
docker / builders (ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Canceled after 0s
docker / builders (ci/rust-ci-noble.Dockerfile, punktfunk-rust-ci-noble) (push) Canceled after 0s
docker / builders (ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Canceled after 0s
docker / builders-arm64cross (push) Canceled after 0s
docker / apps (., web/Dockerfile, punktfunk-web) (push) Canceled after 0s
docker / apps (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Canceled after 0s
docker / deploy-docs (push) Canceled after 0s
flatpak / build-publish (push) Canceled after 8s
nix / flake (push) Canceled after 7s
release / apple (push) Canceled after 5m6s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Canceled after 15s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Canceled after 14s
windows-host / package (push) Canceled after 3m27s
windows-host / canary-manifest (push) Canceled after 0s
windows-host / winget-source (push) Canceled after 0s
windows-msix / package (arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Canceled after 0s
windows-msix / package (x64, , x86_64-pc-windows-msvc, C:\t) (push) Canceled after 0s
windows / build (aarch64-pc-windows-msvc) (push) Canceled after 0s
windows / build (x86_64-pc-windows-msvc) (push) Canceled after 0s

Reviewed-on: #132
This commit was merged in pull request #132.
This commit is contained in:
2026-08-08 23:23:12 +00:00
34 changed files with 4704 additions and 477 deletions
+23
View File
@@ -168,6 +168,10 @@ pub struct PortalCapturer {
/// downgrade ([`pf_zerocopy::note_raw_dmabuf_negotiation_failed`]) so the pipeline rebuild
/// retries on the CPU offer instead of failing identically forever.
vaapi_dmabuf: bool,
/// PW3: this capture's dmabuf offer has been confirmed to negotiate (a frame arrived), so the
/// negotiation retry budget has already been credited back. One-shot — the credit is per
/// capture, not per frame.
negotiation_confirmed: bool,
/// This capture ran the HDR (10-bit PQ/BT.2020 dmabuf) offer — see [`Self::open`]'s
/// `want_hdr`. Read by the negotiation-timeout diagnosis (a failed HDR offer latches the
/// process-wide SDR downgrade) and by [`hdr_meta`](Capturer::hdr_meta).
@@ -412,6 +416,7 @@ impl PwHandles {
signals: self.signals,
stall_since: None,
vaapi_dmabuf: self.vaapi_dmabuf,
negotiation_confirmed: false,
hdr_offer: self.hdr_offer,
hdr_source,
node_id,
@@ -468,6 +473,13 @@ fn spawn_pipewire(
} else {
want_hdr
};
// PW3: tell the raw-dmabuf latch which capture this is BEFORE reading its verdict below. A
// different node id is a different question — a fresh virtual output, a compositor restart,
// the Bazzite Gaming↔Desktop switch — and inheriting "dmabuf does not work here" from an
// unrelated capture is how one transient timeout used to cost a host CPU capture until it was
// restarted. The portal bit is in the key because a portal-fd capture and a virtual-output
// capture with the same node number are genuinely different sources.
pf_zerocopy::note_raw_dmabuf_capture(u64::from(node_id) | (u64::from(fd.is_some()) << 32));
// THE negotiation decision, resolved once here and handed to the thread — no mirror (L3/F1).
// Every environment/latch read the decision depends on happens at this single point.
let plan = pipewire::negotiation_plan(pipewire::NegotiationInputs {
@@ -705,6 +717,7 @@ impl PortalCapturer {
// The slot before the wakeup: a publish that coalesced its edge (or landed while we were
// not waiting) is still visible here.
if let Some(f) = self.take_frame() {
self.note_negotiation_confirmed();
return Ok(f);
}
let slice = Duration::from_millis(500)
@@ -728,6 +741,16 @@ impl PortalCapturer {
self.slot.lock().ok().and_then(|mut s| s.take())
}
/// PW3: a frame arrived, so this capture's dmabuf-only offer DID negotiate — credit the
/// negotiation retry budget back. Only meaningful for a capture that actually made that offer,
/// and only once per capture (the budget counts consecutive failed BUILDS, not frames).
fn note_negotiation_confirmed(&mut self) {
if self.vaapi_dmabuf && !self.negotiation_confirmed {
self.negotiation_confirmed = true;
pf_zerocopy::note_raw_dmabuf_negotiation_ok();
}
}
/// The [`frame_within`](Self::frame_within) budget expired (or the thread ended) — turn it
/// into the diagnosis-bearing error. Split out of the slicing loop above; behavior unchanged.
fn next_frame_timed_out(
File diff suppressed because it is too large Load Diff
+90 -6
View File
@@ -288,16 +288,57 @@ pub(super) fn build_shm_only_buffers() -> Result<Vec<u8>> {
})
}
/// Build a Buffers param requesting dmabuf-only buffers.
/// PW5 stage 2: the buffer-pool depth we ASK for on the zero-copy path, as a Choice range.
///
/// The zero-copy path hands the SPA buffer back to the producer at `.process` return, while the
/// encode thread still holds a dup of its dmabuf fd and has not yet imported, let alone read, the
/// contents. Nothing bounds that window — see the `queue_raw_buffer` comment in `pipewire.rs` — so
/// the only thing that keeps capture untorn is the producer round-robining a pool deeper than our
/// import+encode latency. Until PW5 stage 1 nobody had ever counted what that pool was; we never
/// even asked for a size (`build_dmabuf_buffers` set `dataType` and nothing else).
///
/// A **range**, deliberately, not a fixed count: SPA intersects the consumer's and producer's
/// Buffers params, so a fixed 8 against a producer that can only afford 4 empties the intersection
/// and the link silently stalls in "negotiating" — the exact failure mode the cursor-meta `size`
/// property already cost this codebase once (see `build_cursor_meta_param`). With a range the
/// producer clamps into it and negotiation still succeeds.
///
/// The numbers: `min` stays at 2 so nothing that works today stops working; `default` 8 is ~133 ms
/// of buffer at 60 Hz and ~33 ms at 240 Hz, comfortably past the ~3-4 ms capture→fence latency
/// measured in PW3/PW4 even with a second frame in flight; `max` 16 is a ceiling, not a request
/// (a 4K 4:4:4 buffer is ~25 MB, so 16 is ~400 MB of compositor allocation and worth capping).
/// **What the producer actually picks is logged by the stage-1 census — trust that line, not
/// these constants.**
const POOL_MIN: i32 = 2;
const POOL_DEFAULT: i32 = 8;
const POOL_MAX: i32 = 16;
/// Build a Buffers param requesting dmabuf-only buffers, with pool headroom (see [`POOL_DEFAULT`]).
pub(super) fn build_dmabuf_buffers() -> Result<Vec<u8>> {
serialize_pod(pw::spa::pod::Object {
type_: pw::spa::utils::SpaTypes::ObjectParamBuffers.as_raw(),
id: pw::spa::param::ParamType::Buffers.as_raw(),
properties: vec![pw::spa::pod::Property {
key: pw::spa::sys::SPA_PARAM_BUFFERS_dataType,
flags: pw::spa::pod::PropertyFlags::empty(),
value: pw::spa::pod::Value::Int(1i32 << pw::spa::sys::SPA_DATA_DmaBuf),
}],
properties: vec![
pw::spa::pod::Property {
key: pw::spa::sys::SPA_PARAM_BUFFERS_dataType,
flags: pw::spa::pod::PropertyFlags::empty(),
value: pw::spa::pod::Value::Int(1i32 << pw::spa::sys::SPA_DATA_DmaBuf),
},
pw::spa::pod::Property {
key: pw::spa::sys::SPA_PARAM_BUFFERS_buffers,
flags: pw::spa::pod::PropertyFlags::empty(),
value: pw::spa::pod::Value::Choice(pw::spa::pod::ChoiceValue::Int(
pw::spa::utils::Choice(
pw::spa::utils::ChoiceFlags::empty(),
pw::spa::utils::ChoiceEnum::Range {
default: POOL_DEFAULT,
min: POOL_MIN,
max: POOL_MAX,
},
),
)),
},
],
})
}
@@ -512,4 +553,47 @@ mod tests {
"libspa renumbered spa_video_transfer_function — update the hardcoded PQ id"
);
}
/// PW5 stage 2: the pool request must be a **Choice Range**, never a fixed Int.
///
/// This is the whole safety argument for asking at all: SPA intersects the two sides' Buffers
/// params, so a fixed count a producer cannot afford empties the intersection and the link
/// stalls in "negotiating" with no error anywhere — the same trap that cost this codebase the
/// entire Linux cursor channel once (see `build_cursor_meta_param`). Asserting the pod shape
/// is what keeps a later "simplify" from turning the range back into a number.
#[test]
fn the_dmabuf_pool_request_is_a_range_not_a_fixed_count() {
let pod = build_dmabuf_buffers().unwrap();
let key = spa::sys::SPA_PARAM_BUFFERS_buffers.to_ne_bytes();
let at = pod
.windows(4)
.position(|w| w == key)
.expect("the dmabuf Buffers pod must carry a buffers count");
let word = |off: usize| u32::from_ne_bytes(pod[off..off + 4].try_into().unwrap());
// Property = { key, flags, value_pod }; value_pod = { size, type, body }. A Choice body
// is { type: u32, flags: u32, child_size: u32, child_type: u32, values… }.
assert_eq!(
word(at + 12),
spa::sys::SPA_TYPE_Choice,
"the buffers count must be a Choice, not a bare Int — a fixed count can fail \
negotiation outright"
);
assert_eq!(
word(at + 16),
spa::sys::SPA_CHOICE_Range,
"the Choice must be a Range (default, min, max)"
);
assert_eq!(word(at + 24), 4, "Choice child pods are 4-byte Ints");
assert_eq!(word(at + 28), spa::sys::SPA_TYPE_Int, "…of type Int");
let vals: Vec<i32> = (0..3)
.map(|i| i32::from_ne_bytes(pod[at + 32 + i * 4..at + 36 + i * 4].try_into().unwrap()))
.collect();
assert_eq!(
vals,
vec![POOL_DEFAULT, POOL_MIN, POOL_MAX],
"Range values are serialized default-first"
);
// The minimum must not exceed what producers already serve, or the ask becomes a demand.
const { assert!(POOL_MIN <= 2) };
}
}
File diff suppressed because it is too large Load Diff
+320
View File
@@ -53,6 +53,24 @@ pub(crate) fn stamp_color_bits(bitstream: &mut [u8], seq_offset: usize, bt2020_p
}
}
/// Read the 3-bit wire sequence counter out of a pyrowave block header.
///
/// Every block header is `{ u16 ballot; u16 payload_words:12, sequence:3, extended:1; u32 ... }`
/// (`pyrowave_common.hpp`, `static_assert(sizeof == 8)`), so the counter is bits 12..14 of the
/// little-endian half-word at `packet_offset + 2` — the same word `stamp_color_bits` reaches into
/// from the other end.
///
/// This field is the entire frame-boundary signal on the wire: the decoder restarts a frame only
/// when the value CHANGES (`diff = (hdr.sequence - last_seq) & 0x7; restart = diff != 0`), so a
/// repeated value is read as more blocks of the same frame. That is why PW5's alternating encoder
/// handles need `pyrowave_encoder_set_next_sequence`, and why a test asserts this reader sees
/// +1 mod 8 across the pair.
pub(crate) fn wire_sequence(bitstream: &[u8], packet_offset: usize) -> Option<u8> {
let lo = *bitstream.get(packet_offset + 2)?;
let hi = *bitstream.get(packet_offset + 3)?;
Some(((u16::from_le_bytes([lo, hi]) >> 12) & 0x7) as u8)
}
/// The wavelet block space's total 32x32-block count for a mode — the exact counting walk of
/// upstream `WaveletBuffers::init_block_meta` (also ported to the Apple `WaveletLayout`, whose
/// golden tests pin it against real host AUs). Needed because the vendored RDO pass packs the
@@ -201,6 +219,193 @@ pub(crate) fn build_au(
au
}
// ---------------------------------------------------------------------------
// Streamed-AU chunk cutting (PW6 — latency plan §T3.4, wave-2 plan PW6)
// ---------------------------------------------------------------------------
/// Default per-chunk target — ~34 chunks for a 400 Mb/s 60 fps AU (~833 KB). Deliberately coarse,
/// because the SEALER, not this size, sets how early bytes actually leave:
///
/// * Toward a plain `VIDEO_CAP_STREAMED_AU` client, `Packetizer::push_streamed` flushes only when
/// its pending buffer exceeds one FEC block — `fec.max_data_per_block × shard_payload`, which is
/// 200 × 1408 = 281 600 B on the shipped 1500-MTU IPv4 geometry. Anything smaller than that is
/// simply buffered. (256 KiB sits just under one block, so the first flush lands on the SECOND
/// chunk; the win is intact either way — the whole-AU path seals all ~3 blocks before its first
/// datagram may leave.) Only a client that ALSO negotiated `VIDEO_CAP_MULTI_SLICE` gets the
/// finer `MIN_STREAM_BLOCK_SHARDS` floor (16 shards ≈ 22 KB), where the chunk size does set the
/// flush granularity directly. pf-encode is not told the session's FEC geometry, so this is a
/// fixed byte target rather than a block-derived one.
/// * Chunks are not free: the send thread paces each sealed batch on its own
/// (`stream.rs::pace_sealed`), and every call grants a fresh `max(bytes/4, 128 KiB)` microburst
/// allowance. Cutting an AU into dozens of chunks therefore erodes the pacing this host does to
/// stop line-rate bursts from overrunning the NIC — the failure mode the pacer exists for.
const STREAM_CHUNK_TARGET_BYTES: usize = 256 * 1024;
/// Clamp on the `PUNKTFUNK_PYROWAVE_CHUNK_KIB` override (see [`stream_chunk_step`]).
const STREAM_CHUNK_MIN_KIB: usize = 4;
const STREAM_CHUNK_MAX_KIB: usize = 8192;
/// Whether streamed-AU output is armed for this host process.
///
/// **Default OFF, and deliberately so.** The streamed wire shape costs one PyroWave-specific
/// regression that has not been measured: an UNPINNED streamed frame (its final block never
/// arrived, so `frame_bytes` is still the 0 sentinel) is excluded from partial delivery
/// (`reassemble.rs`, 2026-07 security-review finding 10) — where today's whole-AU path hands the
/// consumer a usable blurred partial, a streamed frame that loses its final block delivers
/// NOTHING. PyroWave clients opt into partial delivery unconditionally
/// (`client/pump/handshake.rs`), so this is a live behaviour change for every one of them. The
/// netem loss-harness leg (2 % on `lo`, FEC pinned off — the Phase-4 recipe) comparing
/// partial-delivery rates streamed vs whole-AU is the prerequisite for flipping the default;
/// until it has run, `PUNKTFUNK_PYROWAVE_STREAMED_AU=1` is how you get it.
///
/// The client's `VIDEO_CAP_STREAMED_AU` and the host's `PUNKTFUNK_STREAMED_AU` remain the outer
/// gates (`stream.rs`) — this only decides whether the ENCODER offers chunks at all.
fn stream_armed() -> bool {
static ARMED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
// Latched once: `supports_chunked_poll` is re-queried per AU, and a knob that could change
// mid-session would flip the wire shape under an open `StreamedAu`.
*ARMED.get_or_init(|| {
matches!(
std::env::var("PUNKTFUNK_PYROWAVE_STREAMED_AU").as_deref(),
Ok("1")
)
})
}
/// Bytes per streamed chunk, rounded DOWN to a whole number of `window`-sized windows (never
/// below one). The rounding is the whole point — see [`AuChunker`].
fn chunk_step(window: usize, target: usize) -> usize {
(target / window.max(1)).max(1) * window.max(1)
}
/// The streamed-AU chunk size for a backend whose wire chunking is `wire_chunk`, or `None` when
/// this session must stay on the whole-AU path — which is the answer whenever the feature is not
/// armed ([`stream_armed`]) or the encoder is in DENSE mode.
///
/// Dense mode is excluded on purpose: there the AU is ONE atomic pyrowave packet with no window
/// framing, so a cut is neither shard-aligned nor a framing boundary. Every real PyroWave session
/// runs datagram-aligned (`stream.rs` sets `plan.wire_chunk = Some(session.shard_payload())`), so
/// nothing is lost — but the invariant this file promises stays true instead of nearly true.
///
/// `PUNKTFUNK_PYROWAVE_CHUNK_KIB` overrides the target (clamped to
/// [`STREAM_CHUNK_MIN_KIB`]..=[`STREAM_CHUNK_MAX_KIB`]); garbage falls back to the default.
pub(crate) fn stream_chunk_step(wire_chunk: Option<usize>) -> Option<usize> {
let window = wire_chunk.filter(|&w| w > 0)?;
if !stream_armed() {
return None;
}
static TARGET: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
let target = *TARGET.get_or_init(|| {
std::env::var("PUNKTFUNK_PYROWAVE_CHUNK_KIB")
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.filter(|k| (STREAM_CHUNK_MIN_KIB..=STREAM_CHUNK_MAX_KIB).contains(k))
.map(|k| k * 1024)
.unwrap_or(STREAM_CHUNK_TARGET_BYTES)
});
Some(chunk_step(window, target))
}
/// Hands a **finished** datagram-aligned AU out in window-aligned pieces for the streamed-AU wire
/// ([`crate::Encoder::poll_chunk`], `punktfunk_core::quic::VIDEO_CAP_STREAMED_AU`). Shared by both
/// pyrowave backends so the cut rule cannot drift between Linux and Windows — the Windows backend
/// cannot even be compiled from a Linux/macOS dev box, so logic written into it directly ships
/// unverified.
///
/// ## What this does NOT buy (read before quoting PW6 as a latency win)
///
/// pyrowave's `encode_frame` is **synchronous**: `submit` returns only once the whole AU sits in
/// `pending`, so by the time the host can poll a chunk the encode is over. `poll_chunk` is
/// therefore NOT "emit slices as the encoder produces them" — it is "hand the finished AU out in
/// pieces so the wire work pipelines with itself". Concretely, what moves:
///
/// * whole-AU path: `Session::seal_frame_at` FEC-protects, packetizes and AEAD-seals the ENTIRE
/// ~830 KB AU before its first datagram may leave the socket;
/// * streamed path: each FEC block seals and paces as it completes, so the first byte reaches the
/// wire after one block's seal, and the remaining seal work overlaps its own transmission.
///
/// There is NO encode/send overlap here — unlike the H.26x sub-frame slice path, where chunks
/// genuinely appear while the encoder is still working. PW6 and PW5 (encode overlap) are
/// independent packages, not sequential ones.
///
/// It also does **not** give the client decode-while-arriving: the reassembler completes a
/// streamed AU exactly like a whole one (`reassemble.rs` — `block_count != 0 && blocks_ok ==
/// block_count`) and hands up ONE `Frame`. Client-side prefix decode is the separate
/// `Session::set_deliver_frame_parts` opt-in, which PyroWave's newest-wins frame channel cannot
/// take — see the PW6 section of `design/linux-host-performance-wave2-pyrowave.md`.
///
/// ## The cut rule
///
/// A chunk is a whole number of `chunk`-sized WINDOWS. [`build_au`] gives every window exactly ONE
/// `kind` in its 4-byte prefix (`WIN_PACKED` or one link of a `WIN_FRAG_*` chain), so a cut inside
/// a window would split a unit the clients parse atomically. Whole windows are `shard_payload`
/// multiples by construction, which is what makes the sealer's sentinel block bases shard-aligned
/// for free (plan §4.4) — the streamed path's placement contract.
pub(crate) struct AuChunker {
au: Vec<u8>,
/// Bytes already handed out.
cursor: usize,
/// Bytes per chunk — a whole number of windows ([`chunk_step`]).
step: usize,
pts_ns: u64,
keyframe: bool,
recovery_anchor: bool,
chunk_aligned: bool,
/// Set once anything has been emitted, so the degenerate EMPTY AU still owes exactly one
/// chunk and not an infinite stream of them.
emitted: bool,
}
impl AuChunker {
pub(crate) fn new(frame: crate::EncodedFrame, step: usize) -> AuChunker {
AuChunker {
au: frame.data,
cursor: 0,
step: step.max(1),
pts_ns: frame.pts_ns,
keyframe: frame.keyframe,
recovery_anchor: frame.recovery_anchor,
chunk_aligned: frame.chunk_aligned,
emitted: false,
}
}
/// The next piece, or `None` once the AU is spent. The pieces concatenate to exactly the bytes
/// [`crate::Encoder::poll`] would have returned; `first` opens the wire frame and `last` closes
/// it (the host's `handle_chunk` keys its `begin`/`finish` off precisely those two).
pub(crate) fn next(&mut self) -> Option<crate::AuChunk> {
if self.cursor >= self.au.len() {
// A zero-byte AU is not reachable through `build_au` (it always emits at least one
// window), but the host would leak its open `StreamedAu` if a chunked poll returned
// nothing at all — so the degenerate case still owes one self-closing chunk.
if self.emitted {
return None;
}
self.emitted = true;
return Some(self.chunk(Vec::new(), true, true));
}
let first = self.cursor == 0;
let end = (self.cursor + self.step).min(self.au.len());
let data = self.au[self.cursor..end].to_vec();
self.cursor = end;
self.emitted = true;
Some(self.chunk(data, first, end == self.au.len()))
}
/// AU-level metadata rides every chunk (the `AuChunk` contract only makes it authoritative on
/// `first`, but a truthful copy on each one costs nothing and keeps a mid-AU log honest).
fn chunk(&self, data: Vec<u8>, first: bool, last: bool) -> crate::AuChunk {
crate::AuChunk {
data,
pts_ns: self.pts_ns,
keyframe: self.keyframe,
recovery_anchor: self.recovery_anchor,
chunk_aligned: self.chunk_aligned,
first,
last,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -362,4 +567,119 @@ mod tests {
stamp_color_bits(&mut bs, 0, true);
assert_eq!(bs[7], 0x78);
}
// --- streamed-AU chunk cutting (PW6) ------------------------------------
// Appended at module END per the wave plan's ownership rule.
fn frame(data: Vec<u8>) -> crate::EncodedFrame {
crate::EncodedFrame {
data,
pts_ns: 1_234_567,
keyframe: true,
recovery_anchor: false,
chunk_aligned: true,
}
}
/// Drain a chunker into `(concatenated bytes, per-chunk lengths, first flags, last flags)`.
fn drain(mut c: AuChunker) -> (Vec<u8>, Vec<usize>, Vec<bool>, Vec<bool>) {
let (mut bytes, mut lens, mut firsts, mut lasts) = (Vec::new(), Vec::new(), vec![], vec![]);
while let Some(ch) = c.next() {
lens.push(ch.data.len());
firsts.push(ch.first);
lasts.push(ch.last);
bytes.extend_from_slice(&ch.data);
assert_eq!(ch.pts_ns, 1_234_567, "AU metadata rides every chunk");
assert!(ch.keyframe && ch.chunk_aligned && !ch.recovery_anchor);
}
(bytes, lens, firsts, lasts)
}
/// The invariant PW6 rests on: chunks concatenate to EXACTLY the AU, every cut lands on a
/// whole-window boundary (so no window's single `kind` is split across two wire frames), and
/// the reassembled stream still walks back to the same codec packets. A cut inside a window
/// would hand the client a 4-byte prefix whose body arrives in a different chunk — the
/// framing is one-kind-per-window, so there is no way to express that.
#[test]
fn stream_chunks_tile_the_au_on_window_boundaries() {
let bs: Vec<u8> = (0..4000u32).map(|i| (i % 251) as u8).collect();
let packets = [(0, 20), (20, 300), (320, 55), (375, 900), (1275, 40)];
let chunk = 64;
let au = build_au(&packets, &bs, Some(chunk));
assert!(au.len() / chunk > 4, "need several windows to cut between");
let step = chunk_step(chunk, 3 * chunk);
assert_eq!(step, 3 * chunk);
let (bytes, lens, firsts, lasts) = drain(AuChunker::new(frame(au.clone()), step));
assert_eq!(bytes, au, "chunks concatenate to exactly the AU");
assert!(
lens.iter().all(|l| l % chunk == 0),
"every chunk is a whole number of windows: {lens:?}"
);
assert!(
lens[..lens.len() - 1].iter().all(|&l| l == step),
"only the tail chunk may be short: {lens:?}"
);
assert_eq!(
firsts,
(0..lens.len()).map(|i| i == 0).collect::<Vec<_>>(),
"exactly one opening chunk"
);
assert_eq!(
lasts,
(0..lens.len())
.map(|i| i + 1 == lens.len())
.collect::<Vec<_>>(),
"exactly one closing chunk"
);
// And the client's parse is unchanged by the cutting.
let mut expect = Vec::new();
for &(o, s) in &packets {
expect.extend_from_slice(&bs[o..o + s]);
}
assert_eq!(walk(&bytes, chunk), expect);
}
/// The step always rounds DOWN to whole windows and never to zero — a target below one window
/// degenerates to one window per chunk rather than an empty chunk (which would spin forever).
#[test]
fn chunk_step_rounds_down_to_whole_windows() {
// 262144 / 1408 = 186.2 → 186 whole windows (261 888 B), never the 262 144 asked for.
assert_eq!(chunk_step(1408, 256 * 1024), 186 * 1408);
assert_eq!(chunk_step(1408, 1408), 1408);
assert_eq!(chunk_step(1408, 1407), 1408); // below one window → one window
assert_eq!(chunk_step(1408, 0), 1408);
assert_eq!(chunk_step(0, 4096), 4096); // defensive: never divides by zero
}
/// An AU that fits one chunk is a single `first && last` piece — the shape the host's
/// `handle_chunk` turns into begin+finish on one message, and byte-identical on the wire to
/// what the whole-AU path would have sealed.
#[test]
fn single_chunk_au_opens_and_closes_itself() {
let au = vec![7u8; 512];
let (bytes, lens, firsts, lasts) = drain(AuChunker::new(frame(au.clone()), 4096));
assert_eq!(bytes, au);
assert_eq!(lens, vec![512]);
assert_eq!(firsts, vec![true]);
assert_eq!(lasts, vec![true]);
}
/// The degenerate empty AU still owes exactly ONE self-closing chunk: a chunked poll that
/// returned nothing would leave the host's `StreamedAu` open forever (its `begin` fires on
/// `first`, its `finish` on `last`).
#[test]
fn empty_au_still_emits_one_self_closing_chunk() {
let mut c = AuChunker::new(frame(Vec::new()), 4096);
let ch = c.next().expect("one chunk");
assert!(ch.first && ch.last && ch.data.is_empty());
assert!(c.next().is_none(), "and never a second one");
}
/// Dense (non-windowed) AUs never stream: there is no window framing to cut on, so a chunk
/// boundary would be neither shard-aligned nor a parse boundary.
#[test]
fn dense_mode_never_streams() {
assert!(stream_chunk_step(None).is_none());
assert!(stream_chunk_step(Some(0)).is_none());
}
}
@@ -128,6 +128,11 @@ pub struct PyroWaveEncoder {
wire_budget: pyrowave_wire::WireBudget,
bitstream: Vec<u8>,
pending: VecDeque<EncodedFrame>,
/// The AU currently being handed out in streamed chunks (PW6 — `Some` strictly between a
/// `first` chunk and its `last`). See [`pyrowave_wire::AuChunker`]: this backend's encode is
/// synchronous, so the AU is COMPLETE before the first chunk leaves — the split is for the
/// send side, never an encode/send overlap.
chunker: Option<pyrowave_wire::AuChunker>,
}
// SAFETY: used only from the single encode thread; the pyrowave handles are owned and only touched
@@ -255,6 +260,7 @@ impl PyroWaveEncoder {
wire_budget: pyrowave_wire::WireBudget::new(),
bitstream: Vec::new(),
pending: VecDeque::new(),
chunker: None,
})
}
}
@@ -676,10 +682,55 @@ impl Encoder for PyroWaveEncoder {
}
fn poll(&mut self) -> Result<Option<EncodedFrame>> {
// Trait contract: each AU is drained through ONE method. Erroring beats double-emitting
// the bytes the chunk cursor already handed out (which would reach the wire twice, under
// the same frame index, and fail the receiver's retro-validation).
if self.chunker.is_some() {
bail!("pyrowave: poll() on an AU already being drained through poll_chunk");
}
Ok(self.pending.pop_front())
}
// --- streamed AU (PW6) — see `pyrowave_wire::AuChunker` for what this does and does NOT buy.
// Byte-identical to the Linux twin BY CONSTRUCTION: all of the cutting lives in the shared
// helper, which compiles and unit-tests on every platform. This file cannot be compiled from
// a Linux/macOS dev box, so anything written here directly would ship unverified.
fn supports_chunked_poll(&self) -> bool {
pyrowave_wire::stream_chunk_step(self.wire_chunk).is_some()
}
fn poll_chunk(&mut self) -> Result<Option<crate::AuChunk>> {
// Finish the AU already in flight before opening the next one — the host's `handle_chunk`
// keys begin/finish off `first`/`last` and cannot interleave two AUs.
if let Some(c) = self.chunker.as_mut() {
if let Some(chunk) = c.next() {
return Ok(Some(chunk));
}
self.chunker = None;
}
let Some(f) = self.pending.pop_front() else {
return Ok(None);
};
// No blocking wait here (the trait allows one): `submit` already ran the whole encode
// synchronously, so an AU in `pending` is complete by construction.
match pyrowave_wire::stream_chunk_step(self.wire_chunk) {
Some(step) => Ok(self
.chunker
.insert(pyrowave_wire::AuChunker::new(f, step))
.next()),
// Unarmed / dense: the trait's own default shape, so a host that polls chunks anyway
// still gets whole AUs.
None => Ok(Some(crate::AuChunk::whole(f))),
}
}
fn reset(&mut self) -> bool {
// A rebuild forfeits every in-flight frame — including an AU only half-handed-out through
// `poll_chunk`. Dropping the cursor here (ahead of every `pending.clear()` arm below) is
// what keeps the next `poll_chunk` from splicing the tail of a dead AU onto a fresh one;
// the host sees a `first` without the previous `last`, logs "streamed AU abandoned
// mid-flight" and lets the client age that frame out.
self.chunker = None;
// Cheap in-place rebuild: recreate only the pyrowave encoder object (no rate-control /
// reference state to preserve). The device, imported textures and fence survive.
// SAFETY: encode is synchronous (no work in flight); the device outlives the swapped encoder.
+322 -28
View File
@@ -19,7 +19,7 @@ pub mod vkslot;
pub mod vulkan;
pub mod worker;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
pub use cuda::DeviceBuffer;
pub use egl::{DmabufPlane, EglImporter};
@@ -261,56 +261,223 @@ pub fn gpu_import_disabled() -> bool {
/// operator found `PUNKTFUNK_ZEROCOPY=0` by hand. The host already knows how to encode that
/// machine — capture just has to stop handing it dmabufs. Latching here is what makes the next
/// session negotiate CPU frames on its own.
static RAW_DMABUF_FAILURE_STREAK: AtomicU32 = AtomicU32::new(0);
static RAW_DMABUF_DISABLED: AtomicBool = AtomicBool::new(false);
/// Below the encoder's own rebuild budget, so the latch is set before the session it doomed ends.
const RAW_DMABUF_FAILURE_LATCH: u32 = 3;
/// Record an encoder-side raw-dmabuf import failure. Latches the process-wide disable after
/// `RAW_DMABUF_FAILURE_LATCH` consecutive failures.
/// Consecutive capture rebuilds whose dmabuf-only offer never negotiated before the passthrough is
/// latched off. **2 = one retry**, deliberately: each failed negotiation costs a ~10 s stall, so a
/// larger budget is paid by the user in dead air. One retry is enough to survive a compositor
/// caught mid-restart, which is the transient this exists for; a compositor that genuinely never
/// accepts keeps the same capture identity, so its streak accumulates and it latches on the second
/// try — one extra stall versus the old behaviour, once per host lifetime.
const RAW_DMABUF_NEGOTIATION_LATCH: u32 = 2;
/// The raw-dmabuf passthrough's off-switch — **two causes with two different lifetimes**, which is
/// the whole point of this type.
///
/// They used to share one `AtomicBool`, so the cheap recoverable cause (a negotiation that timed
/// out, possibly because the compositor was mid-restart) was as permanent as the expensive
/// unrecoverable one (an encoder that cannot import what this compositor allocates). Once either
/// fired, EVERY later session on the host captured CPU frames until the process was restarted —
/// including sessions against a completely different compositor and node, which had never failed
/// at anything.
///
/// * **Import failures stay sticky.** A driver that will not take what the compositor allocates
/// refuses identically on every retry, and the encode-stall recovery above cannot tell that from
/// a transient — it rebuilt the same failing encoder five times and then ended the session, on
/// every connection, forever. That is what this latch was born to stop, and it must keep
/// stopping it.
/// * **Negotiation timeouts get a retry budget** ([`RAW_DMABUF_NEGOTIATION_LATCH`]).
/// * **Both are keyed to a capture identity.** A new node id — a fresh virtual output, the
/// Bazzite Gaming↔Desktop switch, a compositor restart — is a genuinely different question, so
/// it earns a fresh dmabuf attempt instead of inheriting a verdict about something else.
///
/// Atomics rather than a lock because [`note_import_ok`](Self::note_import_ok) is on the per-frame
/// import path; everything else here runs at pipeline build or on failure.
#[derive(Debug)]
pub struct RawDmabufLatch {
import_streak: AtomicU32,
import_latched: AtomicBool,
negotiation_streak: AtomicU32,
negotiation_latched: AtomicBool,
/// The capture identity the counters above describe. `u64::MAX` = nothing observed yet (a real
/// identity is a node id, so it can never collide with the sentinel).
identity: AtomicU64,
}
/// Nothing observed yet — distinct from any real capture identity.
const NO_IDENTITY: u64 = u64::MAX;
impl RawDmabufLatch {
pub const fn new() -> Self {
RawDmabufLatch {
import_streak: AtomicU32::new(0),
import_latched: AtomicBool::new(false),
negotiation_streak: AtomicU32::new(0),
negotiation_latched: AtomicBool::new(false),
identity: AtomicU64::new(NO_IDENTITY),
}
}
/// Whether the raw-dmabuf passthrough is currently off, for either cause.
pub fn disabled(&self) -> bool {
self.import_latched.load(Ordering::Relaxed)
|| self.negotiation_latched.load(Ordering::Relaxed)
}
/// Tell the latch which capture is about to be built. A DIFFERENT capture from the one the
/// current verdict was formed against clears every counter and both latches, so the new
/// pipeline earns a fresh dmabuf attempt.
///
/// Returns `true` only when that clear actually **re-armed something** — i.e. the identity
/// changed *and* a latch was set. Deliberately not "the identity changed": every session on a
/// fresh virtual output changes it, and a caller that logged on that would print a re-arm line
/// on every healthy session open, which is noise. `true` means "this capture would have been
/// forced to CPU by an earlier capture's verdict, and no longer is".
///
/// Call this BEFORE reading [`disabled`](Self::disabled) for a negotiation decision, or the
/// decision is made against the previous capture's verdict.
pub fn observe_capture(&self, identity: u64) -> bool {
if self.identity.swap(identity, Ordering::Relaxed) == identity {
return false;
}
let was_latched = self.disabled();
self.import_streak.store(0, Ordering::Relaxed);
self.import_latched.store(false, Ordering::Relaxed);
self.negotiation_streak.store(0, Ordering::Relaxed);
self.negotiation_latched.store(false, Ordering::Relaxed);
was_latched
}
/// Record an encoder-side raw-dmabuf import failure. Returns `true` if this failure is the one
/// that latched the passthrough off.
pub fn note_import_failure(&self) -> Option<u32> {
let streak = self.import_streak.fetch_add(1, Ordering::Relaxed) + 1;
(streak >= RAW_DMABUF_FAILURE_LATCH && !self.import_latched.swap(true, Ordering::Relaxed))
.then_some(streak)
}
/// Record a raw dmabuf that imported and encoded — resets the failure streak. The per-frame
/// hot path, hence a single relaxed store.
///
/// Deliberately does NOT clear `import_latched`: once the latch fires, capture has already
/// moved to CPU frames, so there are no more dmabuf imports to succeed. Only a new capture
/// identity clears it.
pub fn note_import_ok(&self) {
self.import_streak.store(0, Ordering::Relaxed);
}
/// Record a capture rebuild whose dmabuf-only offer never negotiated. Returns `Some(streak)`
/// if this is the failure that latched the passthrough off, `None` while retries remain.
pub fn note_negotiation_timeout(&self) -> Option<u32> {
let streak = self.negotiation_streak.fetch_add(1, Ordering::Relaxed) + 1;
(streak >= RAW_DMABUF_NEGOTIATION_LATCH
&& !self.negotiation_latched.swap(true, Ordering::Relaxed))
.then_some(streak)
}
/// Record a capture whose dmabuf offer DID negotiate — the retry budget is per consecutive
/// run of failures, so a success spends none of it.
pub fn note_negotiation_ok(&self) {
self.negotiation_streak.store(0, Ordering::Relaxed);
}
/// Diagnostic for the session-open line: which cause (if any) currently holds it off.
pub fn state(&self) -> &'static str {
match (
self.import_latched.load(Ordering::Relaxed),
self.negotiation_latched.load(Ordering::Relaxed),
) {
(true, true) => "latched: encoder-import + negotiation",
(true, false) => "latched: encoder-import failures",
(false, true) => "latched: negotiation timeouts",
(false, false) => "live",
}
}
}
impl Default for RawDmabufLatch {
fn default() -> Self {
Self::new()
}
}
static RAW_DMABUF: RawDmabufLatch = RawDmabufLatch::new();
/// Record an encoder-side raw-dmabuf import failure. Latches the passthrough off after
/// `RAW_DMABUF_FAILURE_LATCH` consecutive failures, until the capture identity changes.
pub fn note_raw_dmabuf_import_failure(reason: &str) {
let streak = RAW_DMABUF_FAILURE_STREAK.fetch_add(1, Ordering::Relaxed) + 1;
if streak >= RAW_DMABUF_FAILURE_LATCH && !RAW_DMABUF_DISABLED.swap(true, Ordering::Relaxed) {
if let Some(streak) = RAW_DMABUF.note_import_failure() {
tracing::error!(
streak,
reason,
"zero-copy raw-dmabuf passthrough disabled for this host process: the encoder failed \
to import the compositor's dmabuf {streak} times in a row captures fall back to the \
CPU path (slower, but this host could not stream at all otherwise)"
"zero-copy raw-dmabuf passthrough disabled: the encoder failed to import the \
compositor's dmabuf {streak} times in a row captures fall back to the CPU path \
(slower, but this host could not stream at all otherwise). A new capture (different \
node / compositor) clears this."
);
}
}
/// Record a raw dmabuf that imported and encoded — resets the failure streak.
pub fn note_raw_dmabuf_import_ok() {
RAW_DMABUF_FAILURE_STREAK.store(0, Ordering::Relaxed);
RAW_DMABUF.note_import_ok();
}
/// Latch the raw-dmabuf passthrough off because its dmabuf-only *offer never negotiated* — the
/// CAPTURE-side counterpart to [`note_raw_dmabuf_import_failure`]'s encoder-side streak. One
/// timeout is conclusive for this offer (a compositor that cannot allocate the requested
/// LINEAR/modifier BGRx dmabuf refuses it identically on every retry), so there is no streak to
/// count: the next capture skips the passthrough and negotiates SHM/CPU instead of re-running the
/// same 10 s timeout on every reconnect.
/// CAPTURE-side counterpart to [`note_raw_dmabuf_import_failure`]'s encoder-side streak.
///
/// Unlike the import streak this gets a retry budget: the offer can time out because the
/// compositor was mid-restart rather than because it will never accept, and the old behaviour
/// (one timeout = CPU capture for the rest of the host's life, for every compositor and every
/// node) turned a transient into a permanent downgrade nobody could see.
///
/// Scoped deliberately. This used to be `note_vaapi_dmabuf_failed`, which fed [`enabled`] and so
/// disabled ALL zero-copy host-wide — see [`enabled`]. `RAW_DMABUF_DISABLED` gates only the
/// raw-passthrough decision, so the EGL→CUDA importer that a later NVENC session builds is
/// untouched.
/// disabled ALL zero-copy host-wide — see [`enabled`]. It gates only the raw-passthrough decision,
/// so the EGL→CUDA importer that a later NVENC session builds is untouched.
pub fn note_raw_dmabuf_negotiation_failed() {
if !RAW_DMABUF_DISABLED.swap(true, Ordering::Relaxed) {
tracing::warn!(
"zero-copy raw-dmabuf passthrough disabled for this host process: the compositor never \
accepted the dmabuf-only capture offer, so later captures negotiate the CPU path \
instead of repeating that timeout (the EGLCUDA import path is NOT affected)"
);
match RAW_DMABUF.note_negotiation_timeout() {
Some(streak) => tracing::warn!(
streak,
"zero-copy raw-dmabuf passthrough disabled: the compositor did not accept the \
dmabuf-only capture offer {streak} builds in a row, so later captures negotiate the \
CPU path instead of repeating that timeout (the EGLCUDA import path is NOT \
affected). A new capture (different node / compositor) clears this."
),
None => tracing::warn!(
"the compositor did not accept the dmabuf-only capture offer — retrying dmabuf on the \
next capture build before giving up on it"
),
}
}
/// True once repeated encoder import failures latched the raw-dmabuf passthrough off (see
/// [`note_raw_dmabuf_import_failure`]).
/// Record a capture whose dmabuf offer negotiated — spends none of the retry budget.
pub fn note_raw_dmabuf_negotiation_ok() {
RAW_DMABUF.note_negotiation_ok();
}
/// Tell the latch which capture is about to be built, so a verdict formed against a DIFFERENT
/// compositor/node is not inherited. Returns `true` if a latch was cleared by the change.
pub fn note_raw_dmabuf_capture(identity: u64) -> bool {
let cleared = RAW_DMABUF.observe_capture(identity);
if cleared {
tracing::info!(
identity,
"zero-copy raw-dmabuf passthrough re-armed: this is a different capture from the one \
that failed, so it gets a fresh dmabuf attempt"
);
}
cleared
}
/// True while either cause holds the raw-dmabuf passthrough off (see [`RawDmabufLatch`]).
pub fn raw_dmabuf_import_disabled() -> bool {
RAW_DMABUF_DISABLED.load(Ordering::Relaxed)
RAW_DMABUF.disabled()
}
/// Which cause holds the passthrough off, for the session-open diagnostic line.
pub fn raw_dmabuf_latch_state() -> &'static str {
RAW_DMABUF.state()
}
/// The EGL→CUDA twin of the raw-passthrough negotiation latch: the capture advertised the GPU
@@ -564,4 +731,131 @@ mod tests {
note_gpu_import_death(); // third consecutive death
assert!(gpu_import_disabled());
}
// ---- PW3: the raw-dmabuf latch's two lifetimes ------------------------------------------
//
// Against a LOCAL `RawDmabufLatch`, never the process-wide static: these assertions are about
// the state machine, and sharing one global across a test binary's threads is how a latch test
// becomes order-dependent.
/// The expensive cause stays sticky. A driver that cannot import what this compositor
/// allocates refuses identically every time, and the encode-stall recovery cannot tell that
/// from a transient — this latch is what stops it rebuilding the same doomed encoder forever.
#[test]
fn import_failures_latch_and_stay_latched() {
let l = RawDmabufLatch::new();
assert!(!l.disabled());
assert_eq!(l.note_import_failure(), None); // 1
assert_eq!(l.note_import_failure(), None); // 2
assert!(!l.disabled(), "must not latch before the streak completes");
assert_eq!(l.note_import_failure(), Some(3));
assert!(l.disabled());
// Only the FIRST crossing reports, so the error line cannot repeat per frame.
assert_eq!(l.note_import_failure(), None);
// A success resets the streak but must NOT unlatch: once capture moved to CPU frames there
// are no more dmabuf imports, so an "ok" here would be about something else entirely.
l.note_import_ok();
assert!(l.disabled());
}
/// A run of failures broken by a success spends none of the budget — the streak is
/// consecutive-only, which is what makes an occasional failure survivable.
#[test]
fn a_success_breaks_the_import_streak() {
let l = RawDmabufLatch::new();
l.note_import_failure();
l.note_import_failure();
l.note_import_ok();
assert_eq!(l.note_import_failure(), None, "streak restarted at 1");
assert_eq!(l.note_import_failure(), None);
assert!(!l.disabled());
assert_eq!(l.note_import_failure(), Some(3));
}
/// The cheap cause gets a retry. This is the behaviour change PW3 exists for: one timeout used
/// to mean CPU capture for the rest of the host's life, on every compositor and every node.
#[test]
fn a_negotiation_timeout_is_retried_before_it_latches() {
let l = RawDmabufLatch::new();
assert_eq!(l.note_negotiation_timeout(), None, "first one retries");
assert!(
!l.disabled(),
"the next capture build must still be allowed to try dmabuf"
);
assert_eq!(l.note_negotiation_timeout(), Some(2));
assert!(l.disabled());
assert_eq!(l.note_negotiation_timeout(), None, "reports once");
}
/// A capture that negotiates credits the budget back, so a compositor that fails once and then
/// works never accumulates its way to a latch across an evening of reconnects.
#[test]
fn a_negotiated_capture_credits_the_retry_budget() {
let l = RawDmabufLatch::new();
for _ in 0..10 {
assert_eq!(l.note_negotiation_timeout(), None);
l.note_negotiation_ok();
}
assert!(!l.disabled());
}
/// A different capture is a different question. New node id (fresh virtual output, compositor
/// restart, the Bazzite Gaming↔Desktop switch) clears BOTH causes — the same capture does not.
#[test]
fn a_new_capture_identity_clears_the_latch_and_the_same_one_does_not() {
let l = RawDmabufLatch::new();
// Nothing is latched yet, so observing a new capture re-arms NOTHING — that is what the
// return value means, and it is why a healthy session open logs no re-arm line.
assert!(
!l.observe_capture(7),
"nothing was latched, nothing re-armed"
);
assert!(!l.observe_capture(7), "same capture, no clear");
for _ in 0..RAW_DMABUF_FAILURE_LATCH {
l.note_import_failure();
}
assert!(l.disabled());
assert!(
!l.observe_capture(7),
"the SAME capture must keep its verdict — this is the 10s-stall hazard the latch exists for"
);
assert!(l.disabled());
assert!(l.observe_capture(9), "a different node re-arms it");
assert!(!l.disabled());
// ...and the streaks reset with it, so the fresh attempt gets a full budget.
assert_eq!(l.note_import_failure(), None);
}
/// The negotiation latch is keyed the same way — a compositor restart must not inherit the
/// previous one's timeout verdict.
#[test]
fn a_new_capture_identity_clears_the_negotiation_latch_too() {
let l = RawDmabufLatch::new();
l.observe_capture(1);
l.note_negotiation_timeout();
l.note_negotiation_timeout();
assert!(l.disabled());
assert!(l.observe_capture(2));
assert!(!l.disabled());
}
/// The session-open line has to name WHICH cause holds it off — "cpu because nothing here
/// does dmabuf" and "cpu because something failed earlier" are different bugs.
#[test]
fn latch_state_names_the_cause() {
let l = RawDmabufLatch::new();
assert_eq!(l.state(), "live");
l.note_negotiation_timeout();
l.note_negotiation_timeout();
assert_eq!(l.state(), "latched: negotiation timeouts");
let l = RawDmabufLatch::new();
for _ in 0..RAW_DMABUF_FAILURE_LATCH {
l.note_import_failure();
}
assert_eq!(l.state(), "latched: encoder-import failures");
for _ in 0..RAW_DMABUF_NEGOTIATION_LATCH {
l.note_negotiation_timeout();
}
assert_eq!(l.state(), "latched: encoder-import + negotiation");
}
}
@@ -353,6 +353,18 @@ impl FrameChannel {
/// all-intra stream ([`Self::set_all_intra`]) a multi-deep queue drains to the NEWEST AU
/// instead — the skipped ones are already superseded and decode independently, so showing
/// them only adds latency.
///
/// ⚠ **The all-intra drain counts QUEUE ENTRIES and assumes one entry == one AU.** That holds
/// today only because slice-progressive delivery is refused on PyroWave
/// (`client/pump/handshake.rs`; see [`crate::session::Session::set_deliver_frame_parts`]).
/// Turn parts on for an all-intra stream and one AU pushes several entries, at which point
/// `len > 1` no longer means "the consumer is behind": this fires mid-AU, hands back a SUFFIX
/// and `clear()`s that AU's own prefixes — a headerless frame, every frame. Anyone making the
/// two composable must skip whole SUPERSEDED AUs (drop up to the newest entry whose
/// `part.first` is set, never split an AU), give `push`'s `FRAME_QUEUE_HARD_CAP` eviction the
/// same rule, and count `skipped_total` in AUs. Host-side streamed AUs
/// ([`crate::quic::VIDEO_CAP_STREAMED_AU`]) are NOT affected — they still arrive as one
/// completed `Frame` per AU.
pub(crate) fn pop(&self, timeout: Duration) -> FramePop {
let mut st = self.inner.lock().unwrap();
if st.q.is_empty() && !st.closed {
@@ -229,7 +229,10 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result<Handshake
}
// Slice-progressive delivery (the embedder's opt-in): AU prefixes hand up as
// `Frame::part` pieces while the tail is still on the wire. Never on PyroWave — its
// all-intra frame channel drains newest-wins, which assumes whole AUs.
// all-intra frame channel drains newest-wins per QUEUE ENTRY, so parts of one AU read as
// separate AUs and the drain shreds the AU it is mid-way through (`FrameChannel::pop`
// spells out the mechanism and what a fix would take). Unrelated to the host's streamed-AU
// wire (`VIDEO_CAP_STREAMED_AU`), which still completes one whole `Frame` per AU.
if args.frame_parts && welcome.codec != crate::quic::CODEC_PYROWAVE {
session.set_deliver_frame_parts(true);
}
+128 -1
View File
@@ -82,6 +82,49 @@ fn stream_transport_idle(idle: std::time::Duration) -> Arc<quinn::TransportConfi
Arc::new(t)
}
/// Endpoint config for the CLIENT endpoint — the half of the jumbo opt-in that lives on the
/// receiving side, and without which the whole jumbo leg is unreachable.
///
/// `EndpointConfig::max_udp_payload_size` is the QUIC transport parameter this endpoint
/// advertises: "the largest UDP payload I accept". quinn defaults it to **1472** (a 1500-byte
/// Ethernet MTU), and a peer's MTU-discovery search is upper-bounded by
/// `min(MtuDiscoveryConfig::upper_bound, the value the OTHER side advertised)`
/// (`quinn_proto::connection::mtud::SearchState::new`). So raising the host's probe ceiling
/// alone — which is all [`stream_transport_idle`] did — can never make a host's discovery
/// settle above 1472: the *client's* default advertisement caps it, and the host's
/// settled-at-jumbo proof (`native/wire_mtu.rs`, both the mid-session grow and the
/// session-start one) could never fire. This raises the advertisement to the sealed jumbo
/// datagram size so the proof is obtainable at all.
///
/// Gated on the SAME operator opt-in as the probe ceiling ([`crate::config::jumbo_wire_mtu`],
/// i.e. `PUNKTFUNK_JUMBO=1` / `PUNKTFUNK_WIRE_MTU` > 1500) because it is not free: quinn sizes
/// its endpoint receive buffer as `max_udp_payload_size × max_receive_segments × BATCH_SIZE`,
/// which on a GRO-capable Linux/Android client is 64 × 32 segments — ~2.9 MiB at the 1472
/// default, ~18 MiB at jumbo. A jumbo LAN is a deliberate deployment; every other client keeps
/// today's buffer to the byte. Without the opt-in this returns the stock config, so the
/// advertisement, the wire, and the memory are all unchanged.
fn endpoint_config() -> quinn::EndpointConfig {
let mut cfg = quinn::EndpointConfig::default();
if let Some(mtu) = crate::config::jumbo_wire_mtu() {
// Derived exactly like the probe ceiling above (IPv4 overhead — a v6 peer's sealed
// target is smaller, so this covers it), and clamped into quinn's accepted range.
let shard = crate::config::jumbo_shard_payload_for(
mtu,
std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED),
);
let accept = crate::config::sealed_datagram_bytes(shard).clamp(1200, 65_527) as u16;
if cfg.max_udp_payload_size(accept).is_ok() {
tracing::info!(
max_udp_payload_size = accept,
wire_mtu = mtu,
"jumbo opt-in: this endpoint advertises a jumbo QUIC receive ceiling, so the \
peer's MTU discovery can prove a jumbo path (it is capped by this value)"
);
}
}
cfg
}
/// Server endpoint with a fresh self-signed certificate (tests/dev — production hosts
/// persist an identity and use [`server_with_identity`] so clients can pin it).
pub fn server(addr: std::net::SocketAddr) -> anyhow_result::Result<quinn::Endpoint> {
@@ -238,7 +281,15 @@ pub fn client_pinned_with_identity(
.map_err(|e| anyhow_result::Error::msg(format!("quic client config: {e}")))?;
let mut client_cfg = quinn::ClientConfig::new(Arc::new(quic_cfg));
client_cfg.transport_config(stream_transport()); // keep-alive — see stream_transport
let mut ep = quinn::Endpoint::client("0.0.0.0:0".parse().unwrap())?;
// `Endpoint::client` hardcodes `EndpointConfig::default()`, whose 1472-byte
// `max_udp_payload_size` caps the HOST's MTU discovery (see `endpoint_config`), so the
// endpoint is built by hand to carry the jumbo opt-in. Same bind as before
// (`0.0.0.0:0`, v4 — no dual-stack flag to reproduce) and the same default runtime.
let socket = std::net::UdpSocket::bind("0.0.0.0:0")?;
let runtime = quinn::default_runtime()
.ok_or_else(|| anyhow_result::Error::msg("no async runtime found".into()))?;
let mut ep = quinn::Endpoint::new(endpoint_config(), None, socket, runtime)?;
ep.set_default_client_config(client_cfg);
Ok(ep)
})();
@@ -348,4 +399,80 @@ mod tests {
let _ = super::stream_transport_idle(std::time::Duration::MAX);
let _ = super::stream_transport_idle(std::time::Duration::ZERO);
}
/// Where a connection's MTU discovery is allowed to climb to, measured rather than argued
/// (PW7a). Loopback's own MTU is 64 KiB, so the ONLY thing that can stop the search here is
/// configuration — which makes this a clean instrument for the two ceilings:
///
/// * **leg A** — server opted in, client NOT: the search stalls at the client's default
/// `max_udp_payload_size` advertisement (1472) no matter how high the server's probe
/// ceiling is. This is why the shipped jumbo grow could never fire: `wire_mtu.rs` waits
/// for a settle at the sealed jumbo size and the peer's transport parameter forbids it.
/// * **leg B** — both opted in: the search reaches the sealed jumbo datagram, and the
/// elapsed time is what the `Welcome`'s bounded proof-wait has to cover.
///
/// `#[ignore]`d: it sets process-wide env (each endpoint reads the opt-in at construction,
/// which is exactly how the two legs are built) and spends seconds of wall clock.
/// Run it alone: `cargo test -p punktfunk-core --features quic mtu_discovery -- --ignored
/// --nocapture --test-threads=1`.
#[tokio::test]
#[ignore = "measurement: sets process env and takes ~15 s of wall clock"]
async fn mtu_discovery_climbs_only_as_high_as_the_peer_advertises() {
async fn climb(server_jumbo: bool, client_jumbo: bool) -> (u16, u128) {
let set = |on: bool| {
if on {
std::env::set_var("PUNKTFUNK_JUMBO", "1");
} else {
std::env::remove_var("PUNKTFUNK_JUMBO");
}
};
set(server_jumbo);
let server = endpoint::server("127.0.0.1:0".parse().unwrap()).unwrap();
let addr = server.local_addr().unwrap();
set(client_jumbo);
let client = endpoint::client_insecure().unwrap();
set(false);
let accept = tokio::spawn(async move {
let incoming = server.accept().await.expect("incoming");
let conn = incoming.await.expect("host side connects");
(server, conn)
});
let client_conn = client.connect(addr, "punktfunk").unwrap().await.unwrap();
let (_server_ep, host_conn) = accept.await.unwrap();
// A stream write gives the driver something to transmit, which is what starts the
// search (probes ride `poll_transmit`); after that each probe's ack drives the next.
let mut s = host_conn.open_uni().await.unwrap();
s.write_all(b"go").await.unwrap();
let want = crate::config::sealed_datagram_bytes(crate::config::jumbo_shard_payload_for(
9000,
std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED),
)) as u16;
let t0 = std::time::Instant::now();
let mut mtu = host_conn.stats().path.current_mtu;
while t0.elapsed() < std::time::Duration::from_secs(6) && mtu < want {
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
mtu = host_conn.stats().path.current_mtu;
}
let elapsed = t0.elapsed().as_millis();
drop(client_conn);
drop(client);
(mtu, elapsed)
}
let (capped, _) = climb(true, false).await;
println!("leg A (server opted in, client not): settled at {capped} B UDP payload");
assert_eq!(
capped, 1472,
"a peer that advertises the stock max_udp_payload_size caps the search at 1472 — \
the whole point of raising it on the client endpoint"
);
let (grown, ms) = climb(true, true).await;
println!("leg B (both opted in): reached {grown} B UDP payload in {ms} ms");
assert!(
grown >= 8972,
"both sides opted in, loopback MTU is 64 KiB — discovery should reach the sealed \
jumbo datagram, got {grown}"
);
}
}
+15 -2
View File
@@ -677,8 +677,21 @@ impl Session {
/// [`Frame::part`]` = Some` while the rest is still on the wire, instead of one whole-AU
/// delivery (the slice-progressive decode path — [`crate::packet::USER_FLAG_SLICE_STREAM`]).
/// With it on, EVERY video frame delivery carries `part: Some` (a frame with no early
/// parts arrives as the degenerate `{offset: 0, first, last}` whole). Do not combine with
/// an all-intra (PyroWave) stream: its newest-wins draining assumes whole AUs.
/// parts arrives as the degenerate `{offset: 0, first, last}` whole).
///
/// **Do not combine with an all-intra (PyroWave) stream**, and the reason is sharper than
/// "newest-wins draining assumes whole AUs" (2026-08-08, PW6): the drain
/// (`client::frame_channel::FrameChannel::pop`) counts QUEUE ENTRIES and takes one entry to be
/// one AU. With parts on, a single AU pushes K entries, so `len > 1` stops meaning "the consumer
/// is behind" — the drain fires mid-AU, returns the newest entry (a SUFFIX) and clears that
/// same AU's prefixes. For PyroWave that is unrecoverable rather than lossy: the sequence
/// header lives in window 0 of every AU, so every frame would arrive headerless. Making the
/// two composable means teaching the drain to skip whole superseded AUs (never to split one)
/// — see the PW6 section of `design/linux-host-performance-wave2-pyrowave.md`.
///
/// Note this is a DIFFERENT axis from the host's streamed-AU wire
/// ([`crate::quic::VIDEO_CAP_STREAMED_AU`]): a streamed AU still completes as ONE `Frame`
/// here, so it is unaffected by any of the above.
pub fn set_deliver_frame_parts(&mut self, on: bool) {
self.reassembler.set_deliver_parts(on);
}
+23 -2
View File
@@ -844,6 +844,7 @@ fn parse_spike(args: &[String]) -> Result<Options> {
let mut bitrate_mbps = 20u64;
let mut out: Option<PathBuf> = None;
let mut loopback = true;
let mut wire_chunk: Option<usize> = None;
let mut i = 0;
while i < args.len() {
@@ -890,7 +891,13 @@ fn parse_spike(args: &[String]) -> Result<Options> {
"h264" => Codec::H264,
"h265" | "hevc" => Codec::H265,
"av1" => Codec::Av1,
other => bail!("unknown --codec '{other}' (h264|h265|av1)"),
// The spike is the only way to drive a PyroWave capture→encode pass without
// a client, which is what the Linux-host PyroWave work measures against.
// Needs the `pyrowave` feature (default-on) and pairs with
// `PUNKTFUNK_ENCODER=pyrowave`, which is what puts the CAPTURE side on the
// raw-dmabuf passthrough.
"pyrowave" => Codec::PyroWave,
other => bail!("unknown --codec '{other}' (h264|h265|av1|pyrowave)"),
}
}
"--bitrate" => {
@@ -900,6 +907,12 @@ fn parse_spike(args: &[String]) -> Result<Options> {
}
"--out" => out = Some(PathBuf::from(next()?)),
"--no-loopback" => loopback = false,
"--wire-chunk" => {
let v: usize = next()?
.parse()
.map_err(|_| anyhow::anyhow!("bad --wire-chunk (bytes)"))?;
wire_chunk = (v > 0).then_some(v);
}
"-h" | "--help" => {
print_usage();
std::process::exit(0);
@@ -934,6 +947,7 @@ fn parse_spike(args: &[String]) -> Result<Options> {
bitrate_bps: bitrate_mbps.saturating_mul(1_000_000),
out,
loopback,
wire_chunk,
})
}
@@ -1007,11 +1021,18 @@ SPIKE OPTIONS:
KWin virtual output at --width x --height and captures it
--seconds <N> capture duration in seconds (default: 5)
--fps <N> target frame rate (default: 60)
--codec <h264|h265|av1> NVENC codec (default: h265)
--codec <h264|h265|av1|pyrowave>
encode codec (default: h265). 'pyrowave' also wants
PUNKTFUNK_ENCODER=pyrowave so capture takes the passthrough
--bitrate <MBPS> target bitrate in Mbps (default: 20)
--width <W> --height <H> synthetic source size (default: 1920x1080)
--out <PATH> raw Annex-B output (default: /tmp/punktfunk-spike.<ext>)
--no-loopback skip the punktfunk_core round-trip verification
--wire-chunk <BYTES> PyroWave datagram-aligned packetization at this shard payload
(a real session passes its negotiated shard_payload, e.g. 1408).
With PUNKTFUNK_PYROWAVE_STREAMED_AU=1 also armed, the AU is
drained through poll_chunk and sealed as a STREAMED wire frame
(VIDEO_CAP_STREAMED_AU), then byte-verified by the loopback
-h, --help this help
NOTES:
+6 -1
View File
@@ -1148,7 +1148,12 @@ async fn serve_session(
// path verdict (WARN + learned clamp for the next session on a constrained path; clears
// a stale clamp on a healthy one) — and, with the driver above, heal or grow THIS
// session mid-stream. Bounded ~10 s task unless a jumbo grow leaves it as revert guard.
wire_mtu::spawn_watch(conn.clone(), welcome.shard_payload as usize, shard_reneg);
wire_mtu::spawn_watch(
conn.clone(),
welcome.shard_payload as usize,
hello.max_shard_payload,
shard_reneg,
);
// Negotiated cursor forwarding: the HOST_CAP_CURSOR bit the Welcome advertised, read back
// rather than recomputed (`handshake::cursor_forward` computed it once, with the encoder
// blend-capability gate — re-running it here could drift, and would re-probe).
+14 -9
View File
@@ -148,7 +148,6 @@ pub(super) async fn negotiate(
Option<crate::vdisplay::GamescopeRoute>,
Option<super::stream::PrepHandle>,
)> {
let peer = conn.remote_address();
let mut hello = Hello::decode(first).map_err(|e| anyhow!("Hello decode: {e:?}"))?;
if hello.abi_version != punktfunk_core::WIRE_VERSION {
close_rejected(
@@ -497,6 +496,11 @@ pub(super) async fn negotiate(
let (data_sock, direct) = bind_data_socket(data_port)?;
let udp_port = data_sock.local_addr()?.port();
// The session's video geometry (see the `shard_payload` field below). Resolved before the
// Welcome struct because a path a previous session proved jumbo is given a bounded moment
// to re-prove itself live on THIS connection — the awaited part of `negotiated_shard_payload`.
let shard_payload = wire_mtu::negotiated_shard_payload(conn, hello.max_shard_payload).await;
let mut key = [0u8; 16];
rand::thread_rng().fill_bytes(&mut key);
// Fresh per-session salt alongside the fresh key. GCM nonce uniqueness only *requires* one
@@ -548,14 +552,15 @@ pub(super) async fn negotiate(
// hardcoded 1452 overshot the v4 ceiling (its math forgot the header/crypto ride
// inside the UDP payload) and silently IP-fragmented EVERY video datagram, doubling
// per-datagram loss on Wi-Fi — the "100 Mbps badly fails on the phone" root cause.
// Negotiated, so the client follows. Jumbo (≈8900) is a future negotiated bump (needs
// MAX_DATAGRAM_BYTES raised + end-to-end 9000 MTU).
// Resolution order (wire_mtu.rs): `PUNKTFUNK_WIRE_MTU` operator override, then a path
// budget learned from a prior session whose QUIC MTU discovery settled below the
// video-datagram ceiling (the "VPN on the host blackholes every video packet" field
// shape — small flows pass, the stream is an endless black screen), then this family
// default. Healthy paths take the default branch and are byte-identical to before.
shard_payload: wire_mtu::negotiated_shard_payload(peer.ip()) as u16,
// Negotiated, so the client follows.
// Resolution order (wire_mtu.rs): a JUMBO start (≈8900) on a path a previous session
// proved AND this connection has just re-proved live, then the `PUNKTFUNK_WIRE_MTU`
// operator override, then a path budget learned from a prior session whose QUIC MTU
// discovery settled below the video-datagram ceiling (the "VPN on the host blackholes
// every video packet" field shape — small flows pass, the stream is an endless black
// screen), then this family default. Healthy paths take the default branch and are
// byte-identical to before.
shard_payload: shard_payload as u16,
encrypt: true,
key,
salt,
+501 -23
View File
@@ -24,6 +24,14 @@
//! - **Heal** — the next handshake from that peer clamps `shard_payload` to the recorded
//! budget, so a reconnect fixes the stream. A later session that reaches the ceiling erases
//! the record (the learn/heal loop is self-correcting in both directions).
//! - **Grow** (PW7a) — the mirror image, for the jumbo half: a connection whose discovery
//! settles at the sealed JUMBO size has proven the path carries ~8.9 KB video datagrams, and
//! the next session on that same path *starts* there instead of at the 1500-byte default.
//! PyroWave sessions cannot be re-keyed mid-stream (the client's parse window is the
//! `Welcome` value, read once over the C ABI), so the session-start value is the ONLY way
//! they ever reach jumbo — and it is exactly where ~6× fewer datagrams per frame is worth
//! the most. See [`jumbo_session_start`] for why a remembered verdict alone is never
//! allowed to seal one byte above the default.
use std::collections::HashMap;
use std::net::IpAddr;
@@ -63,10 +71,155 @@ fn learned() -> &'static Mutex<HashMap<IpAddr, u16>> {
LEARNED.get_or_init(|| Mutex::new(HashMap::new()))
}
/// The shard payload for a new session to `peer`: `PUNKTFUNK_WIRE_MTU` override, else the
/// peer's learned path budget, else the family default (today's exact behavior). Logs whenever
/// the result differs from the default.
pub(super) fn negotiated_shard_payload(peer: IpAddr) -> usize {
/// Identity of a PATH, not of a peer — the key the jumbo verdict is filed under.
///
/// The clamp above is keyed by peer IP alone, and that is safe *because being wrong is benign*:
/// a stale clamp only makes video datagrams smaller than they had to be. A stale GROW is the
/// opposite — one oversized datagram on a 1500-byte path is silently dropped, which is the
/// "connects fine, black screen forever" shape this whole module exists to kill. So the grow
/// keys strictly: a verdict earned over the host's 10 GbE NIC does not apply to the same peer
/// IP reached over the host's Wi-Fi or a VPN adapter, because those are different routes with
/// different MTUs.
///
/// `local` is `Connection::local_ip()` (the address the connection was actually received on);
/// `None` where the platform can't report it, which degrades this key to the clamp's — safely,
/// because the live re-proof in [`jumbo_session_start`] is what actually protects the grow.
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
struct PathKey {
local: Option<IpAddr>,
peer: IpAddr,
}
/// A path that a completed MTU-discovery search proved carries jumbo video datagrams.
#[derive(Clone, Copy, Debug)]
struct JumboVerdict {
/// The settled UDP-payload budget the proof measured.
udp_budget: u16,
/// The operator's jumbo target when the proof was taken. A changed `PUNKTFUNK_JUMBO` /
/// `PUNKTFUNK_WIRE_MTU` invalidates it rather than being silently reinterpreted.
target_wire_mtu: usize,
/// When it was taken ([`JUMBO_VERDICT_TTL`]).
at: std::time::Instant,
}
/// How long a jumbo verdict may be redeemed for. Contrary evidence erases it long before this
/// (any settle below the sealed target, on any later session over the same path — the same
/// self-correction the clamp has), so the TTL is not the safety mechanism; it is a bound on how
/// stale an *unrefreshed* memory can get, for the case where the path changes while no session
/// is running.
const JUMBO_VERDICT_TTL: std::time::Duration = std::time::Duration::from_secs(6 * 3600);
/// How long the `Welcome` may wait for THIS connection's MTU discovery to re-prove a jumbo
/// path.
///
/// The wait is structural, not laziness: every connection restarts discovery from ~1200 bytes,
/// so the live proof the grow requires does not exist yet when the `Welcome` is built — and the
/// binary search up to sealed-jumbo needs an ACKED probe per step, each of which a peer may sit
/// on for its ack delay. Without a wait the gate would never pass and the feature would be dead.
///
/// It is honestly on the bring-up critical path (`handshake.rs` sends the `Welcome` and only
/// THEN kicks the display prep), so it is bounded, returns the instant the proof lands, and is
/// entered ONLY for a path a previous session already proved jumbo — i.e. an opted-in operator
/// on a jumbo LAN, never anyone else. The worst case (the full wait, no proof) is the moved
/// laptop, and it is self-limiting: that session's watcher erases the verdict, so the next
/// connect doesn't wait at all.
const JUMBO_PROOF_WAIT: std::time::Duration = std::time::Duration::from_millis(300);
const JUMBO_PROOF_POLL: std::time::Duration = std::time::Duration::from_millis(10);
/// Proven-jumbo paths. Same lifetime rules as [`learned`] — in-memory, re-earned in one session
/// after a host restart.
fn jumbo_verdicts() -> &'static Mutex<HashMap<PathKey, JumboVerdict>> {
static JUMBO: OnceLock<Mutex<HashMap<PathKey, JumboVerdict>>> = OnceLock::new();
JUMBO.get_or_init(|| Mutex::new(HashMap::new()))
}
fn path_key(conn: &quinn::Connection) -> PathKey {
PathKey {
local: conn.local_ip(),
peer: conn.remote_address().ip(),
}
}
/// Everything the session-start jumbo decision reads. Every field but `proven_udp_budget` is
/// observed on THIS connection during THIS handshake — which is the point (see
/// [`jumbo_session_start`]).
#[derive(Clone, Copy, Debug)]
struct JumboStart {
/// The host operator's opt-in ([`jumbo_wire_mtu`]) — `None` = no jumbo, ever.
target_wire_mtu: Option<usize>,
/// `Hello::max_shard_payload`: the client's own receive ceiling (0 = legacy client, which
/// never gets a geometry it didn't ask for).
client_ceiling: u16,
/// `conn.stats().path.current_mtu` right now: the largest UDP payload quinn has had ACKED
/// on this connection.
live_udp_mtu: u16,
/// What a previous session over this same [`PathKey`] settled at, if any.
proven_udp_budget: Option<u16>,
/// The constrained-path clamp [`learned`] for this peer, if any. Contradictory evidence
/// (this peer black-screened on a small MTU recently) vetoes the grow — the two memories
/// are keyed differently and the safe one wins.
clamped_udp_budget: Option<u16>,
}
/// The jumbo shard payload a session to `peer` could use, or `None` when there is nothing to
/// gain (no opt-in, a legacy/low client ceiling, or a target that isn't bigger than the family
/// default). Shared by the decision, the wait, and the watcher so all three agree on the number.
fn jumbo_target(
target_wire_mtu: Option<usize>,
client_ceiling: u16,
peer: IpAddr,
) -> Option<usize> {
let mtu = target_wire_mtu?;
let t = jumbo_shard_payload_for(mtu, peer).min(client_ceiling as usize);
let t = t - t % 2; // FEC requires even shards
(t > mtu1500_shard_payload_for(peer)).then_some(t)
}
/// The session-START jumbo decision: `Some(shard_payload)` only when every gate below holds.
///
/// **Why a remembered verdict is never enough.** A laptop that proved jumbo on the wired LAN
/// and comes back on Wi-Fi, a switch that lost its jumbo config, a client IP recycled by DHCP —
/// all of them present a path that cannot carry an 8.9 KB datagram, and a PyroWave session
/// sealed at that size cannot be re-keyed mid-stream, so it would black-screen for its whole
/// life. The memory therefore only decides whether it is worth WAITING for a proof; what
/// actually authorises the grow is `live_udp_mtu` — a datagram of exactly that size, acked by
/// this client, on this connection, seconds ago. That is why this is as safe as the clamp
/// despite the failure modes being opposite: a wrong memory cannot produce a jumbo `Welcome`,
/// only a live measurement can.
///
/// The gates, in order: the host operator opted in; the client advertised enough receive
/// headroom; the target beats the family default (nothing to gain otherwise); no constrained-path
/// clamp contradicts it; a prior session over this exact path settled at or above the sealed
/// target; and this connection has re-proven it live.
fn jumbo_session_start(i: JumboStart, peer: IpAddr) -> Option<usize> {
let target = jumbo_target(i.target_wire_mtu, i.client_ceiling, peer)?;
let sealed = sealed_datagram_bytes(target);
if let Some(clamp) = i.clamped_udp_budget {
if (clamp as usize) < sealed {
return None;
}
}
if (i.proven_udp_budget? as usize) < sealed {
return None;
}
if (i.live_udp_mtu as usize) < sealed {
return None;
}
Some(target)
}
/// The shard payload for a new session on `conn`: a proven-jumbo grow, else the
/// `PUNKTFUNK_WIRE_MTU` override, else the peer's learned path budget, else the family default
/// (today's exact behavior). Logs whenever the result differs from the default.
///
/// `client_ceiling` is the client's `Hello::max_shard_payload`. Async only for the bounded
/// [`JUMBO_PROOF_WAIT`], which is entered *only* on a path a previous session already proved
/// jumbo — every other session resolves without awaiting anything.
pub(super) async fn negotiated_shard_payload(
conn: &quinn::Connection,
client_ceiling: u16,
) -> usize {
let peer = conn.remote_address().ip();
let env = match std::env::var("PUNKTFUNK_WIRE_MTU") {
Ok(v) => match v.trim().parse::<usize>() {
Ok(mtu) => Some(mtu),
@@ -78,13 +231,80 @@ pub(super) fn negotiated_shard_payload(peer: IpAddr) -> usize {
Err(_) => None,
};
let learned_budget = learned().lock().unwrap().get(&peer).copied();
resolve(env, learned_budget, peer)
let target_wire_mtu = jumbo_wire_mtu();
let proven_udp_budget = fresh_verdict(path_key(conn), target_wire_mtu);
let mut jumbo = JumboStart {
target_wire_mtu,
client_ceiling,
live_udp_mtu: conn.stats().path.current_mtu,
proven_udp_budget,
clamped_udp_budget: learned_budget,
};
// A proven path is worth waiting a moment for: MTU discovery starts when the handshake
// completes and needs an acked probe per binary-search step, so at `Welcome` time it may
// simply not have got there yet. Bounded, and only on paths that already proved it once.
let awaited_proof = proven_udp_budget
.and_then(|_| jumbo_target(target_wire_mtu, client_ceiling, peer))
.map(|t| sealed_datagram_bytes(t) as u16);
if let Some(sealed) = awaited_proof {
if jumbo.live_udp_mtu < sealed {
let t0 = std::time::Instant::now();
while t0.elapsed() < JUMBO_PROOF_WAIT {
tokio::time::sleep(JUMBO_PROOF_POLL).await;
jumbo.live_udp_mtu = conn.stats().path.current_mtu;
if jumbo.live_udp_mtu >= sealed {
break;
}
}
tracing::debug!(
peer = %peer,
waited_ms = t0.elapsed().as_millis() as u64,
live_udp_mtu = jumbo.live_udp_mtu,
needed = sealed,
"wire MTU: waited for this connection to re-prove its jumbo path"
);
}
}
resolve(env, learned_budget, jumbo, peer)
}
/// Pure resolution (env override > learned budget > family default) — the tested core of
/// [`negotiated_shard_payload`].
fn resolve(env_wire_mtu: Option<usize>, learned_udp_budget: Option<u16>, peer: IpAddr) -> usize {
/// The peer's jumbo verdict if it is still redeemable: same operator target, inside the TTL.
/// A verdict that fails either test is dropped on the spot rather than left to rot.
fn fresh_verdict(key: PathKey, target_wire_mtu: Option<usize>) -> Option<u16> {
let target = target_wire_mtu?;
let mut map = jumbo_verdicts().lock().unwrap();
let v = *map.get(&key)?;
if v.target_wire_mtu != target || v.at.elapsed() > JUMBO_VERDICT_TTL {
map.remove(&key);
return None;
}
Some(v.udp_budget)
}
/// Pure resolution (proven jumbo > env override > learned budget > family default) — the tested
/// core of [`negotiated_shard_payload`].
fn resolve(
env_wire_mtu: Option<usize>,
learned_udp_budget: Option<u16>,
jumbo: JumboStart,
peer: IpAddr,
) -> usize {
let default = mtu1500_shard_payload_for(peer);
// First, because the two are mutually exclusive by construction: `jumbo_wire_mtu()` only
// fires above 1500, and the env branch below CLAMPS to the family default, so a
// `PUNKTFUNK_WIRE_MTU=9000` operator would otherwise get 1408 and never a jumbo start.
if let Some(p) = jumbo_session_start(jumbo, peer) {
tracing::info!(
peer = %peer,
shard_payload = p,
default,
live_udp_mtu = jumbo.live_udp_mtu,
proven_udp_budget = jumbo.proven_udp_budget,
"wire MTU: session starts at the JUMBO shard — this path proved it in a previous \
session AND re-proved it live on this connection (~6× fewer datagrams per frame)"
);
return p;
}
if let Some(mtu) = env_wire_mtu {
let p = shard_payload_for_wire_mtu(mtu, peer);
if p != default {
@@ -119,34 +339,73 @@ fn resolve(env_wire_mtu: Option<usize>, learned_udp_budget: Option<u16>, peer: I
/// into a verdict — and, with a [`ShardReneg`] driver, act on it MID-SESSION
/// (design/shard-payload-reneg.md Phase 2): a below-ceiling verdict shrinks the live wire at
/// the ~310 s mark (session 1 heals instead of staying black), and a settled-at-jumbo
/// verdict grows it, ack-gated, when the operator opted in. Spawned once per negotiated
/// session; without a grow the task ends after the final sample (bounded ~10 s lifetime,
/// holding only a cheap `Connection` handle) — after a grow it stays as the revert guard
/// until the connection closes.
/// verdict grows it, ack-gated, when the operator opted in. The same settled-at-jumbo reading
/// also writes this path's next-session verdict (PW7a) — `client_ceiling` is the client's
/// `Hello::max_shard_payload`, which decides what "jumbo" is worth proving for this peer.
/// Spawned once per negotiated session; without a grow the task ends after the final sample
/// (bounded ~10 s lifetime, holding only a cheap `Connection` handle) — after a grow, or on a
/// session that STARTED jumbo, it stays as the revert guard until the connection closes.
pub(super) fn spawn_watch(
conn: quinn::Connection,
session_shard_payload: usize,
client_ceiling: u16,
reneg: Option<ShardReneg>,
) {
tokio::spawn(async move {
let peer = conn.remote_address().ip();
let ceiling = video_datagram_udp_ceiling() as u16;
// The sealed size a JUMBO proof has to reach on this path (PW7a) — `None` unless the
// operator opted in AND this client advertised the headroom. Read once: the verdict
// records the target it was proven under, and the two must be the same number.
let target_wire_mtu = jumbo_wire_mtu();
let jumbo_proof =
jumbo_target(target_wire_mtu, client_ceiling, peer).map(sealed_datagram_bytes);
// Discovery finishes in a handful of RTTs on a LAN (well under the first sample) but
// needs a loss timeout per failed probe on a constrained path — the second sample
// covers that with margin. Max, because discovery only ever raises `current_mtu`
// (the post-grow revert guard below re-reads it live, where blackhole detection CAN
// lower it again).
// lower it again). Stop early only once nothing more is expected: with a jumbo opt-in
// the search keeps climbing past the 1500-byte ceiling, and stopping there would throw
// away the very measurement the proof needs.
let goal = jumbo_proof
.unwrap_or(ceiling as usize)
.max(ceiling as usize) as u16;
let mut settled = 0u16;
for wait_s in [3u64, 7] {
tokio::time::sleep(std::time::Duration::from_secs(wait_s)).await;
settled = settled.max(conn.stats().path.current_mtu);
if settled >= ceiling {
if settled >= goal {
break;
}
}
// The wire this session is CURRENTLY sealed at — moves on a mid-session shrink/grow.
let mut current = session_shard_payload;
let mut reneg = reneg;
// PW7a bookkeeping, before anything else can return: this is where a jumbo path earns
// its next-session verdict — and, far more importantly, where it LOSES it. Recording
// needs a live connection that reached the sealed target; anything else (a lower
// settle, a connection that died before the window closed, i.e. exactly what a client
// staring at a black screen does) erases, so the next session falls back to the
// 1500-byte default and has to prove itself again from scratch.
if let Some(need) = jumbo_proof {
let key = path_key(&conn);
if settled as usize >= need && conn.close_reason().is_none() {
jumbo_verdicts().lock().unwrap().insert(
key,
JumboVerdict {
udp_budget: settled,
target_wire_mtu: target_wire_mtu.unwrap_or_default(),
at: std::time::Instant::now(),
},
);
tracing::info!(peer = %peer, discovered_udp_mtu = settled, needed = need,
"wire MTU: this path carries JUMBO video datagrams — the next session over \
it starts at the big shard (it still has to re-prove the path live)");
} else if jumbo_verdicts().lock().unwrap().remove(&key).is_some() {
tracing::info!(peer = %peer, discovered_udp_mtu = settled, needed = need,
"wire MTU: jumbo verdict cleared — this path no longer proves it");
}
}
if settled >= ceiling {
// The path carries full-size video datagrams — erase any stale learned clamp so
// the next session returns to the default wire.
@@ -154,6 +413,34 @@ pub(super) fn spawn_watch(
tracing::info!(peer = %peer,
"wire MTU: path re-measured at full size — learned clamp cleared");
}
// …but "full size" is the 1500-byte ceiling, and this session may have STARTED
// above it (a PW7a jumbo start whose path changed since the proof, or a client
// that roamed onto a 1500-MTU link). Then every video datagram is dying right now.
// The verdict is already erased above; heal the live wire if this session can be
// re-keyed at all — a PyroWave client cannot (its parse window is the `Welcome`
// value), so for those the WARN plus a corrected next session is all there is.
if sealed_datagram_bytes(current) > settled as usize {
tracing::warn!(
peer = %peer,
discovered_udp_mtu = settled,
shard_payload = current,
"wire MTU: this session started at a JUMBO shard but the path does not \
carry it video datagrams are oversized for a hop, which streams as a \
black screen with zero reported loss. The jumbo verdict for this path is \
cleared: the next connect starts at the standard 1500-byte wire."
);
if let Some(r) = reneg.as_ref() {
let back = shard_payload_for_udp_budget(settled as usize, peer);
if back < current
&& r.change_tx.send(back as u16).is_ok()
&& r.apply_tx.send(back).is_ok()
{
tracing::info!(peer = %peer, shard_payload = back, was = current,
"wire MTU: video re-keyed mid-session back to the standard wire");
current = back;
}
}
}
} else {
// A closed connection stops discovering, so a session that ended before the final
// sample proves nothing (a healthy high-RTT path could still be mid-search): learn
@@ -203,12 +490,41 @@ pub(super) fn spawn_watch(
}
}
}
// PW7a revert guard for a session that STARTED jumbo and has no re-key channel (the
// PyroWave case, and the only reason the session-start grow exists). Nothing can save
// this session if the path stops fitting mid-stream — but the NEXT one must not repeat
// it, so keep sampling and drop the verdict the moment quinn's blackhole detection or
// a re-search says the path shrank. Cheap: one `Connection` handle, one sample per 5 s.
// Only for a session that is currently FITTING — one that already failed the check
// above has been warned about and had its verdict erased there.
if current > mtu1500_shard_payload_for(peer)
&& reneg.is_none()
&& sealed_datagram_bytes(current) <= settled as usize
{
loop {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
if conn.close_reason().is_some() {
return;
}
let mtu_now = conn.stats().path.current_mtu;
if (mtu_now as usize) < sealed_datagram_bytes(current) {
jumbo_verdicts().lock().unwrap().remove(&path_key(&conn));
tracing::warn!(peer = %peer, discovered_udp_mtu = mtu_now,
shard_payload = current,
"wire MTU: the jumbo path this session started on stopped fitting — this \
session cannot be re-keyed (chunk-aligned client parse window), so it \
will not recover, but the verdict is cleared and the next connect \
starts at the standard wire");
return;
}
}
}
// Phase 2 up-leg: jumbo grow — operator opt-in (PUNKTFUNK_JUMBO / PUNKTFUNK_WIRE_MTU
// > 1500, which also raised the endpoint's probe ceiling so `settled` can even reach
// here), client-advertised headroom, and a settled-at-jumbo proof. The grow is
// ACK-GATED: not one sealed datagram above the old size leaves before the client's
// ack, even though its buffers are statically sized — the rule must not erode.
let (Some(mtu), Some(r)) = (jumbo_wire_mtu(), reneg.as_mut()) else {
let (Some(mtu), Some(r)) = (target_wire_mtu, reneg.as_mut()) else {
return;
};
let target = jumbo_shard_payload_for(mtu, peer).min(r.client_ceiling as usize);
@@ -275,34 +591,196 @@ mod tests {
const V4: IpAddr = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2));
const V6: IpAddr = IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1));
/// No jumbo anywhere — what every session that isn't on an opted-in jumbo LAN passes.
const NO_JUMBO: JumboStart = JumboStart {
target_wire_mtu: None,
client_ceiling: 0,
live_udp_mtu: 0,
proven_udp_budget: None,
clamped_udp_budget: None,
};
/// A 9000-MTU LAN, a modern client, a path proven last session and re-proven live now.
fn proven_jumbo() -> JumboStart {
JumboStart {
target_wire_mtu: Some(9000),
client_ceiling: punktfunk_core::config::max_shard_payload() as u16,
live_udp_mtu: 8972,
proven_udp_budget: Some(8972),
clamped_udp_budget: None,
}
}
#[test]
fn default_when_nothing_known() {
assert_eq!(resolve(None, None, V4), mtu1500_shard_payload_for(V4));
assert_eq!(resolve(None, None, V6), mtu1500_shard_payload_for(V6));
assert_eq!(
resolve(None, None, NO_JUMBO, V4),
mtu1500_shard_payload_for(V4)
);
assert_eq!(
resolve(None, None, NO_JUMBO, V6),
mtu1500_shard_payload_for(V6)
);
}
#[test]
fn env_override_beats_learned() {
// 1280 wire 28 IP/UDP 64 header/crypto = 1188.
assert_eq!(resolve(Some(1280), Some(1472), V4), 1188);
assert_eq!(resolve(Some(1280), Some(1472), NO_JUMBO, V4), 1188);
}
#[test]
fn learned_budget_clamps() {
// A WARP-shaped path: 1280-byte UDP budget → 1280 64 = 1216.
assert_eq!(resolve(None, Some(1280), V4), 1216);
assert_eq!(resolve(None, Some(1280), NO_JUMBO, V4), 1216);
}
#[test]
fn learned_at_or_above_ceiling_is_the_default_wire() {
assert_eq!(resolve(None, Some(1472), V4), mtu1500_shard_payload_for(V4));
assert_eq!(resolve(None, Some(2000), V4), mtu1500_shard_payload_for(V4));
assert_eq!(
resolve(None, Some(1472), NO_JUMBO, V4),
mtu1500_shard_payload_for(V4)
);
assert_eq!(
resolve(None, Some(2000), NO_JUMBO, V4),
mtu1500_shard_payload_for(V4)
);
}
#[test]
fn env_full_mtu_is_the_default_wire_both_families() {
assert_eq!(resolve(Some(1500), None, V4), mtu1500_shard_payload_for(V4));
assert_eq!(resolve(Some(1500), None, V6), mtu1500_shard_payload_for(V6));
assert_eq!(
resolve(Some(1500), None, NO_JUMBO, V4),
mtu1500_shard_payload_for(V4)
);
assert_eq!(
resolve(Some(1500), None, NO_JUMBO, V6),
mtu1500_shard_payload_for(V6)
);
}
/// The happy path, both families: 9000 28 (IPv4) 64 = 8908, and 9000 48 64 = 8888.
#[test]
fn proven_and_reproven_path_starts_jumbo() {
assert_eq!(jumbo_session_start(proven_jumbo(), V4), Some(8908));
let mut v6 = proven_jumbo();
v6.live_udp_mtu = 8952;
v6.proven_udp_budget = Some(8952);
assert_eq!(jumbo_session_start(v6, V6), Some(8888));
// …and it is what `resolve` returns, ahead of the env branch that would clamp a
// >1500 `PUNKTFUNK_WIRE_MTU` back down to the family default.
assert_eq!(resolve(Some(9000), None, proven_jumbo(), V4), 8908);
}
/// THE guard: the laptop that proved jumbo on the wired LAN and came back on a 1500-MTU
/// link. The memory still says jumbo; the live connection says otherwise; the live one
/// wins, every time. This is what makes the grow as safe as the clamp.
#[test]
fn a_remembered_verdict_never_grows_without_a_live_reproof() {
let mut moved = proven_jumbo();
moved.live_udp_mtu = 1472; // a clean 1500-MTU path, freshly measured
assert_eq!(jumbo_session_start(moved, V4), None);
assert_eq!(
resolve(None, None, moved, V4),
mtu1500_shard_payload_for(V4)
);
// Not even one byte of headroom short of the sealed target is enough.
let mut nearly = proven_jumbo();
nearly.live_udp_mtu = 8971;
assert_eq!(jumbo_session_start(nearly, V4), None);
}
/// …and the mirror: a live-proven path with no prior verdict still starts at the default.
/// Both halves are required, so a single fluke on either side cannot seal a jumbo wire.
#[test]
fn a_live_proof_alone_does_not_grow() {
let mut first_ever = proven_jumbo();
first_ever.proven_udp_budget = None;
assert_eq!(jumbo_session_start(first_ever, V4), None);
let mut weak_memory = proven_jumbo();
weak_memory.proven_udp_budget = Some(1472);
assert_eq!(jumbo_session_start(weak_memory, V4), None);
}
/// The two memories are keyed differently (clamp: peer; verdict: route), so they can
/// disagree. When they do, the one that keeps datagrams small wins.
#[test]
fn a_constrained_path_clamp_vetoes_the_grow() {
let mut contradicted = proven_jumbo();
contradicted.clamped_udp_budget = Some(1280);
assert_eq!(jumbo_session_start(contradicted, V4), None);
// A clamp that is itself at or above the sealed target isn't contrary evidence.
let mut roomy = proven_jumbo();
roomy.clamped_udp_budget = Some(8972);
assert_eq!(jumbo_session_start(roomy, V4), Some(8908));
}
#[test]
fn without_the_operator_opt_in_nothing_grows() {
let mut no_optin = proven_jumbo();
no_optin.target_wire_mtu = None;
assert_eq!(jumbo_session_start(no_optin, V4), None);
}
/// A legacy client (no `Hello::max_shard_payload`) is never handed a geometry it did not
/// advertise, and a client whose ceiling lands under the family default is left alone
/// rather than being "grown" to something smaller.
#[test]
fn the_client_ceiling_is_binding() {
let mut legacy = proven_jumbo();
legacy.client_ceiling = 0;
assert_eq!(jumbo_session_start(legacy, V4), None);
let mut small = proven_jumbo();
small.client_ceiling = 1408;
assert_eq!(jumbo_session_start(small, V4), None);
// A ceiling between the default and the path target caps the grow — and the proof
// then only has to cover the SMALLER sealed size.
let mut capped = proven_jumbo();
capped.client_ceiling = 4000;
assert_eq!(jumbo_session_start(capped, V4), Some(4000));
}
/// Every shard payload the grow can produce is even (Leopard FEC splits shards in halves)
/// and fits the receive ceiling every client sizes its buffers from.
#[test]
fn grown_shards_stay_even_and_inside_the_receive_ceiling() {
for mtu in [2000usize, 4000, 4001, 9000, 9216, 64000] {
for peer in [V4, V6] {
let Some(t) = jumbo_target(Some(mtu), u16::MAX, peer) else {
continue;
};
assert_eq!(t % 2, 0, "odd shard for mtu {mtu}");
assert!(t <= punktfunk_core::config::max_shard_payload());
assert!(t > mtu1500_shard_payload_for(peer));
assert!(
sealed_datagram_bytes(t) <= punktfunk_core::packet::MAX_DATAGRAM_BYTES,
"sealed datagram overflows the receive ceiling at mtu {mtu}"
);
}
}
// Below the family default there is nothing to grow to.
assert_eq!(jumbo_target(Some(1500), u16::MAX, V4), None);
assert_eq!(jumbo_target(None, u16::MAX, V4), None);
}
/// A path is a (local interface, peer) pair, not a peer: the same client reached over the
/// host's other NIC is a different route with a different MTU.
#[test]
fn the_verdict_key_separates_routes_to_the_same_peer() {
let over_10g = PathKey {
local: Some(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1))),
peer: V4,
};
let over_wifi = PathKey {
local: Some(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1))),
peer: V4,
};
assert_ne!(over_10g, over_wifi);
assert_ne!(
over_10g,
PathKey {
local: None,
peer: V4
}
);
}
}
+23 -6
View File
@@ -193,12 +193,29 @@ impl SessionPlan {
// Surface the trade loudly: this is the single biggest per-frame cost a 4:4:4
// session adds (full-res CPU readback + swscale RGB→YUV444P every frame), and
// it looks like an unexplained fps ceiling if you don't know it happened.
tracing::warn!(
"4:4:4 session on the NVENC path without PUNKTFUNK_ZEROCOPY: zero-copy GPU \
capture DISABLED every frame is CPU RGB + swscale RGBYUV444P; expect a \
lower fps ceiling than 4:2:0 at this mode (set PUNKTFUNK_ZEROCOPY=1 for the \
GPU 4:4:4 convert)"
);
//
// Name the SESSION's codec, not the backend the gate is named after. The gate
// keys on `linux_zero_copy_is_vaapi()`, which reads the host-global encoder pref
// — so a per-session PyroWave negotiation on an NVENC/auto host lands here and
// was told it was "on the NVENC path", which is false in every particular: the
// wavelet encoder never touches NVENC, never swscales to YUV444P, and what it
// actually loses is the raw-dmabuf passthrough its whole design assumes.
if self.codec == crate::encode::Codec::PyroWave {
tracing::warn!(
"4:4:4 PyroWave session with PUNKTFUNK_ZEROCOPY off: zero-copy GPU \
capture DISABLED the wavelet encoder loses its raw-dmabuf passthrough \
and every frame becomes a full-resolution CPU readback plus an upload \
into its own Vulkan device; expect a materially lower fps ceiling (set \
PUNKTFUNK_ZEROCOPY=1 to restore the passthrough)"
);
} else {
tracing::warn!(
"4:4:4 session on the NVENC path without PUNKTFUNK_ZEROCOPY: zero-copy \
GPU capture DISABLED every frame is CPU RGB + swscale RGBYUV444P; \
expect a lower fps ceiling than 4:2:0 at this mode (set \
PUNKTFUNK_ZEROCOPY=1 for the GPU 4:4:4 convert)"
);
}
}
gpu && !force_cpu_for_nvenc_444
};
+194 -1
View File
@@ -48,6 +48,17 @@ pub struct Options {
pub out: PathBuf,
/// Also round-trip every AU through a `punktfunk_core` host→client loopback and verify.
pub loopback: bool,
/// PyroWave datagram-aligned packetization at this shard payload
/// ([`Encoder::set_wire_chunking`], plan §4.4) — what a real session passes from its
/// negotiated `shard_payload`. `None` = the dense one-packet-per-AU shape.
///
/// This is also the switch that makes the STREAMED-AU wire reachable from the spike: with
/// it set and `PUNKTFUNK_PYROWAVE_STREAMED_AU=1` armed, the encoder's `poll_chunk` hands the
/// AU out in window-aligned pieces and the loopback seals them through
/// `begin_streamed_frame_at`/`seal_streamed_chunk`/`seal_streamed_finish` — the same path a
/// `VIDEO_CAP_STREAMED_AU` client drives. Without it there is no way to exercise PW6 end to
/// end outside a real client session.
pub wire_chunk: Option<usize>,
}
pub fn run(opts: Options) -> Result<()> {
@@ -114,9 +125,21 @@ pub fn run(opts: Options) -> Result<()> {
refresh_hz: opts.fps,
})
.context("create virtual output")?;
// `resolve` is the shared GameStream/spike constructor and hard-codes `pyrowave: false`
// (GameStream never negotiates it). The spike DOES know its codec, and on Linux that
// flag is what puts the capture on the raw-dmabuf passthrough
// (`ZeroCopyPolicy::pyrowave_session`, set from the same comparison in
// `session_plan::output_format`). Left false, `--codec pyrowave` encoded PyroWave off a
// capture negotiated for somebody else, and the only way to exercise the real path was
// the host-global `PUNKTFUNK_ENCODER=pyrowave` lever — which ALSO flips
// `backend_is_vaapi`, so it cannot reproduce a per-session PyroWave negotiation on an
// auto/NVENC host at all. That is precisely the configuration PW2 exists for.
let mut want =
capture::OutputFormat::resolve(false, crate::encode::resolved_backend_is_gpu());
want.pyrowave = opts.codec == Codec::PyroWave;
capture::capture_virtual_output(
vout,
capture::OutputFormat::resolve(false, crate::encode::resolved_backend_is_gpu()),
want,
crate::session_plan::CaptureBackend::resolve(),
compositor == crate::vdisplay::Compositor::Kwin,
)
@@ -155,6 +178,18 @@ pub fn run(opts: Options) -> Result<()> {
)
.context("open encoder")?;
// Datagram-aligned packetization (§4.4) — and, with the PW6 knob armed, the gate that makes
// `supports_chunked_poll()` true so the drain below takes the streamed-AU path.
if let Some(c) = opts.wire_chunk {
encoder.set_wire_chunking(c);
tracing::info!(
shard_payload = c,
chunked_poll = encoder.supports_chunked_poll(),
"spike: wire chunking on (chunked_poll=false means PUNKTFUNK_PYROWAVE_STREAMED_AU \
is not armed the AU still goes out whole)"
);
}
let mut sink = BufWriter::new(
File::create(&opts.out).with_context(|| format!("create {}", opts.out.display()))?,
);
@@ -194,6 +229,12 @@ pub fn run(opts: Options) -> Result<()> {
out = %opts.out.display(),
elapsed_s = format!("{elapsed:.2}"),
encode_fps = format!("{:.1}", stats.encoded as f64 / elapsed.max(1e-9)),
// 0 = the whole-AU drain; > encoded = the streamed drain actually cut AUs into pieces.
chunks = stats.chunks,
chunks_per_au = format!(
"{:.1}",
stats.chunks as f64 / (stats.encoded.max(1)) as f64
),
"spike capture→encode→file complete"
);
@@ -217,6 +258,9 @@ struct Stats {
encoded: u64,
keyframes: u64,
bytes_out: u64,
/// Streamed-AU drain only: total chunks polled across all AUs (1 per AU means the cut never
/// engaged — the knob is off or the AU fits one chunk).
chunks: u64,
}
fn drain_encoder(
@@ -225,6 +269,12 @@ fn drain_encoder(
mut lb: Option<&mut Loopback>,
stats: &mut Stats,
) -> Result<()> {
// Streamed-AU drain (PW6): the encoder hands the finished AU out in shard-aligned pieces and
// the loopback seals each piece as it arrives, exactly as the native host's send thread does.
// Re-queried per drain, never cached — the trait's contract.
if encoder.supports_chunked_poll() {
return drain_encoder_chunked(encoder, sink, lb, stats);
}
while let Some(au) = encoder.poll().context("encoder poll")? {
sink.write_all(&au.data).context("write AU to file")?;
stats.encoded += 1;
@@ -239,6 +289,49 @@ fn drain_encoder(
Ok(())
}
/// The streamed-AU drain. Each chunk is sealed into the open wire frame the moment it is polled;
/// the concatenation is kept only so the completed AU can still be written to the file sink and
/// byte-compared against what the client reassembled — which is the point of the leg: it proves
/// the chunks the encoder cut, sealed through the sentinel-block wire, reassemble to EXACTLY the
/// AU `poll()` would have produced.
fn drain_encoder_chunked(
encoder: &mut dyn Encoder,
sink: &mut impl Write,
mut lb: Option<&mut Loopback>,
stats: &mut Stats,
) -> Result<()> {
let mut whole: Vec<u8> = Vec::new();
let mut chunks = 0u32;
while let Some(c) = encoder.poll_chunk().context("encoder poll_chunk")? {
if c.first {
whole.clear();
chunks = 0;
if let Some(lb) = lb.as_deref_mut() {
lb.streamed_begin(c.pts_ns, c.keyframe)?;
}
}
whole.extend_from_slice(&c.data);
chunks += 1;
if let Some(lb) = lb.as_deref_mut() {
lb.streamed_chunk(&c.data)?;
}
if !c.last {
continue;
}
sink.write_all(&whole).context("write AU to file")?;
stats.encoded += 1;
stats.bytes_out += whole.len() as u64;
stats.chunks += chunks as u64;
if c.keyframe {
stats.keyframes += 1;
}
if let Some(lb) = lb.as_deref_mut() {
lb.streamed_finish(&whole)?;
}
}
Ok(())
}
/// A host↔client `punktfunk_core` pair over a lossless in-process loopback. Each encoded AU is
/// FEC-protected, packetized, sent, then reassembled on the client and byte-compared to the
/// original — exercising the core on real encoder output (the spike "feed into a Session" goal).
@@ -249,6 +342,14 @@ struct Loopback {
recovered: u64,
mismatches: u64,
bytes: u64,
/// The streamed AU currently open (PW6). `Some` strictly between `streamed_begin` and
/// `streamed_finish`, mirroring the native send thread's `StreamedOpen`.
open: Option<punktfunk_core::packet::StreamedAu>,
/// Wire frame index for the streamed path. `submit_frame` uses the packetizer's internal
/// counter and `begin_streamed_frame_at` takes an explicit one; a session must use ONE
/// numbering style, and the spike never mixes them (`supports_chunked_poll()` is constant
/// for a PyroWave session, so every AU takes the same route).
next_index: u32,
}
impl Loopback {
@@ -265,9 +366,101 @@ impl Loopback {
recovered: 0,
mismatches: 0,
bytes: 0,
open: None,
next_index: 0,
})
}
/// Open a streamed AU on the wire (PW6). The client side needs no opt-in: a streamed frame
/// completes exactly like a whole one and is handed up as a single `Frame` — which is the
/// finding this leg exists to demonstrate rather than assert.
fn streamed_begin(&mut self, pts_ns: u64, keyframe: bool) -> Result<()> {
if self.open.is_some() {
return Err(anyhow!(
"streamed AU still open at begin — a previous AU never sent its `last` chunk"
));
}
let mut flags = FLAG_PIC as u32;
if keyframe {
flags |= FLAG_SOF as u32;
}
let idx = self.next_index;
self.next_index = self.next_index.wrapping_add(1);
self.open = Some(
self.host
.begin_streamed_frame_at(pts_ns, flags, idx)
.map_err(|e| anyhow!("begin_streamed_frame_at: {e:?}"))?,
);
Ok(())
}
/// Seal + send one encoder chunk. The returned batch is often EMPTY (the sealer buffers
/// until a whole FEC block accumulates) — that is the normal case, not an error.
fn streamed_chunk(&mut self, data: &[u8]) -> Result<()> {
let au = self
.open
.as_mut()
.ok_or_else(|| anyhow!("streamed chunk with no open AU"))?;
let wires = self
.host
.seal_streamed_chunk(au, data, false)
.map_err(|e| anyhow!("seal_streamed_chunk: {e:?}"))?;
self.send(wires)
}
/// Close the AU (final block carries the real totals) and verify what the client got.
fn streamed_finish(&mut self, expect: &[u8]) -> Result<()> {
let au = self
.open
.take()
.ok_or_else(|| anyhow!("streamed finish with no open AU"))?;
let wires = self
.host
.seal_streamed_finish(au)
.map_err(|e| anyhow!("seal_streamed_finish: {e:?}"))?;
self.send(wires)?;
self.submitted += 1;
self.bytes += expect.len() as u64;
self.verify(expect)
}
fn send(&mut self, wires: Vec<Vec<u8>>) -> Result<()> {
if wires.is_empty() {
return Ok(());
}
let refs: Vec<&[u8]> = wires.iter().map(|w| w.as_slice()).collect();
self.host
.send_sealed(&refs)
.map_err(|e| anyhow!("send_sealed: {e:?}"))?;
drop(refs);
self.host.reclaim_wires(wires);
Ok(())
}
/// Drain whatever the client can now reassemble and byte-compare it to `expect`.
fn verify(&mut self, expect: &[u8]) -> Result<()> {
loop {
match self.client.poll_frame() {
Ok(frame) => {
self.recovered += 1;
if frame.data != expect {
self.mismatches += 1;
tracing::warn!(
recovered = self.recovered,
got = frame.data.len(),
expected = expect.len(),
complete = frame.complete,
"loopback AU mismatch"
);
}
}
Err(punktfunk_core::PunktfunkError::NoFrame) => break,
Err(e) => return Err(anyhow!("client poll_frame: {e:?}")),
}
}
Ok(())
}
fn submit(&mut self, au: &EncodedFrame) -> Result<()> {
let mut flags = FLAG_PIC as u32;
if au.keyframe {
@@ -17,7 +17,38 @@ VK_ERROR_NOT_PERMITTED_KHR so a refused class NEVER regresses the encoder. Gated
NOTE: on an RTX 4090 / Windows / WDDM this did not reduce the spikes (the graphics-vs-compute
preemption granularity is the wall) — kept because it is correct, harmless (graceful fallback), and
may help other GPUs/drivers. Reduce the encode's GPU cost (4:2:0/8-bit) or use H.265 for a
GPU-saturated game.
GPU-saturated game. **That measurement is Windows/WDDM and does NOT transfer to Linux** — a
different driver stack with a different preemption model.
MEASURED ON LINUX/NVIDIA 2026-08-08, and it comes out the OTHER WAY: the elevated queue DOES cut
the tail. RTX 5070 Ti (driver 610.57.04), GRID 2 benchmark loop saturating the GPU at 54-87 %,
PyroWave 1080p, same binary in both arms (the only difference is CAP_SYS_NICE, i.e. whether the
class is granted at all), steady-state windows of 30 frames:
arm p50 p99 worst frame
default priority (refused) ~2.6 ms ~6.4 ms 9.5 ms
REALTIME granted ~3.2 ms ~4.4 ms 5.4 ms (repeat: p50 ~3.35, p99 ~4.8)
So on this stack the priority class buys a materially tighter TAIL — p99 down ~30 %, worst frame
roughly halved — at the cost of ~0.6 ms on the median. For a streaming encoder that is the right
side of the trade: the tail is what shows up as a visible hitch. Do NOT delete this patch on the
strength of the RTX 4090/WDDM result above; the two stacks disagree.
Caveats, so the number is not over-read: the arms were not interleaved and the background game
load drifted between them, capture was frame-starved (~2.5 fps) so this measures encode latency
under contention rather than a full-rate stream, and it is two granted runs against one refused
run. The direction was consistent across all 25 measurement windows.
NOTE 2 — WHERE THIS PATCH IS ACTUALLY LIVE. It is gated `if (!inherit_info)`, and only the WINDOWS
path leaves `inherit_info` null: `crates/pf-encode/src/enc/windows/pyrowave.rs` calls
`pyrowave_create_device_by_compat`, so Granite builds the device itself and this block runs.
**On LINUX it has never done anything.** `crates/pf-encode/src/enc/linux/pyrowave.rs::open_inner`
passes its own instance/device create-infos into `pyrowave_device_create_info`, Granite's
`MyDeviceFactory::get_existing_create_info()` returns them, `create_device` takes the inherit
branch, and the whole block above is skipped. The Linux request is therefore wired natively in
**`crates/pf-encode/src/enc/linux/pyrowave.rs`** (search `queue_priority_candidates`), which
implements the SAME env grammar and the SAME downgrade ladder so one knob means one thing on both
platforms. If you change the grammar here, change it there in the same commit.
diff --git a/crates/pyrowave-sys/vendor/pyrowave/Granite/vulkan/context.cpp b/crates/pyrowave-sys/vendor/pyrowave/Granite/vulkan/context.cpp
index 5257fc33..479eeded 100644
@@ -0,0 +1,123 @@
Encoder wire-sequence override — PUNKTFUNK LOCAL PATCH.
Not upstream. Exposes `Encoder::set_next_sequence(uint32_t)` (and a
`pyrowave_encoder_set_next_sequence` C entry) so the caller can stamp the 3-bit wire sequence
counter itself instead of relying on the encoder object's private one.
WHY IT EXISTS. PyroWave's `Encoder` structurally cannot hold two frames in flight: `Encoder::Impl`
owns ONE each of `wavelet_img_high_res`, `bucket_buffer`, `meta_buffer`, `block_stat_buffer`,
`payload_data` and `quant_buffer`, and `Impl::encode` OPENS by discarding them — an image barrier
with `VK_IMAGE_LAYOUT_UNDEFINED` as the old layout (a written promise nothing else is reading it)
plus three `fill_buffer` clears. Two `encode()` calls recorded into two command buffers and
submitted to the same queue have no execution dependency in Vulkan, so encode N+1's DWT would
overwrite the bands and zero the RDO buckets while encode N's block packing still reads them.
So overlapping frames means TWO encoder handles on one device, alternated — which is fine for
every resource above, because each handle gets its own. It is NOT fine for `sequence_count`, which
also lives on `Impl` and is stamped into every block header (pyrowave_encoder.cpp `packing_push`).
Two alternating handles each count 1,2,3... independently, so the wire sees 1,1,2,2,3,3...
That is silently fatal on the decode side. `pyrowave_decoder.cpp` computes
`diff = (hdr.sequence - last_seq) & 0x7` and treats `restart = diff != 0`, so a REPEATED value
reads as "more blocks of the same frame": `clear()` never runs, `decoded_frame_for_current_sequence`
stays true, and every second frame is swallowed. The symptom is "it works, just at half rate, with
occasional mixed-frame blocks" — the kind of failure that passes a smoke test. It would hit every
client, since pf-client-core and the Apple Metal hand-port parse the same field.
WHAT IT DOES. `set_next_sequence(seq)` stores `(seq - 1) & SequenceCountMask`, because
`Impl::encode` pre-increments before stamping — the setter's contract is about the next ENCODE, not
the next store. The Rust side keeps one monotonic counter across both handles and calls this before
each encode, so the wire sequence increments by exactly 1 mod 8 regardless of which handle produced
the frame.
INERT WHEN UNUSED. Nothing calls it unless the caller does, so the single-handle paths — including
the whole Windows backend — behave exactly as before. No `.def` change is needed: the C API is
built as a static archive (crates/pyrowave-sys/CMakeLists.txt).
Upstream status: not reported. It is a hook for a use case upstream explicitly designed against
("For low-latency use cases, overlapping frames in encode is meaningless due to latency and the
encoder is so fast anyway" — pyrowave.h). That reasoning holds at 1080p60 and stops holding at 4K
or under a GPU-bound game, which is what PW5 measured.
diff --git a/crates/pyrowave-sys/vendor/pyrowave/pyrowave.h b/crates/pyrowave-sys/vendor/pyrowave/pyrowave.h
index fc0d5834..aeb22ffc 100644
--- a/crates/pyrowave-sys/vendor/pyrowave/pyrowave.h
+++ b/crates/pyrowave-sys/vendor/pyrowave/pyrowave.h
@@ -476,6 +476,19 @@ PYROWAVE_PUBLIC_API pyrowave_result
pyrowave_encoder_packetize(pyrowave_encoder encoder, pyrowave_packet *packets, size_t packet_boundary,
size_t *out_packets, void *bitstream, size_t size);
+// PUNKTFUNK LOCAL EXTENSION (patches/0007-encoder-sequence-override.patch), not upstream.
+// The wire sequence counter is 3 bits (PyroWave::SequenceCountMask, pyrowave_common.hpp);
+// exported here so callers mask with the codec's own value instead of a copied literal.
+#define PYROWAVE_SEQUENCE_MASK 0x7u
+
+// Overrides the 3-bit wire sequence counter the NEXT encode will stamp into every block header.
+// The counter lives on the encoder object, so a caller that alternates TWO encoders to overlap
+// frames emits 1,1,2,2,3,3... and the decoder — which restarts a frame only when the value
+// CHANGES — reads the repeat as more blocks of the same frame and silently swallows every second
+// frame. Stamp a single monotonic counter across the handles with this. Value is masked to 3 bits.
+PYROWAVE_PUBLIC_API pyrowave_result
+pyrowave_encoder_set_next_sequence(pyrowave_encoder encoder, uint32_t sequence);
+
// Implementation ensures GPU is idle before destroying objects.
PYROWAVE_PUBLIC_API void
pyrowave_encoder_destroy(pyrowave_encoder encoder);
diff --git a/crates/pyrowave-sys/vendor/pyrowave/pyrowave_c.cpp b/crates/pyrowave-sys/vendor/pyrowave/pyrowave_c.cpp
index 985cd0a9..fcd7d6f8 100644
--- a/crates/pyrowave-sys/vendor/pyrowave/pyrowave_c.cpp
+++ b/crates/pyrowave-sys/vendor/pyrowave/pyrowave_c.cpp
@@ -1196,6 +1196,17 @@ pyrowave_encoder_packetize(pyrowave_encoder encoder, pyrowave_packet *packets, s
return PYROWAVE_SUCCESS;
}
+// PUNKTFUNK LOCAL EXTENSION (patches/0007-encoder-sequence-override.patch), not upstream.
+pyrowave_result
+pyrowave_encoder_set_next_sequence(pyrowave_encoder encoder, uint32_t sequence)
+{
+ Util::set_thread_logging_interface(&null_logger);
+ if (!encoder)
+ return PYROWAVE_ERROR_GENERIC;
+ encoder->encoder.set_next_sequence(sequence);
+ return PYROWAVE_SUCCESS;
+}
+
void pyrowave_encoder_destroy(pyrowave_encoder encoder)
{
auto *device = encoder->device;
diff --git a/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.cpp b/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.cpp
index ad4e9746..f23717f3 100644
--- a/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.cpp
+++ b/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.cpp
@@ -1230,6 +1230,14 @@ bool Encoder::encode(CommandBuffer &cmd, const ViewBuffers &views, const Bitstre
return impl->encode(cmd, views, buffers);
}
+// PUNKTFUNK: see the declaration in pyrowave_encoder.hpp. Impl::encode PRE-increments
+// (sequence_count = (sequence_count + 1) & mask before stamping), so store one less than the value
+// the caller wants stamped — the setter's contract is about the next ENCODE, not the next store.
+void Encoder::set_next_sequence(uint32_t sequence)
+{
+ impl->sequence_count = (sequence - 1) & SequenceCountMask;
+}
+
const Vulkan::ImageView &Encoder::get_wavelet_band(int component, int level)
{
return *impl->component_layer_views[component][level];
diff --git a/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.hpp b/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.hpp
index a65447d5..8c0ef0d0 100644
--- a/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.hpp
+++ b/crates/pyrowave-sys/vendor/pyrowave/pyrowave_encoder.hpp
@@ -37,6 +37,12 @@ public:
bool init(Vulkan::Device *device, int width, int height, ChromaSubsampling chroma);
bool encode(Vulkan::CommandBuffer &cmd, const ViewBuffers &views, const BitstreamBuffers &buffers);
+ // PUNKTFUNK: override the 3-bit wire sequence counter the NEXT encode will stamp.
+ // The counter is per-Encoder, so alternating two encoder objects to overlap frames emits
+ // 1,1,2,2,3,3... and the decoder reads a repeated value as "more blocks of the same frame".
+ // See crates/pyrowave-sys/patches/0007-encoder-sequence-override.patch.
+ void set_next_sequence(uint32_t sequence);
+
// Debug hackery
const Vulkan::ImageView &get_wavelet_band(int component, int level);
bool encode_pre_transformed(Vulkan::CommandBuffer &cmd, const BitstreamBuffers &buffers, float quant_scale);
+6 -1
View File
@@ -46,4 +46,9 @@ upstream:
realtime) so the wavelet encode can preempt a GPU-bound game on the shared shader cores. A
create loop downgrades on NOT_PERMITTED so a refused class never regresses the encoder. Did
not overcome the graphics-vs-compute preemption wall on an RTX 4090 (kept: correct + harmless,
may help other HW/drivers).
may help other HW/drivers) — that measurement is Windows/WDDM and does not transfer to Linux.
GATED ON !inherit_info, so it is LIVE ONLY ON THE WINDOWS PATH (pyrowave_create_device_by_compat,
where Granite builds its own device). Linux passes its own create-infos and takes the inherit
branch, so this patch is inert there; the Linux request lives natively in
crates/pf-encode/src/enc/linux/pyrowave.rs (queue_priority_candidates), with the same grammar
and the same downgrade ladder. Change one, change both.
+13
View File
@@ -476,6 +476,19 @@ PYROWAVE_PUBLIC_API pyrowave_result
pyrowave_encoder_packetize(pyrowave_encoder encoder, pyrowave_packet *packets, size_t packet_boundary,
size_t *out_packets, void *bitstream, size_t size);
// PUNKTFUNK LOCAL EXTENSION (patches/0007-encoder-sequence-override.patch), not upstream.
// The wire sequence counter is 3 bits (PyroWave::SequenceCountMask, pyrowave_common.hpp);
// exported here so callers mask with the codec's own value instead of a copied literal.
#define PYROWAVE_SEQUENCE_MASK 0x7u
// Overrides the 3-bit wire sequence counter the NEXT encode will stamp into every block header.
// The counter lives on the encoder object, so a caller that alternates TWO encoders to overlap
// frames emits 1,1,2,2,3,3... and the decoder — which restarts a frame only when the value
// CHANGES — reads the repeat as more blocks of the same frame and silently swallows every second
// frame. Stamp a single monotonic counter across the handles with this. Value is masked to 3 bits.
PYROWAVE_PUBLIC_API pyrowave_result
pyrowave_encoder_set_next_sequence(pyrowave_encoder encoder, uint32_t sequence);
// Implementation ensures GPU is idle before destroying objects.
PYROWAVE_PUBLIC_API void
pyrowave_encoder_destroy(pyrowave_encoder encoder);
+11
View File
@@ -1196,6 +1196,17 @@ pyrowave_encoder_packetize(pyrowave_encoder encoder, pyrowave_packet *packets, s
return PYROWAVE_SUCCESS;
}
// PUNKTFUNK LOCAL EXTENSION (patches/0007-encoder-sequence-override.patch), not upstream.
pyrowave_result
pyrowave_encoder_set_next_sequence(pyrowave_encoder encoder, uint32_t sequence)
{
Util::set_thread_logging_interface(&null_logger);
if (!encoder)
return PYROWAVE_ERROR_GENERIC;
encoder->encoder.set_next_sequence(sequence);
return PYROWAVE_SUCCESS;
}
void pyrowave_encoder_destroy(pyrowave_encoder encoder)
{
auto *device = encoder->device;
@@ -1230,6 +1230,14 @@ bool Encoder::encode(CommandBuffer &cmd, const ViewBuffers &views, const Bitstre
return impl->encode(cmd, views, buffers);
}
// PUNKTFUNK: see the declaration in pyrowave_encoder.hpp. Impl::encode PRE-increments
// (sequence_count = (sequence_count + 1) & mask before stamping), so store one less than the value
// the caller wants stamped — the setter's contract is about the next ENCODE, not the next store.
void Encoder::set_next_sequence(uint32_t sequence)
{
impl->sequence_count = (sequence - 1) & SequenceCountMask;
}
const Vulkan::ImageView &Encoder::get_wavelet_band(int component, int level)
{
return *impl->component_layer_views[component][level];
@@ -37,6 +37,12 @@ public:
bool init(Vulkan::Device *device, int width, int height, ChromaSubsampling chroma);
bool encode(Vulkan::CommandBuffer &cmd, const ViewBuffers &views, const BitstreamBuffers &buffers);
// PUNKTFUNK: override the 3-bit wire sequence counter the NEXT encode will stamp.
// The counter is per-Encoder, so alternating two encoder objects to overlap frames emits
// 1,1,2,2,3,3... and the decoder reads a repeated value as "more blocks of the same frame".
// See crates/pyrowave-sys/patches/0007-encoder-sequence-override.patch.
void set_next_sequence(uint32_t sequence);
// Debug hackery
const Vulkan::ImageView &get_wavelet_band(int component, int level);
bool encode_pre_transformed(Vulkan::CommandBuffer &cmd, const BitstreamBuffers &buffers, float quant_scale);
+1
View File
@@ -241,6 +241,7 @@ notes for context.
| `PUNKTFUNK_NVENC_SPLIT_ARBITRATE` | `1` | Opt-in: let the host change its split-encode decision **live**, mid-session, as the pixel rate moves, instead of only choosing once at session start. Currently wired on the Linux direct-NVENC path. Only interesting alongside `PUNKTFUNK_SPLIT_ENCODE=auto` at very high pixel rates. |
| `PUNKTFUNK_GPU_PRIORITY_CLASS` | `off` · `normal` · `high` · `realtime` · `auto` | **(Windows)** GPU scheduling priority for capture/encode under a GPU-saturating game. Default `auto` (starts `high`, upgrades to `realtime` when it's safe — e.g. HAGS off); `high` pins the static pre-gate behaviour; `realtime` is the strongest lever but can freeze NVENC on some setups. |
| `PUNKTFUNK_IDD_DEPTH` | `N` (default `2`) | **(Windows)** IDD-push pipeline depth. `1` cuts latency once GPU priority is raised; higher smooths a contended GPU. |
| `PYROWAVE_QUEUE_PRIORITY` | `realtime` *(default)* · `high` · `off` | [PyroWave](/docs/pyrowave) sessions only. PyroWave encodes on the same GPU shader cores a game uses, so a demanding game can starve it and the frame rate drops. This asks the driver to schedule the encode ahead of the game. `realtime` tries the strongest class and falls back to `high`; `high` asks only for the middle one; `off` disables the request. A driver that refuses simply encodes at normal priority — it can never stop a session starting. The packages grant the host the `CAP_SYS_NICE` capability this needs; on a host built or installed by hand it will be refused, and the host says so once at session start. Set `off` if you see the desktop stutter while streaming. |
## Diagnostics
+21
View File
@@ -101,6 +101,27 @@ dropped packets.
The stats overlay shows `pyrowave` as the decode path when the mode is active.
## Checking the host is really zero-copy
On a Linux host the CPU fallback mentioned above is not an error — the session still streams, it
just pays a full-resolution copy of every frame, which shows up as a lower frame-rate ceiling and
higher CPU use rather than as anything obviously broken. The host log states which path a session
took, once, when the capture starts:
```
capture pipeline resolved: dmabuf-passthrough → pyrowave
```
`dmabuf-passthrough` is the good one: the compositor's buffer goes straight into the wavelet
encoder. `cpu` means the copy is happening, and a second line says why — a compositor that would
not allocate a dmabuf, `PUNKTFUNK_ZEROCOPY` set to `0`, or a per-frame fall-through such as the
compositor serving shared memory after agreeing to dmabufs. Each distinct reason is logged once per
session with a running count, so a persistent downgrade is easy to tell from a hiccup while the
display mode settles.
If you see `cpu` and did not ask for it, check that `PUNKTFUNK_ZEROCOPY` is unset (it defaults to
on) and read the accompanying line — it names the cause and the fix.
## Current limits
- Linux and Windows hosts; Linux clients (the GTK desktop app and the session client, including
@@ -205,6 +205,36 @@ the host.
If the host answers, it's up. If not, check `journalctl --user -u punktfunk-host` on the host — on
a Windows host, run `punktfunk-host service status` from an elevated prompt on the machine itself.
## GPU scheduling priority
The Linux packages give the host binary one Linux capability, `CAP_SYS_NICE`, and it is worth
knowing why it is there and how to take it away.
The [PyroWave](/docs/pyrowave) codec encodes on the same GPU shader cores your game is using, so a
demanding game can crowd it out and the stream's frame rate drops with it. The fix is to ask the
driver to schedule the encode ahead of the game, and every driver we tested gates that request on
this capability: without it the request is simply refused and nothing changes. The other codecs use
a separate video engine on the GPU and are unaffected either way.
`CAP_SYS_NICE` lets a process raise its own scheduling priority. It grants no access to files,
the network or other users' processes, and it is **not** the same as running as root — the host
still runs as you, under your user session.
To check, or to take it away:
```sh
getcap /usr/bin/punktfunk-host # shows cap_sys_nice=ep when granted
sudo setcap -r /usr/bin/punktfunk-host # remove it; streaming still works
```
Removing it costs you nothing unless you stream PyroWave, and you can also just set
`PYROWAVE_QUEUE_PRIORITY=off` to stop the host asking. Note that a package update replaces the
binary and re-applies the capability.
Two side effects, if you are debugging the host: a binary carrying a capability is treated as
security-sensitive by the dynamic loader, so `LD_LIBRARY_PATH` and `LD_PRELOAD` are ignored for it,
and it does not write core dumps by default.
## Stopping and removing
After a Linux package update the user service keeps running the old binary until it's restarted, and
+26
View File
@@ -12,9 +12,33 @@ _ensure_punktfunk_group() {
getent group punktfunk >/dev/null 2>&1 || groupadd --system punktfunk 2>/dev/null || true
}
# CAP_SYS_NICE on the host binary — the GPU-scheduling grant.
#
# WHY: PyroWave encodes on the GPU's shader cores, so a GPU-bound game starves it (measured: the
# encode dispatch goes from ~2 ms to 15-18 ms at 95 % game load). The fix is an elevated
# global-priority Vulkan queue, which the driver gates on CAP_SYS_NICE — measured 2026-08-08 on an
# RTX 5070 Ti: WITHOUT the capability every priority class is refused, WITH it the encoder is
# granted REALTIME on the first attempt. RADV is the same. Without this line the knob exists and
# does nothing. Same capability, same mechanism, as our gamescope package sets on its own binary.
#
# NARROW: CAP_SYS_NICE only permits raising scheduling priority (nice/ioprio/affinity/RT class). It
# grants no filesystem, network or user-switching privilege, and it is NOT setuid.
#
# TWO CONSEQUENCES worth knowing before you debug something odd on this host:
# * a file capability makes the process AT_SECURE, so the dynamic loader IGNORES LD_LIBRARY_PATH
# and LD_PRELOAD for it. A library-path shim that used to work will silently stop.
# * core dumps are suppressed for capability-carrying binaries by default (fs.suid_dumpable).
#
# Never fails the install: a box without libcap, or a filesystem that cannot store capabilities
# (some overlay/NFS setups), just runs at default priority exactly as before.
_grant_sched_capability() {
setcap 'cap_sys_nice=ep' usr/bin/punktfunk-host 2>/dev/null || true
}
post_install() {
_ensure_update_group
_ensure_punktfunk_group
_grant_sched_capability
udevadm control --reload-rules 2>/dev/null || true
udevadm trigger --subsystem-match=misc 2>/dev/null || true
# Apply the UDP socket-buffer tuning now (also auto-applied at boot by systemd-sysctl).
@@ -73,6 +97,8 @@ post_upgrade() {
# root-only, and the virtual Steam Deck pad silently unable to attach. groupadd is idempotent, so
# this is a no-op on boxes that installed fresh.
_ensure_punktfunk_group
# A replaced binary is a NEW inode — file capabilities do not survive the upgrade, so re-grant.
_grant_sched_capability
udevadm control --reload-rules 2>/dev/null || true
sysctl -p /usr/lib/sysctl.d/99-punktfunk-net.conf >/dev/null 2>&1 || true
_warn_stale_firewall_ports
+25
View File
@@ -130,6 +130,31 @@ SYSEXT_VERSION_ID=$PF_VR
EXTENSION_RELOAD_MANAGER=1
EOF
# CAP_SYS_NICE on the host binary — the GPU-scheduling grant. PyroWave encodes on the GPU shader
# cores a game saturates, and the driver gates the elevated global-priority Vulkan queue that fixes
# it on this capability (measured 2026-08-08 on an RTX 5070 Ti: refused without it, granted REALTIME
# with it; RADV the same). Narrow — scheduling priority only, no filesystem/network privilege, not
# setuid.
#
# It has to be applied HERE, not in the merge hook: a merged sysext's /usr is a read-only squashfs,
# so nothing can setcap it afterwards. And it cannot ride in from the RPM either — the spec declares
# it with %caps, but rpm stores capabilities in its own header and `rpm2cpio | cpio` carries only
# the payload, so the staged file arrives with no capability at all. mksquashfs DOES record
# security.capability (only security.selinux is excluded below), so a setcap on the staging tree is
# what ends up in the image.
#
# Needs CAP_SETFCAP, i.e. root (or fakeroot) — a plain-user CI build cannot do it. That is not fatal:
# the image just ships as it does today and the encode runs at default GPU priority, so warn and
# carry on rather than fail a release build over a performance lever.
if [ -f "$STAGE/usr/bin/punktfunk-host" ]; then
if setcap 'cap_sys_nice=ep' "$STAGE/usr/bin/punktfunk-host" 2>/dev/null; then
echo "granted CAP_SYS_NICE to usr/bin/punktfunk-host (GPU-priority lever active)"
else
echo "WARNING: could not setcap CAP_SYS_NICE (need root/CAP_SETFCAP) — the image will ship" >&2
echo " without it and PyroWave will encode at default GPU priority." >&2
fi
fi
# SELinux labels as pseudo-xattrs (see header). matchpathcon resolves each target path against
# the targeted policy's file_contexts; <<none>> means "no specific entry" — skip those (the
# handful of matches all resolve to real contexts for our payload).
+10
View File
@@ -294,6 +294,16 @@ if [ "$1" = "configure" ]; then
# primitive that must not ride on the group users are told to join for gamepads
# (security-review 2026-08-05 M-4).
getent group punktfunk >/dev/null 2>&1 || addgroup --system punktfunk 2>/dev/null || true
# CAP_SYS_NICE — the GPU-scheduling grant. PyroWave encodes on the shader cores a game
# saturates, and the driver gates the elevated global-priority Vulkan queue that fixes it on
# this capability: measured 2026-08-08 on an RTX 5070 Ti, WITHOUT it every priority class is
# refused and WITH it the encoder is granted REALTIME first try (RADV behaves the same).
# Without this line the knob exists and does nothing. Narrow: it permits raising scheduling
# priority only — no filesystem, network or user-switching privilege, and no setuid. Note a
# capability-carrying binary is AT_SECURE, so the loader ignores LD_LIBRARY_PATH/LD_PRELOAD for
# it and core dumps are suppressed by default. Best-effort: a box without libcap, or a
# filesystem that cannot store capabilities, just runs at default priority as before.
setcap 'cap_sys_nice=ep' /usr/bin/punktfunk-host 2>/dev/null || true
# Pick up the /dev/uinput rule without a reboot (best-effort, no-op in containers).
udevadm control --reload-rules 2>/dev/null || true
udevadm trigger --subsystem-match=misc 2>/dev/null || true
+25 -1
View File
@@ -356,6 +356,26 @@ in
allowedUDPPorts = nativeUDP ++ optionals cfg.host.gamestream gamestreamUDP;
};
# CAP_SYS_NICE — the GPU-scheduling grant. PyroWave encodes on the GPU shader cores a game
# saturates; the elevated global-priority Vulkan queue that fixes it is gated on this
# capability (measured 2026-08-08, RTX 5070 Ti: without it EVERY priority class is refused,
# with it the encoder gets REALTIME on the first attempt; RADV behaves the same).
#
# NixOS cannot `setcap` a store path — it is read-only and shared — so this goes through
# `security.wrappers`, which builds a small setcap'd wrapper in /run/wrappers/bin. The unit's
# ExecStart points at the wrapper below; everything else about the host is unchanged.
#
# Narrow: CAP_SYS_NICE permits raising scheduling priority only — no filesystem, network or
# user-switching privilege, and the wrapper is capability-based, NOT setuid. Two side effects
# to know: the wrapped binary is AT_SECURE (the loader ignores LD_LIBRARY_PATH/LD_PRELOAD for
# it) and core dumps are suppressed by default.
security.wrappers.punktfunk-host = {
source = "${cfg.host.package}/bin/punktfunk-host";
capabilities = "cap_sys_nice=ep";
owner = "root";
group = "root";
};
systemd.user.services.punktfunk-host = {
description = "punktfunk GameStream + punktfunk/1 streaming host";
documentation = [ "https://git.unom.io/unom/punktfunk" ];
@@ -374,8 +394,12 @@ in
# PUNKTFUNK_GAMESCOPE_BIN so an operator's own override of that env still wins.
++ optional cfg.host.gamescopeHdr cfg.host.gamescopePackage;
serviceConfig = {
# Through the wrapper (see `security.wrappers.punktfunk-host` above), NOT the store path
# directly — the store path carries no capability and the GPU-priority lever would be
# inert. `config.security.wrapperDir` rather than a hard-coded /run/wrappers/bin so an
# operator who has moved it is still correct.
ExecStart =
"${cfg.host.package}/bin/punktfunk-host serve" + optionalString cfg.host.gamestream " --gamestream";
"${config.security.wrapperDir}/punktfunk-host serve" + optionalString cfg.host.gamestream " --gamestream";
Restart = "on-failure";
RestartSec = 2;
EnvironmentFile =
+9 -1
View File
@@ -477,7 +477,15 @@ install -Dm0644 scripts/punktfunk-scripting.service %{buildroot}%{_userunitdir}/
%files
%license LICENSE-MIT LICENSE-APACHE THIRD-PARTY-NOTICES.txt
%doc README.md packaging/README.md
%{_bindir}/punktfunk-host
# CAP_SYS_NICE — the GPU-scheduling grant, declared the RPM-native way so rpm applies it at
# install, restores it on upgrade, and VERIFIES it (a plain %post setcap does none of those).
# PyroWave encodes on the shader cores a game saturates; the elevated global-priority Vulkan queue
# that fixes it is gated on this capability. Measured 2026-08-08 on an RTX 5070 Ti: without it
# every priority class is refused, with it the encoder gets REALTIME first try (RADV the same).
# Narrow — scheduling priority only, no filesystem/network/user-switching privilege, not setuid.
# Consequences: the binary becomes AT_SECURE, so the loader ignores LD_LIBRARY_PATH/LD_PRELOAD for
# it, and core dumps are suppressed by default.
%caps(cap_sys_nice=ep) %{_bindir}/punktfunk-host
%{_bindir}/punktfunk-tray
%{_udevrulesdir}/60-punktfunk.rules
%dir %{_libexecdir}/punktfunk
+19
View File
@@ -344,6 +344,25 @@ if [ "$SUDO_OK" = 1 ]; then
warn "(everything else works; the pad arrives as a generic Xbox 360 controller). By hand:"
warn " sudo groupadd --system punktfunk; sudo usermod -aG punktfunk $USER"
fi
# CAP_SYS_NICE on the host binary — the GPU-scheduling grant, and the Deck is the box that
# needs it most: a Van Gogh APU shares one small GPU between the game and PyroWave's encode
# dispatch. The driver gates the elevated global-priority Vulkan queue on this capability
# (measured 2026-08-08 on an RTX 5070 Ti: refused without it, granted REALTIME with it; RADV
# behaves the same), so without this the knob exists and does nothing.
#
# The binary lives under $HOME, not /usr — so unlike the /etc drop-ins above this survives a
# SteamOS A/B update on its own and needs no atomic-keep entry. It DOES need re-applying after
# every rebuild, because a fresh binary is a new inode; re-running this installer does that.
#
# Narrow (scheduling priority only, no filesystem/network privilege, not setuid) and
# best-effort — a failure just means the encode runs at default priority as it does today.
if [ -x "$BIN" ]; then
if sudo setcap 'cap_sys_nice=ep' "$BIN" 2>/dev/null; then
ok "granted CAP_SYS_NICE (PyroWave encode can outrank a GPU-bound game)"
else
warn "could not grant CAP_SYS_NICE to $BIN — PyroWave encode stays at default GPU priority"
fi
fi
# SteamOS A/B updates rebuild /etc and DROP everything not on Valve's keep list — verified
# live: an OS update stripped the udev rule + vhci autoload + UDP sysctl (gamepads silently
# degrade to Xbox 360, buffers back to 208 KB). The sanctioned fix is a preserve drop-in in
+278
View File
@@ -9,6 +9,7 @@
use punktfunk_core::config::{Config, FecConfig, FecScheme, ProtocolPhase, Role};
use punktfunk_core::crypto::SessionKey;
use punktfunk_core::error::PunktfunkError;
use punktfunk_core::packet::{FLAG_PIC, FLAG_SOF, USER_FLAG_CHUNK_ALIGNED};
use punktfunk_core::session::Session;
use punktfunk_core::transport::loopback_pair;
@@ -83,6 +84,281 @@ fn run(
(completed, frames)
}
// ---------------------------------------------------------------------------
// PW6: partial delivery under loss — streamed AU vs whole AU
// ---------------------------------------------------------------------------
//
// The question this answers (wave-2 plan PW6, security-review finding 10): a PyroWave client
// enables `set_deliver_partial_frames` unconditionally, so a chunk-aligned AU that loses shards
// is still handed up as blocks-with-holes — one frame of localized blur instead of a freeze. But
// a STREAMED frame is excluded from that when it is UNPINNED: its size lives only on the FINAL
// block's headers (`frame_bytes` is the 0 sentinel until then), and `advance_window` refuses to
// deliver a partial it cannot truncate. So where the whole-AU path delivers blur, a streamed
// frame whose final block is entirely lost delivers NOTHING.
//
// Three legs, because a bare 2 % sweep cannot see the effect (see `partial_sweep`'s note):
// 1. `final_block_probe` — DETERMINISTIC: drop exactly the frame's last block in both shapes.
// Proves the mechanism exists (or does not) without any statistics.
// 2. `partial_sweep` — RANDOM Bernoulli loss, both shapes, same seed: the delivery rates.
// 3. the stress rows — the same sweep at higher loss, where the gap becomes measurable.
/// The realistic PyroWave wire geometry: 1500-MTU IPv4 shards, 200 data shards per FEC block,
/// and **FEC pinned OFF** (the Phase-4 recipe — parity would mask exactly the loss under study).
fn partial_config(role: Role) -> Config {
Config {
role,
phase: ProtocolPhase::P2Punktfunk,
fec: FecConfig {
scheme: FecScheme::Gf16,
fec_percent: 0,
max_data_per_block: 200,
},
shard_payload: 1408,
max_frame_bytes: 8 * 1024 * 1024,
encrypt: false,
key: SessionKey::Aes128Gcm([0u8; 16]),
salt: [0u8; 4],
loopback_drop_period: 0, // loss is injected here, per packet, so it can be random
}
}
/// Reproducible xorshift64* — the harness must be re-runnable to the same numbers, and
/// `loopback_drop_period`'s deterministic 1-in-N cannot model independent per-packet loss
/// (it would systematically hit or miss the final block, which is the whole question).
struct Rng(u64);
impl Rng {
fn new(seed: u64) -> Rng {
Rng(seed | 1)
}
fn next_u64(&mut self) -> u64 {
let mut x = self.0;
x ^= x >> 12;
x ^= x << 25;
x ^= x >> 27;
self.0 = x;
x.wrapping_mul(0x2545_F491_4F6C_DD1D)
}
/// True with probability `pct`/10000 (basis points, so 2 % = 200).
fn hits(&mut self, bp: u32) -> bool {
(self.next_u64() % 10_000) < bp as u64
}
fn range(&mut self, lo: usize, hi: usize) -> usize {
lo + (self.next_u64() as usize) % (hi - lo).max(1)
}
}
/// How each source frame ended up at the client.
#[derive(Default, Clone, Copy)]
struct Outcome {
complete: usize,
partial: usize,
nothing: usize,
}
impl Outcome {
fn total(&self) -> usize {
self.complete + self.partial + self.nothing
}
/// Partial deliveries as a percentage of frames that did NOT arrive complete — "when the
/// frame was damaged, how often did the user still get a picture?". That ratio, not the raw
/// count, is what the two wire shapes must be compared on: they damage different numbers of
/// frames at the same packet-loss rate (streamed adds no parity but does add a final block
/// whose loss is fatal, and the shapes' block splits differ slightly).
fn rescue_pct(&self) -> f64 {
let damaged = self.partial + self.nothing;
if damaged == 0 {
return 100.0;
}
100.0 * self.partial as f64 / damaged as f64
}
}
/// How many packets the frame's LAST FEC block occupies, and how many packets the whole AU
/// should seal into. With FEC pinned off there is no parity, so `packetize_each` emits exactly
/// one packet per data shard in block order — the final block is therefore the last `final_k`
/// packets of the batch. Returned together so the caller can ASSERT the packet count and fail
/// loudly if that emission shape ever changes, rather than silently probing the wrong packets.
fn final_block_span(len: usize, shard: usize, per_block: usize) -> (usize, usize) {
let shards = len.div_ceil(shard);
let blocks = shards.div_ceil(per_block);
let final_k = shards - (blocks - 1) * per_block;
(shards, final_k)
}
/// Drive `frames` AUs through a host→client pair and classify each one. `loss_bp` is the
/// per-packet loss probability in basis points; `final_only` instead forces the frame's LAST
/// block to be dropped wholesale, and nothing else (the deterministic mechanism probe).
///
/// `sizes` gives each frame's AU length. Real PyroWave AUs vary frame to frame under rate
/// control, and the FINAL block's size is what bounds this trap's exposure, so the sweep varies
/// the length across the whole 1..=200-shard range of final-block sizes rather than pinning one.
fn run_partial(
streamed: bool,
sizes: &[usize],
loss_bp: u32,
final_only: bool,
seed: u64,
) -> Outcome {
// Flush frames: `advance_window` only ages a frame out once something NEWER exists and the
// capture-time fuse has passed (PARTIAL_WINDOW_NS = 30 ms vs a 16.67 ms frame period), so
// the tail of the run needs successors before its verdicts land.
const FLUSH: usize = 8;
const FRAME_NS: u64 = 16_666_667;
let (h, c) = loopback_pair(0, 0);
let mut host = Session::new(partial_config(Role::Host), Box::new(h)).unwrap();
let mut client = Session::new(partial_config(Role::Client), Box::new(c)).unwrap();
// The PyroWave client's real setting (`client/pump/handshake.rs` turns this on for every
// CODEC_PYROWAVE session).
client.set_deliver_partial_frames(true);
let mut rng = Rng::new(seed);
// frame_index -> saw a complete delivery
let mut delivered: std::collections::HashMap<u32, bool> = std::collections::HashMap::new();
let flags = FLAG_PIC as u32 | FLAG_SOF as u32 | USER_FLAG_CHUNK_ALIGNED;
let n = sizes.len();
for f in 0..(n + FLUSH) {
let len = sizes[f.min(n - 1)];
// Busy, frame-varying content — never a flat fill (a constant buffer would still
// reassemble byte-identically, but it makes every debug dump look alike).
let data: Vec<u8> = (0..len).map(|b| (b.wrapping_mul(31) ^ f) as u8).collect();
let pts = f as u64 * FRAME_NS;
let fi = f as u32;
// Send one sealed batch. `kill_from` is the index at/after which every packet is dropped
// outright (the deterministic final-block probe); otherwise each packet is lost
// independently at `loss_bp`.
let mut send = |host: &mut Session, wires: Vec<Vec<u8>>, kill_from: usize| {
let refs: Vec<&[u8]> = wires
.iter()
.enumerate()
.filter(|(i, _)| *i < kill_from && !(loss_bp > 0 && rng.hits(loss_bp)))
.map(|(_, w)| w.as_slice())
.collect();
if !refs.is_empty() {
host.send_sealed(&refs).unwrap();
}
drop(refs);
host.reclaim_wires(wires);
};
if streamed {
let mut au = host.begin_streamed_frame_at(pts, flags, fi).unwrap();
// Cut at the encoder's chunk granularity (the PW6 `AuChunker` default: 256 KiB
// rounded down to whole 1408-byte windows = 186 windows).
for chunk in data.chunks(186 * 1408) {
let wires = host.seal_streamed_chunk(&mut au, chunk, false).unwrap();
send(&mut host, wires, usize::MAX);
}
let wires = host.seal_streamed_finish(au).unwrap();
// The finish batch IS the final block — the only one carrying the real totals.
send(&mut host, wires, if final_only { 0 } else { usize::MAX });
} else {
let wires = host.seal_frame_at(&data, pts, flags, fi).unwrap();
let (shards, final_k) = final_block_span(len, 1408, 200);
assert_eq!(
wires.len(),
shards,
"FEC is off, so the whole-AU batch must be exactly one packet per data shard — \
the final-block probe's index rule depends on it"
);
let kill_from = if final_only {
wires.len() - final_k
} else {
usize::MAX
};
send(&mut host, wires, kill_from);
}
loop {
match client.poll_frame() {
Ok(got) => {
let e = delivered.entry(got.frame_index).or_insert(false);
*e |= got.complete;
}
Err(PunktfunkError::NoFrame) => break,
Err(e) => panic!("unexpected error: {e}"),
}
}
}
let mut out = Outcome::default();
for f in 0..n {
match delivered.get(&(f as u32)) {
Some(true) => out.complete += 1,
Some(false) => out.partial += 1,
None => out.nothing += 1,
}
}
out
}
/// AU lengths spanning the full range of FINAL-block sizes (1..=200 shards on top of two full
/// 200-shard blocks) — 564 KB…845 KB, i.e. the 400 Mb/s-at-60fps operating point.
fn varied_sizes(count: usize, seed: u64) -> Vec<usize> {
let mut rng = Rng::new(seed);
(0..count)
.map(|_| rng.range(401 * 1408, 600 * 1408 + 1))
.collect()
}
fn partial_section() {
let frames: usize = std::env::var("PW6_FRAMES")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(2000);
println!("\n\npunktfunk PW6 — partial delivery under loss: STREAMED vs WHOLE AU");
println!("(chunk-aligned AUs, deliver_partial ON, FEC pinned OFF, shard 1408, 200/block)\n");
// ---- Leg 1: the mechanism, deterministically -------------------------------------------
println!("Leg 1 — DETERMINISTIC probe: the frame's LAST block is lost, nothing else.");
let sizes = varied_sizes(200, 0xC0FFEE);
for (label, streamed) in [("whole-AU", false), ("streamed", true)] {
let o = run_partial(streamed, &sizes, 0, true, 1);
println!(
" {label:>8}: complete {:>4} partial {:>4} NOTHING {:>4} (of {})",
o.complete,
o.partial,
o.nothing,
o.total()
);
}
println!(
" → if the streamed row shows NOTHING where whole-AU shows partial, the trap is real."
);
// ---- Leg 2 + 3: rates under random loss -------------------------------------------------
println!("\nLeg 2/3 — RANDOM per-packet loss, same seed and same AU sizes for both shapes.");
println!(" 'rescue' = partials / (partials + nothing): of the frames that arrived DAMAGED,");
println!(" how many still reached the decoder as blur instead of vanishing.\n");
println!(
"{:>7} {:>9} {:>26} {:>26}",
"loss", "shape", "complete / partial / none", "rescue of damaged"
);
println!("{}", "-".repeat(78));
let sizes = varied_sizes(frames, 0xBEEF);
for &bp in &[200u32, 1000, 3000, 5000] {
for (label, streamed) in [("whole-AU", false), ("streamed", true)] {
let o = run_partial(streamed, &sizes, bp, false, 0x5EED);
println!(
"{:>6.1}% {label:>9} {:>8} / {:>7} / {:>5} {:>24.2}%",
bp as f64 / 100.0,
o.complete,
o.partial,
o.nothing,
o.rescue_pct()
);
}
}
println!(
"\nNote: at 2 % the streamed penalty is bounded by P(final block fully lost) =\n\
E[0.02^k] over final-block sizes k ~1e-4 so the 2 % row is EXPECTED to tie.\n\
The higher-loss rows are what make the gap (if any) visible; Leg 1 proves the mechanism."
);
}
fn main() {
let frames = 50;
let frame_len = 100_000; // ~98 shards across 2 FEC blocks
@@ -111,4 +387,6 @@ fn main() {
);
}
println!("\nNote: recovery drops off once per-block loss exceeds the 25% recovery budget.");
partial_section();
}