diff --git a/clients/android/native/src/session/connect.rs b/clients/android/native/src/session/connect.rs index 839e0835..eefbe48e 100644 --- a/clients/android/native/src/session/connect.rs +++ b/clients/android/native/src/session/connect.rs @@ -225,6 +225,9 @@ pub extern "system" fn Java_io_unom_punktfunk_kit_NativeBridge_nativeConnect<'lo // No non-video caps: this client does not render the host cursor locally (no shape/state // planes in the jni surface), so advertising CLIENT_CAP_CURSOR would stream cursor-less. 0, + // Slice-progressive delivery: off until the decode loop feeds MediaCodec with + // BUFFER_FLAG_PARTIAL_FRAME (P2d flips this behind a FEATURE_PartialFrame probe). + false, launch, // a store-qualified library id to boot into a game, or None for the desktop device_name, // Kotlin's Build.MODEL — the host's approval-list / trust-store label pin, // Some → Crypto on host-fp mismatch diff --git a/clients/cli/src/main.rs b/clients/cli/src/main.rs index 011dc883..b359f7b1 100644 --- a/clients/cli/src/main.rs +++ b/clients/cli/src/main.rs @@ -798,14 +798,15 @@ from the config directory for a true factory reset." }, punktfunk_core::config::CompositorPref::Auto, punktfunk_core::config::GamepadPref::Auto, - 0, // bitrate_kbps: the host's default; this connect never presents - 0, // video_caps: nothing decodes here - 2, // audio_channels - 0, // video_codecs: the probe carries no video - 0, // preferred_codec - None, // display_hdr - 0, // client_caps: nothing renders a cursor - None, // launch + 0, // bitrate_kbps: the host's default; this connect never presents + 0, // video_caps: nothing decodes here + 2, // audio_channels + 0, // video_codecs: the probe carries no video + 0, // preferred_codec + None, // display_hdr + 0, // client_caps: nothing renders a cursor + false, // frame_parts: probe/whole-AU consumer + None, // launch Some(punktfunk_core::client::device_name()), Some(pin), Some(identity), diff --git a/clients/linux/src/app.rs b/clients/linux/src/app.rs index 5fef2feb..eb3cb4f2 100644 --- a/clients/linux/src/app.rs +++ b/clients/linux/src/app.rs @@ -696,9 +696,10 @@ impl AppModel { 2, // audio_channels: stereo crate::video::decodable_codecs(), // codecs (unused by the probe, but honest) 0, // preferred_codec: no preference - None, // display_hdr: probe connect, nothing presents - 0, // client_caps: probe connect, nothing renders a cursor - None, // launch: probe connect, no game + None, // display_hdr: probe connect, nothing presents + 0, // client_caps: probe connect, nothing renders a cursor + false, // frame_parts: probe/whole-AU consumer + None, // launch: probe connect, no game // Knock under this device's name, not a fingerprint placeholder, when the // probed host doesn't know us yet. Some(pf_client_core::trust::device_name()), diff --git a/clients/windows/src/probe.rs b/clients/windows/src/probe.rs index e159c5fc..b574bb37 100644 --- a/clients/windows/src/probe.rs +++ b/clients/windows/src/probe.rs @@ -54,10 +54,11 @@ pub fn run_speed_probe( 0, // video_caps: probe connect, nothing is decoded 2, // audio_channels: stereo baseline decodable_codecs(), - 0, // preferred_codec: no preference - None, // display_hdr: probe connect, nothing presents - 0, // client_caps: probe connect, nothing renders a cursor - None, // launch: no game + 0, // preferred_codec: no preference + None, // display_hdr: probe connect, nothing presents + 0, // client_caps: probe connect, nothing renders a cursor + false, // frame_parts: probe/whole-AU consumer + None, // launch: no game // Same label a real session sends — a speed test against a host that doesn't know us yet // should knock under this device's name, not a fingerprint placeholder. Some(punktfunk_core::client::device_name()), diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 65746854..93c6dc38 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -272,6 +272,9 @@ fn pump( } else { 0 }, + // Slice-progressive delivery: off — this presenter feeds FFmpeg whole AUs; a partial + // avcodec feed path can flip it later. + false, params.launch.clone(), // The host's approval-list / trust-store label for this client. Without it every no-PIN // "request access" knock showed up as the fingerprint placeholder "device abcd1234". diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index 54495f87..adadd089 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -1871,6 +1871,11 @@ unsafe fn connect_ex_impl( // compositing the pointer — only an embedder that renders the cursor planes // ([`punktfunk_connection_next_cursor_shape`]/`_state`) may set it. ex7/ex8 pass 0. client_caps, + // The C ABI cannot carry slice-progressive parts yet — `PunktfunkFrame` has no + // part/completeness fields, so a part would be indistinguishable from a whole AU. + // An `ex10` variant adds the opt-in together with those fields when an ABI embedder + // (Apple) grows a partial-feed decode path. + false, launch, // The C ABI has no device-name parameter (only `punktfunk_pair` takes one), so every // embedder gets the OS hostname default — this is what the host's pending-approval diff --git a/crates/punktfunk-core/src/client/frame_channel.rs b/crates/punktfunk-core/src/client/frame_channel.rs index c011d225..4d5c3dad 100644 --- a/crates/punktfunk-core/src/client/frame_channel.rs +++ b/crates/punktfunk-core/src/client/frame_channel.rs @@ -321,9 +321,18 @@ impl FrameChannel { self.ready.notify_one(); } - /// Pump side: current queued depth — the clock-free standing-queue signal. + /// Pump side: current queued depth in ACCESS UNITS — the clock-free standing-queue + /// signal. Slice-progressive parts of a still-open AU don't count (the detector's + /// thresholds are in frames; counting parts would trip it at a fraction of the real + /// backlog), and a queue that stops draining still accumulates completed AUs. pub(crate) fn depth(&self) -> usize { - self.inner.lock().unwrap().q.len() + self.inner + .lock() + .unwrap() + .q + .iter() + .filter(|f| f.complete) + .count() } /// Pump side: discard the whole backlog (the jump-to-live path); returns how many were dropped. @@ -378,10 +387,34 @@ mod frame_channel_tests { pts_ns: i as u64, flags: 0, complete: true, + part: None, received_ns: 0, } } + /// `depth()` is the standing-queue detector's signal, in AU units: parts of a + /// still-open AU must not count (the thresholds are frames — parts would trip + /// jump-to-live at a fraction of the real backlog), while completed AUs always do. + #[test] + fn depth_counts_aus_not_parts() { + let ch = FrameChannel::new(); + let mut p = frame(1); + p.complete = false; + p.part = Some(crate::session::FramePart { + offset: 0, + first: true, + last: false, + }); + ch.push(p); + assert_eq!( + ch.depth(), + 0, + "an open AU's prefix part is not a queued frame" + ); + ch.push(frame(2)); + assert_eq!(ch.depth(), 1); + } + fn popped(ch: &FrameChannel) -> Option { match ch.pop(Duration::from_millis(0)) { FramePop::Frame(f) => Some(f.frame_index), diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index ba36f9f7..59472a0e 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -354,6 +354,11 @@ impl NativeClient { // the video for a session that advertises it, so a non-rendering embedder that sets it // streams with NO visible cursor at all. `0` = today's composited behavior. client_caps: u8, + // Slice-progressive delivery opt-in: AU prefixes arrive as [`Frame`]s with + // [`crate::session::Frame::part`]` = Some` while the rest is still on the wire. Set it + // ONLY when this embedder's decode path understands parts (e.g. feeds MediaCodec with + // BUFFER_FLAG_PARTIAL_FRAME); with it false every AU arrives whole, exactly as before. + frame_parts: bool, launch: Option, // This device's display name, carried in [`crate::quic::Hello::name`]: what the host's // pending-approval list shows when an unpaired client knocks, and what its trust store @@ -438,6 +443,7 @@ impl NativeClient { preferred_codec, display_hdr, client_caps, + frame_parts, launch, name, pin, diff --git a/crates/punktfunk-core/src/client/pump/data.rs b/crates/punktfunk-core/src/client/pump/data.rs index ade6c30c..6b5527dc 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -520,7 +520,12 @@ impl DataPump { if frame.flags & FLAG_PROBE as u32 != 0 { continue; // speed-test filler, not video — measured via the counters above } - if pump_perf_on { + // A prefix part is not an AU arrival: the inter-arrival series, the OWD + // window and the clock-based staleness detector below all measure per-AU + // signals, so only the delivery that completes an AU feeds them (parts + // would bias OWD low and constantly reset the staleness run). + let is_au = frame.complete; + if pump_perf_on && is_au { let now = Instant::now(); if let Some(prev) = last_arrival.replace(now) { // 4096 ≈ 17 s at 240 fps — a stuck window can't grow it unbounded. @@ -552,7 +557,7 @@ impl DataPump { stale_since = None; standing_since = None; } else { - let lat_ns = if clock_offset_ns != 0 { + let lat_ns = if clock_offset_ns != 0 && is_au { now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128 } else { 0 @@ -576,7 +581,7 @@ impl DataPump { && lat_ns > FLUSH_LATENCY.as_nanos() as i128 { stale_since.get_or_insert_with(Instant::now); - } else { + } else if is_au { stale_since = None; } let depth = frames.depth(); diff --git a/crates/punktfunk-core/src/client/pump/handshake.rs b/crates/punktfunk-core/src/client/pump/handshake.rs index 981a48a4..c8f79b04 100644 --- a/crates/punktfunk-core/src/client/pump/handshake.rs +++ b/crates/punktfunk-core/src/client/pump/handshake.rs @@ -221,6 +221,12 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result(( session, send, diff --git a/crates/punktfunk-core/src/client/worker.rs b/crates/punktfunk-core/src/client/worker.rs index 18100b83..07b1f961 100644 --- a/crates/punktfunk-core/src/client/worker.rs +++ b/crates/punktfunk-core/src/client/worker.rs @@ -23,6 +23,10 @@ pub(crate) struct WorkerArgs { pub(crate) preferred_codec: u8, pub(crate) display_hdr: Option, pub(crate) client_caps: u8, + /// Slice-progressive delivery opt-in ([`crate::session::Session::set_deliver_frame_parts`]): + /// only an embedder whose decode path understands [`crate::session::Frame::part`] may set + /// it. Ignored on all-intra (PyroWave) sessions — their newest-wins draining needs whole AUs. + pub(crate) frame_parts: bool, pub(crate) launch: Option, /// This device's display name, sent in `Hello` (the host's approval list / trust store label). pub(crate) name: Option, diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index 4ac5dbfb..26507e62 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -5,7 +5,7 @@ use super::*; use crate::config::Config; use crate::error::Result; use crate::fec::ErasureCoder; -use crate::session::Frame; +use crate::session::{Frame, FramePart}; use crate::stats::StatsCounters; use std::collections::HashMap; use zerocopy::FromBytes; @@ -85,6 +85,11 @@ struct FrameBuf { block_count: usize, pts_ns: u64, user_flags: u32, + /// Slice-progressive delivery cursor ([`Reassembler::deliver_parts`]): count of leading + /// blocks already handed up as prefix parts... + next_part_block: u16, + /// ...and their total data shards — the AU shard offset where the next part starts. + delivered_shards: usize, /// The whole frame's data region — `total_data_shards × shard_bytes` zeroed bytes. Data /// shards are copied to their final offset on arrival; FEC reconstruction writes only the /// missing shards' ranges. On completion this Vec IS [`Frame::data`] (truncated to @@ -187,6 +192,11 @@ pub struct Reassembler { deliver_partial: bool, /// The newest such partial awaiting pickup (newest-wins: partials are a lossy byproduct). pending_partial: Option, + /// Deliver each video AU's newly-contiguous PREFIX as [`Frame`]s with + /// [`Frame::part`]`= Some` while the rest is still in flight (client opt-in — the + /// slice-progressive decode path). Works for any multi-block frame — both streamed shapes + /// and legacy uniform frames tile their blocks contiguously (`BlockState::base_shard`). + deliver_parts: bool, /// The video stream's window — its aged-out incomplete frames count into `frames_dropped` /// (the client's loss-recovery trigger). video: ReassemblyWindow, @@ -210,6 +220,7 @@ impl Reassembler { limits, deliver_partial: false, pending_partial: None, + deliver_parts: false, video: ReassemblyWindow::default(), probe: ReassemblyWindow::default(), recovery_pool: Vec::new(), @@ -225,6 +236,11 @@ impl Reassembler { } } + /// Opt into slice-progressive prefix delivery (see [`Reassembler::deliver_parts`]). + pub fn set_deliver_parts(&mut self, on: bool) { + self.deliver_parts = on; + } + /// Take the newest aged-out partial frame, if one is pending (see `set_deliver_partial`). pub fn take_partial(&mut self) -> Option { self.pending_partial.take() @@ -260,11 +276,13 @@ impl Reassembler { limits, deliver_partial, pending_partial, + deliver_parts, video, probe, recovery_pool, in_flight_bytes, } = self; + let deliver_parts = *deliver_parts; let lim = *limits; let shard_bytes = hdr.shard_bytes as usize; let data_shards = hdr.data_shards as usize; @@ -454,6 +472,8 @@ impl Reassembler { block_count, pts_ns: hdr.pts_ns, user_flags: hdr.user_flags, + next_part_block: 0, + delivered_shards: 0, buf: vec![0; buf_len], blocks: HashMap::new(), blocks_ok: 0, @@ -566,8 +586,13 @@ impl Reassembler { blocks, blocks_ok, block_count: frame_block_count, + pts_ns: frame_pts_ns, + user_flags: frame_user_flags, + next_part_block, + delivered_shards, .. } = frame; + let (frame_pts_ns, frame_user_flags) = (*frame_pts_ns, *frame_user_flags); // First packet of a block sizes its state; `data_shards` is already pinned by the // derived geometry above, but `recovery_shards` is per-block wire input (adaptive FEC @@ -713,15 +738,69 @@ impl Reassembler { ); *in_flight_bytes -= done.buf.len(); done.buf.truncate(done.frame_bytes); // trim trailing-shard zero padding + // Slice-progressive consumers already hold the delivered prefix — the completing + // packet hands up only the SUFFIX (with `last`), or the degenerate whole-AU part + // when nothing was delivered early. Probe filler stays whole either way — the + // speed test accounts AUs, not slices. + let (data, part) = if deliver_parts && !is_probe { + let lo = (done.delivered_shards * shard_bytes).min(done.frame_bytes); + let part = FramePart { + offset: lo as u32, + first: lo == 0, + last: true, + }; + if lo == 0 { + (done.buf, Some(part)) + } else { + (done.buf[lo..].to_vec(), Some(part)) + } + } else { + (done.buf, None) + }; return Ok(Some(Frame { - data: done.buf, + data, frame_index: hdr.frame_index, pts_ns: done.pts_ns, flags: done.user_flags, complete: true, + part, received_ns: 0, // stamped by Session::poll_frame at the session boundary })); } + // Slice-progressive delivery: when this packet completed a block that extends the AU's + // contiguous prefix (possibly unlocking blocks that finished out of order behind it), + // hand the newly-contiguous bytes up as ONE part. Only successfully-completed blocks + // count (`done` alone includes failed reconstructs), and the cursor stops short of the + // FINAL block — its zero-padded tail is only trimmed at completion above. + if deliver_parts && !is_probe { + let start = *delivered_shards; + while let Some(b) = blocks.get(&*next_part_block) { + if !(b.done && (b.reconstructed || b.data_received == b.data_shards)) { + break; + } + if block_count != 0 && (*next_part_block as usize) + 1 >= block_count { + break; + } + *delivered_shards = b.base_shard + b.data_shards; + *next_part_block += 1; + } + if *delivered_shards > start { + let (lo, hi) = (start * shard_bytes, *delivered_shards * shard_bytes); + return Ok(Some(Frame { + data: buf[lo..hi].to_vec(), + frame_index: hdr.frame_index, + pts_ns: frame_pts_ns, + flags: frame_user_flags, + complete: false, + part: Some(FramePart { + offset: lo as u32, + first: start == 0, + last: false, + }), + received_ns: 0, // stamped by Session::poll_frame at the session boundary + })); + } + } Ok(None) } @@ -830,6 +909,7 @@ impl ReassemblyWindow { pts_ns: f.pts_ns, flags: f.user_flags, complete: false, + part: None, received_ns: 0, // stamped by Session::poll_frame at the session boundary }); } @@ -894,6 +974,7 @@ mod reset_tests { pts_ns: 1, flags: 0, complete: false, + part: None, received_ns: 0, }); r.reset(); diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 3097eafb..b0bb3c68 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -1296,6 +1296,188 @@ fn slice_streamed_mixed_flag_packet_dropped() { assert_eq!(stats.snapshot().frames_dropped, 0, "dropped, not killed"); } +// --------------------------------------------------------------------------- +// Slice-progressive prefix delivery (Frame::part — P2c) +// --------------------------------------------------------------------------- + +/// Push packets collecting EVERY delivery (parts and completions), returning them in order. +fn push_collect( + r: &mut Reassembler, + coder: &dyn crate::fec::ErasureCoder, + stats: &StatsCounters, + delivery: &[Vec], +) -> Vec { + let mut out = Vec::new(); + for p in delivery { + if let Some(f) = r.push(p, coder, stats).unwrap() { + out.push(f); + } + } + out +} + +/// In-order slice delivery streams one part per completed block, offsets tiling exactly, the +/// final delivery carrying only the suffix with `last` + `complete`. +#[test] +fn parts_stream_in_order() { + let (pkts, src) = slice_streamed_packets(); + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts); + assert_eq!( + got.len(), + 4, + "three sentinel-block parts + the final suffix" + ); + let mut rebuilt = Vec::new(); + for (i, f) in got.iter().enumerate() { + let part = f + .part + .expect("parts mode: every delivery carries part meta"); + assert_eq!( + part.offset as usize, + rebuilt.len(), + "parts tile with no gaps" + ); + assert_eq!(part.first, i == 0); + assert_eq!(part.last, i + 1 == got.len()); + assert_eq!( + f.complete, part.last, + "complete rides exactly the last part" + ); + rebuilt.extend_from_slice(&f.data); + } + assert_eq!(rebuilt, src, "concatenated parts must be the byte-exact AU"); + assert_eq!( + stats.snapshot().frames_completed, + 0, + "the reassembler leaves the completion count to the session boundary" + ); +} + +/// A block completing BEHIND the prefix emits nothing; the block that closes the gap emits +/// ONE coalesced part spanning everything unlocked. +#[test] +fn parts_coalesce_across_reordered_blocks() { + let (pkts, src) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + // Blocks 1 and 2 fully first, then block 0, then the final block. + let mut delivery: Vec> = Vec::new(); + for want in [1u16, 2, 0] { + delivery.extend( + pkts.iter() + .filter(|p| { + let h = hdr_of(p); + h.block_count == 0 && h.block_index == want + }) + .cloned(), + ); + } + delivery.extend(pkts.iter().filter(|p| hdr_of(p).block_count != 0).cloned()); + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &delivery); + assert_eq!(got.len(), 2, "one coalesced prefix part + the final suffix"); + let p0 = got[0].part.unwrap(); + assert_eq!((p0.offset, p0.first, p0.last), (0, true, false)); + assert_eq!(got[0].data.len(), 720 + 288, "blocks 0-2 in one part"); + let p1 = got[1].part.unwrap(); + assert!(p1.last && got[1].complete); + let mut rebuilt = got[0].data.clone(); + rebuilt.extend_from_slice(&got[1].data); + assert_eq!(rebuilt, src); +} + +/// Loss inside a block delays its part until FEC reconstructs it — the part then carries the +/// recovered bytes, still byte-exact. +#[test] +fn parts_wait_for_fec_reconstruction() { + let (pkts, src) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + // Kill two data shards of block 0 — within the 50% parity budget. + let delivery: Vec> = pkts + .iter() + .filter(|p| { + let h = hdr_of(p); + !(h.block_count == 0 && h.block_index == 0 && [1, 7].contains(&h.shard_index)) + }) + .cloned() + .collect(); + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &delivery); + let mut rebuilt = Vec::new(); + for f in &got { + rebuilt.extend_from_slice(&f.data); + } + assert_eq!( + rebuilt, src, + "reconstructed prefix parts must be byte-exact" + ); + assert!(got.last().unwrap().complete); +} + +/// With parts on, a legacy single-block frame degenerates to ONE whole-AU delivery carrying +/// `{offset 0, first, last}` — the consumer's feed logic stays uniform. +#[test] +fn parts_degenerate_whole_frame() { + let cfg = e2e_config(FecScheme::Gf16, 50); + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let src: Vec = (0..40).map(|i| i as u8).collect(); + let pkts = pk.packetize(&src, 7, 0, coder.as_ref()).unwrap(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts); + assert_eq!(got.len(), 1); + let f = &got[0]; + assert_eq!( + f.part, + Some(crate::session::FramePart { + offset: 0, + first: true, + last: true + }) + ); + assert!(f.complete); + assert_eq!(f.data, src); +} + +/// Parts also flow for the LEGACY streamed shape (uniform full-K sentinels) — the prefix +/// cursor rides `base_shard`, which both wire shapes maintain. +#[test] +fn parts_flow_for_legacy_streamed_frames() { + let chunks: Vec> = (0..3) + .map(|c| (0..50).map(|i| (c * 57 + i * 131 + 7) as u8).collect()) + .collect(); + let chunk_refs: Vec<&[u8]> = chunks.iter().map(|c| c.as_slice()).collect(); + let (pkts, src) = streamed_packets(FecScheme::Gf16, 50, &chunk_refs); + let cfg = e2e_config(FecScheme::Gf16, 50); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + r.set_deliver_parts(true); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts); + assert!(got.len() > 1, "sentinel blocks must deliver early parts"); + let mut rebuilt = Vec::new(); + for f in &got { + rebuilt.extend_from_slice(&f.data); + } + assert_eq!(rebuilt, src); + assert!(got.last().unwrap().complete); +} + /// A sentinel first-packet commits a MAX-sized frame buffer, so the in-flight budget must /// bite after IN_FLIGHT_BUF_FACTOR frames — the amplification bound for one-datagram opens. #[test] diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 544c0962..fd15e672 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -20,6 +20,27 @@ use crate::stats::{Stats, StatsCounters}; use crate::transport::Transport; use zerocopy::IntoBytes; +/// Slice-progressive delivery metadata ([`Session::set_deliver_frame_parts`]): this [`Frame`] +/// carries one contiguous piece of an access unit, handed up while the rest is still on the +/// wire so a `PARTIAL_FRAME`-capable decoder can start ahead of the last packet. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct FramePart { + /// Byte offset of `data` within the whole AU. The reassembler emits parts in order with + /// no gaps — but the pre-decode hand-off may drop entries under memory pressure or a + /// jump-to-live clear, so a consumer must still verify `offset` equals its open AU's next + /// expected byte and treat a mismatch as the AU lost (abandon + flush the decoder's + /// partial input, then discard until the next `first`). The same discard applies to a + /// non-`first` part arriving with no AU open. + pub offset: u32, + /// The AU's first part. A `first` part for a NEW `frame_index` while an earlier AU is + /// still open means that AU died mid-flight (aged out or was cleared) — abandon it and + /// flush the decoder's partial input; no explicit abort part is ever sent. + pub first: bool, + /// The AU's final part — the whole AU has now been delivered ([`Frame::complete`] is set + /// on exactly this part) and the input may be submitted to the decoder as finished. + pub last: bool, +} + /// A reassembled, FEC-recovered access unit, ready to hand to the platform decoder. pub struct Frame { pub data: Vec, @@ -32,6 +53,10 @@ pub struct Frame { /// ([`crate::packet::USER_FLAG_CHUNK_ALIGNED`]) are ever delivered partial; missing /// shard ranges are zero-filled at their exact offsets. pub complete: bool, + /// `Some` = slice-progressive delivery is on and this is ONE PIECE of an AU (see + /// [`FramePart`]); `data` holds only the part's bytes. `None` = a whole AU (or an + /// aged-out chunk-aligned partial — `complete` distinguishes). + pub part: Option, /// Wall-clock instant (ns since the Unix epoch, CLOCK_REALTIME basis — the same clock the /// skew handshake compares and the host stamps `pts_ns` with) at which this AU finished /// reassembly, stamped by [`Session::poll_frame`] as the frame leaves the session. Embedders @@ -611,6 +636,16 @@ impl Session { self.reassembler.set_deliver_partial(on); } + /// Client opt-in: deliver each AU's newly-contiguous prefix as [`Frame`]s with + /// [`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. + pub fn set_deliver_frame_parts(&mut self, on: bool) { + self.reassembler.set_deliver_parts(on); + } + /// The session's negotiated wire shard payload size (bytes of AU per datagram) — /// the window size for chunk-aligned AUs (`USER_FLAG_CHUNK_ALIGNED`). pub fn shard_payload(&self) -> usize { @@ -710,7 +745,11 @@ impl Session { p.packets += 1; } if let Some(frame) = pushed { - StatsCounters::add(&self.stats.frames_completed, 1); + // A prefix part is not a completed frame — only the delivery that closes the + // AU counts, or parts would multiply the completion rate. + if frame.complete { + StatsCounters::add(&self.stats.frames_completed, 1); + } return Ok(stamp_received(frame)); } // A push that completed nothing may still have aged a partial out — deliver it diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 8bff2202..05807e7c 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -2282,17 +2282,18 @@ mod tests { mode, CompositorPref::Auto, GamepadPref::Auto, - 0, // bitrate_kbps - 0, // video_caps - 2, // audio_channels - 0, // video_codecs (HEVC-only) - 0, // preferred_codec - None, // display_hdr - 0, // client_caps - None, // launch - None, // name - None, // pin (TOFU) - None, // identity (host doesn't require pairing) + 0, // bitrate_kbps + 0, // video_caps + 2, // audio_channels + 0, // video_codecs (HEVC-only) + 0, // preferred_codec + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name + None, // pin (TOFU) + None, // identity (host doesn't require pairing) std::time::Duration::from_secs(10), ) .expect("client connects to synthetic host"); @@ -2455,15 +2456,16 @@ mod tests { CompositorPref::Auto, GamepadPref::Auto, 0, - 0, // video_caps - 2, // audio_channels (stereo) - 0, // video_codecs (0 → HEVC-only) - 0, // preferred_codec (auto) - None, // display_hdr - 0, // client_caps - None, // launch - None, // name: absent on purpose — this test asserts the fingerprint-derived label - None, // pin: TOFU — the operator's approval (not a PIN) authorizes this client + 0, // video_caps + 2, // audio_channels (stereo) + 0, // video_codecs (0 → HEVC-only) + 0, // preferred_codec (auto) + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name: absent on purpose — this test asserts the fingerprint-derived label + None, // pin: TOFU — the operator's approval (not a PIN) authorizes this client Some((cert, key)), std::time::Duration::from_secs(15), ) @@ -2524,14 +2526,15 @@ mod tests { CompositorPref::Auto, GamepadPref::Auto, 0, - 0, // video_caps - 2, // audio_channels (stereo) - 0, // video_codecs - 0, // preferred_codec - None, // display_hdr - 0, // client_caps - None, // launch - None, // name + 0, // video_caps + 2, // audio_channels (stereo) + 0, // video_codecs + 0, // preferred_codec + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name None, None, timeout @@ -2555,14 +2558,15 @@ mod tests { CompositorPref::Auto, GamepadPref::Auto, 0, - 0, // video_caps - 2, // audio_channels (stereo) - 0, // video_codecs - 0, // preferred_codec - None, // display_hdr - 0, // client_caps - None, // launch - None, // name + 0, // video_caps + 2, // audio_channels (stereo) + 0, // video_codecs + 0, // preferred_codec + None, // display_hdr + 0, // client_caps + false, // frame_parts (whole-AU delivery) + None, // launch + None, // name Some(host_fp), Some((cert.clone(), key.clone())), timeout,