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(); }