From 20f4d23f2db6f7031b65bce9e27696c98b778227 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Sat, 8 Aug 2026 21:26:25 +0200 Subject: [PATCH] =?UTF-8?q?test(pw6):=20the=20streamed-AU=20trap=20is=20re?= =?UTF-8?q?al=20=E2=80=94=20and=20at=202=20%=20loss=20it=20costs=20exactly?= =?UTF-8?q?=20nothing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PW6 shipped behind a knob because one pre-registered risk was unmeasured: a streamed frame whose FINAL block is lost has no totals, so where the whole-AU path hands the consumer a usable blurred partial, a streamed frame may deliver nothing. PyroWave clients opt into partial delivery unconditionally, so this would have been a live behaviour change for every one of them. Measured now, three ways, instead of reasoned about. `tools/loss-harness` gains a partial-delivery leg: FEC pinned OFF, chunk-aligned AUs, deliver_partial ON, realistic 1408/200 geometry, and AU sizes swept across the whole 1..=200-shard range of FINAL-block sizes — because the final block's size is what bounds the exposure. Loss is injected per packet from a seeded xorshift rather than through `loopback_drop_period`, whose deterministic 1-in-N would systematically always-or-never hit the final block, which is the entire question. `tc netem` on `lo` was deliberately not used: the in-process model gives exact per-frame attribution, needs no sudo, cannot disturb a box running a live desktop session, and — decisively — can drop precisely the final block. Leg 1, deterministic (drop exactly the last block, 200 frames): whole-AU delivers 200 partials and 0 losses; streamed delivers 0 partials and 200 total losses. The trap is real and, when it fires, total. Leg 2, random loss, 20 000 frames per cell, same seed and sizes for both shapes. At 2 % the two are indistinguishable — 20000/20000 partials and ZERO vanished frames on both, matching the analytic bound E[loss^k] over final-block sizes k (~1e-4). The gap only appears at 30 % (99.94 % vs 100 % rescue) and 50 % (99.79 %). `complete` is 0 throughout by construction: with FEC off and ~500 packets per AU, essentially every frame is damaged — which is the regime the partial path exists for. The spike gains `--wire-chunk` and a streamed loopback path, so the wire shape is reachable end to end outside a real client: `poll_chunk` drains the AU, `begin_streamed_frame_at`/`seal_streamed_chunk`/`seal_streamed_finish` seal each piece, and the client byte-compares the reassembly. On 120 real PyroWave AUs the streamed legs (56.5 and 2.0 chunks/AU) and the whole-AU control emit a byte-identical 47 373 568-byte stream with 0 mismatches — the cut changes the wire shape and not one byte of content, and with the knob unset it does not engage at all. A new `#[ignore]`d GPU test closes the picture question on real hardware with a BUSY card (gradients + checker + noise), never a flat fill: chunks are whole windows, exactly one `first` and one `last`, the AU decodes through the client's own window walk, and luma PSNR lands at 40.2 dB. Unset the knob and the test refuses to run, which is the default-off claim verified rather than asserted. Verdict recorded in the plan: KEEP IT OFF. The 2 % tie is an argument about typical loss, but the failure is not graceful when it fires and the measured win is host send-side pipelining that nobody has yet put a millisecond number on. --- crates/pf-encode/src/enc/linux/pyrowave.rs | 143 +++++++++++ crates/punktfunk-host/src/main.rs | 13 + crates/punktfunk-host/src/spike.rs | 181 ++++++++++++++ tools/loss-harness/src/main.rs | 278 +++++++++++++++++++++ 4 files changed, 615 insertions(+) diff --git a/crates/pf-encode/src/enc/linux/pyrowave.rs b/crates/pf-encode/src/enc/linux/pyrowave.rs index 4a316652..574beeba 100644 --- a/crates/pf-encode/src/enc/linux/pyrowave.rs +++ b/crates/pf-encode/src/enc/linux/pyrowave.rs @@ -2607,4 +2607,147 @@ mod tests { assert!(!priority_refused(vk::Result::ERROR_EXTENSION_NOT_PRESENT)); assert!(!priority_refused(vk::Result::SUCCESS)); } + + // ---- PW6: the streamed-AU cut, on real GPU output ---------------------------------------- + // Appended at module END per the wave plan's ownership rule. + + /// Walk a windowed AU back into the flat codec-packet stream — the clients' parse + /// (`video_pyrowave.rs::push_window`, Apple's `MetalWaveletDecoder`), so upstream's decoder + /// sees exactly what a real client would feed it. + fn walk_windows(au: &[u8], window: usize) -> Vec { + let mut stream = Vec::new(); + let mut frag: Vec = Vec::new(); + for win in au.chunks(window) { + let used = u16::from_le_bytes([win[0], win[1]]) as usize; + let kind = u16::from_le_bytes([win[2], win[3]]); + let body = &win[4..4 + used]; + match kind { + 0 => stream.extend_from_slice(body), + 1 => frag = body.to_vec(), + 2 => frag.extend_from_slice(body), + 3 => { + frag.extend_from_slice(body); + stream.extend_from_slice(&frag); + frag.clear(); + } + k => panic!("unknown window kind {k}"), + } + } + stream + } + + /// Luma PSNR (dB) of a decoded Y plane against the BT.709 limited-range luma of the source + /// BGRA — the same math `rgb2yuv.comp` runs on the GPU. Luma only: chroma is subsampled on + /// the 4:2:0 path, and luma is where wavelet quantisation shows. + fn luma_psnr(src_bgra: &[u8], decoded_y: &[u8]) -> f64 { + assert_eq!(src_bgra.len(), decoded_y.len() * 4); + let mut sse = 0.0f64; + for (px, &got) in src_bgra.chunks_exact(4).zip(decoded_y) { + let (b, g, r) = (px[0] as f64, px[1] as f64, px[2] as f64); + let want = 16.0 + 0.1826 * r + 0.6142 * g + 0.0620 * b; + let d = want - got as f64; + sse += d * d; + } + let mse = sse / decoded_y.len() as f64; + if mse <= f64::EPSILON { + return f64::INFINITY; + } + 10.0 * (255.0f64 * 255.0 / mse).log10() + } + + /// PW6 on-glass: with `PUNKTFUNK_PYROWAVE_STREAMED_AU=1` armed, a real GPU encode of a BUSY + /// test card must come out of `poll_chunk` in several window-aligned pieces that concatenate + /// to a decodable AU — and the picture must survive, verified by PSNR against the CSC's own + /// BT.709 math rather than by "it ran". + /// + /// Flat fills are useless here (they false-greened the Windows bring-up): a solid colour + /// reassembles convincingly even when whole subbands are missing. The busy card puts energy + /// in every subband, so a cut that lost or reordered a window shows up as a PSNR collapse. + /// + /// `#[ignore]`d: needs a real Vulkan 1.3 GPU. + /// cargo test -p pf-encode --features pyrowave --no-run + /// PUNKTFUNK_PYROWAVE_STREAMED_AU=1 --ignored --nocapture pyrowave_streamed_chunks + #[test] + #[ignore = "needs a real Vulkan 1.3 compute device (run on a GPU host, not the build box)"] + fn pyrowave_streamed_chunks_reassemble_and_keep_the_picture() { + const WINDOW: usize = 1408; + // 1280x720 at 60 Mb/s ≈ 125 KB/AU — comfortably several 256 KiB-target chunks' worth of + // windows at the default step once the step is clamped to the AU, and big enough that the + // AU spans many windows. + let (w, h) = (1280u32, 720u32); + let mut enc = PyroWaveEncoder::open(w, h, 60, 200_000_000, crate::ChromaFormat::Yuv420) + .expect("open pyrowave encoder"); + enc.set_wire_chunking(WINDOW); + + assert!( + enc.supports_chunked_poll(), + "PUNKTFUNK_PYROWAVE_STREAMED_AU=1 must be set in the ENVIRONMENT of this test binary \ + — without it PW6 is off by design and there is nothing to verify" + ); + + for seed in [7u32, 11, 13] { + let frame = test_card(w, h, seed); + let FramePayload::Cpu(ref src) = frame.payload else { + panic!("test card is a CPU frame") + }; + let src = src.clone(); + enc.submit(&frame).expect("submit"); + + // Drain the AU through the chunked poll, exactly as the native pump does. + let mut au = Vec::new(); + let (mut chunks, mut firsts, mut lasts) = (0u32, 0u32, 0u32); + loop { + let c = enc + .poll_chunk() + .expect("poll_chunk") + .expect("an AU is in flight"); + assert!(c.chunk_aligned, "wire chunking is on"); + assert!(c.keyframe, "every pyrowave AU is a keyframe"); + assert_eq!( + c.data.len() % WINDOW, + 0, + "every chunk is a whole number of windows — a cut inside a window would \ + split the 4-byte framing prefix from its body" + ); + chunks += 1; + firsts += u32::from(c.first); + lasts += u32::from(c.last); + au.extend_from_slice(&c.data); + if c.last { + break; + } + } + assert_eq!(firsts, 1, "exactly one opening chunk"); + assert_eq!(lasts, 1, "exactly one closing chunk"); + assert!( + chunks > 1, + "seed {seed}: the AU came out in ONE piece ({} B) — the cut never engaged, so \ + this run proves nothing about PW6", + au.len() + ); + assert_eq!(au.len() % WINDOW, 0, "the AU is a whole number of windows"); + + // A second `poll_chunk` must report the AU is done, not dribble more bytes. + assert!( + enc.poll_chunk().expect("poll_chunk after last").is_none(), + "no AU is in flight once `last` was handed out" + ); + + // The picture: window-walk (the client's parse) → upstream's own decoder → PSNR. + let stream = walk_windows(&au, WINDOW); + // SAFETY: test-only FFI into the vendored decoder with locally-owned buffers. + let (y, _cb, _cr) = unsafe { decode_planes(w, h, &stream) }; + let psnr = luma_psnr(&src, &y); + eprintln!( + "seed {seed}: {chunks} chunks, {} B AU ({} windows), luma PSNR {psnr:.2} dB", + au.len(), + au.len() / WINDOW + ); + assert!( + psnr > 30.0, + "seed {seed}: luma PSNR {psnr:.2} dB — the streamed reassembly lost or reordered \ + picture data (a flat-fill test would NOT have caught this)" + ); + } + } } diff --git a/crates/punktfunk-host/src/main.rs b/crates/punktfunk-host/src/main.rs index 9f5b797c..209e7e88 100644 --- a/crates/punktfunk-host/src/main.rs +++ b/crates/punktfunk-host/src/main.rs @@ -844,6 +844,7 @@ fn parse_spike(args: &[String]) -> Result { let mut bitrate_mbps = 20u64; let mut out: Option = None; let mut loopback = true; + let mut wire_chunk: Option = None; let mut i = 0; while i < args.len() { @@ -906,6 +907,12 @@ fn parse_spike(args: &[String]) -> Result { } "--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); @@ -940,6 +947,7 @@ fn parse_spike(args: &[String]) -> Result { bitrate_bps: bitrate_mbps.saturating_mul(1_000_000), out, loopback, + wire_chunk, }) } @@ -1020,6 +1028,11 @@ SPIKE OPTIONS: --width --height synthetic source size (default: 1920x1080) --out raw Annex-B output (default: /tmp/punktfunk-spike.) --no-loopback skip the punktfunk_core round-trip verification + --wire-chunk 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: diff --git a/crates/punktfunk-host/src/spike.rs b/crates/punktfunk-host/src/spike.rs index dca0f091..d1e922a0 100644 --- a/crates/punktfunk-host/src/spike.rs +++ b/crates/punktfunk-host/src/spike.rs @@ -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, } pub fn run(opts: Options) -> Result<()> { @@ -167,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()))?, ); @@ -206,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" ); @@ -229,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( @@ -237,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; @@ -251,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 = 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). @@ -261,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, + /// 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 { @@ -277,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>) -> 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 { diff --git a/tools/loss-harness/src/main.rs b/tools/loss-harness/src/main.rs index 0caeb881..2482a37b 100644 --- a/tools/loss-harness/src/main.rs +++ b/tools/loss-harness/src/main.rs @@ -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 = 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 = (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>, 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 { + 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(); }