feat(core+host): LN1 phase-2 — VIDEO_CAP_STREAMED_AU streamed access units
The wire half of sub-frame slice output (latency §7 LN1, planning
design/nvenc-subframe-slice-output.md Phase 2): toward a client that
advertises the new cap bit (0x20), a chunked-poll encoder session ships each
AU's completed FEC blocks while the tail of the frame is still encoding —
the AU's last packet leaves the host as the encode finishes instead of
after it.
Wire semantics (negotiated; zero change for anyone else):
- Non-final blocks ride SENTINEL headers: block_count = 0 (a value no legacy
sender emits) + frame_bytes = 0 + exactly max_data_per_block data shards,
so the receiver's shard-offset formula needs no total.
- The final block's headers carry the real frame_bytes/block_count (+
FLAG_EOF) and RETRO-VALIDATE the whole frame: totals under which a
received sentinel block is out of range or not full-K kill the frame
wholesale (no spliced delivery) and the index can't be resurrected.
- Firewall: sentinels are bounded by the negotiated limits (full-K exactly,
never the last block the limits allow, no total to lie about); the exact
derived-geometry check runs unchanged on every non-sentinel packet and
retroactively at pinning. Sentinel opens commit a max_frame_bytes buffer,
bounded by the existing IN_FLIGHT_BUF_FACTOR budget (amplification test).
- Order-agnostic like legacy: a reversed frame (final block first) opens
legacy-shaped and still accepts its sentinels against the pinned totals.
- Small/empty streamed AUs degenerate to byte-identical legacy headers.
Host: Packetizer::{begin,push,finish}_streamed seal full-K blocks (data +
parity per block) as chunks arrive; Session::seal_streamed_* share the
pooled-wire + two-lane seal machinery via the new seal_run; the send thread
paces each flush under the frame's existing deadline (pace_sealed split out
of paced_submit) and runs the whole-AU accounting at the last chunk; the
encode pump forwards poll_chunk output as ChunkMsg when the client has the
cap AND the encoder chunks (re-queried per AU — an escalation falls back
seamlessly). Probes never run mid-AU. PUNKTFUNK_STREAMED_AU=0 = host escape
hatch. Client core ORs the cap into Hello (the shared reassembler carries
the support). Sampled first_slice_us vs encode_us PERF log measures the
overlap; the 0xCF stage-field extension stays a follow-up.
Core tests: streamed round-trip (clean/loss/reorder/duplicate, both orders),
sentinel firewall bounds, lying-final wholesale kill + no-resurrect,
open-amplification budget, header-shape pins. Gates still owed before
default-on: security review pass, loss-harness curve, GameStream smoke
(plane untouched structurally), bitrate A/B.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -47,11 +47,50 @@ pub struct Packetizer {
|
||||
/// where every packet of the block is dropped wholesale, the frame never completes, and the
|
||||
/// resulting loss pushes adaptive FEC *higher*. See the `recovery_for` clamp in `packetize_each`.
|
||||
max_total_shards: usize,
|
||||
/// The peer's per-frame block ceiling, mirroring [`ReassemblerLimits::from_config`]'s
|
||||
/// `max_blocks` — the streamed path's bound on how many sentinel blocks it may emit (a
|
||||
/// streamed AU's size isn't known up front, so this is the only pre-emission guard against
|
||||
/// producing a frame the receiver must reject).
|
||||
max_blocks: usize,
|
||||
}
|
||||
|
||||
/// One in-progress **streamed** access unit (design/nvenc-subframe-slice-output.md Phase 2):
|
||||
/// the caller feeds encoder chunks in as they exist ([`Packetizer::push_streamed`]) and every
|
||||
/// completed `max_data_per_block × shard_payload` block leaves under SENTINEL headers
|
||||
/// (`frame_bytes = 0`, `block_count = 0` — "not final yet") before the AU's total size is
|
||||
/// known; [`Packetizer::finish_streamed`] seals the tail block with the real totals and
|
||||
/// `FLAG_EOF`. Only ever sent to a peer that advertised
|
||||
/// [`crate::quic::VIDEO_CAP_STREAMED_AU`].
|
||||
pub struct StreamedAu {
|
||||
frame_index: u32,
|
||||
pts_ns: u64,
|
||||
user_flags: u32,
|
||||
/// Bytes not yet sealed into a block. Kept ≤ one block by `push_streamed` (it flushes only
|
||||
/// while STRICTLY more than a block is buffered, so the final block always has ≥ 1 byte and
|
||||
/// a sentinel block is never retroactively the frame's last).
|
||||
pending: Vec<u8>,
|
||||
/// Sentinel blocks already emitted.
|
||||
blocks_out: u16,
|
||||
/// Total AU bytes accumulated so far (`pending` included).
|
||||
total_bytes: u64,
|
||||
/// The frame's first packet (block 0, shard 0 — carries `FLAG_SOF`) has been emitted.
|
||||
opened: bool,
|
||||
}
|
||||
|
||||
impl StreamedAu {
|
||||
/// The wire frame index this AU is sealed with (the caller's RFI bookkeeping domain).
|
||||
pub fn frame_index(&self) -> u32 {
|
||||
self.frame_index
|
||||
}
|
||||
}
|
||||
|
||||
impl Packetizer {
|
||||
pub fn new(config: &Config) -> Self {
|
||||
let max_data = config.fec.max_data_per_block as usize;
|
||||
let total_data_max = config
|
||||
.max_frame_bytes
|
||||
.div_ceil(config.shard_payload.max(1))
|
||||
.max(1);
|
||||
Packetizer {
|
||||
next_frame_index: 0,
|
||||
next_probe_index: 0,
|
||||
@@ -64,6 +103,7 @@ impl Packetizer {
|
||||
// Mirrors `ReassemblerLimits::from_config` — keep the two in step.
|
||||
max_total_shards: (max_data + config.fec.recovery_for(max_data))
|
||||
.min(config.fec.scheme.max_total_shards()),
|
||||
max_blocks: total_data_max.div_ceil(max_data).max(1),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -277,4 +317,202 @@ impl Packetizer {
|
||||
self.next_seq = next_seq;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Open a **streamed** access unit (see [`StreamedAu`]). `frame_index` follows the same
|
||||
/// contract as [`packetize_each`](Self::packetize_each): `Some(i)` = the caller owns the
|
||||
/// video numbering; `None` draws from the internal counter.
|
||||
pub fn begin_streamed(
|
||||
&mut self,
|
||||
pts_ns: u64,
|
||||
user_flags: u32,
|
||||
frame_index: Option<u32>,
|
||||
) -> StreamedAu {
|
||||
let frame_index = frame_index.unwrap_or_else(|| {
|
||||
let i = self.next_frame_index;
|
||||
self.next_frame_index = i.wrapping_add(1);
|
||||
i
|
||||
});
|
||||
StreamedAu {
|
||||
frame_index,
|
||||
pts_ns,
|
||||
user_flags,
|
||||
pending: Vec::new(),
|
||||
blocks_out: 0,
|
||||
total_bytes: 0,
|
||||
opened: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Feed one encoder chunk into a streamed AU, emitting every block that COMPLETES under
|
||||
/// sentinel headers (`frame_bytes = 0`, `block_count = 0`, exactly `max_data_per_block`
|
||||
/// data shards — the receiver's offset formula needs no total). Flushes only while
|
||||
/// STRICTLY more than one block is buffered, so the frame's last block — whose header must
|
||||
/// carry the real totals — is never emitted here. Packets reach `emit` in wire order (the
|
||||
/// caller's nonce order), data then parity per block.
|
||||
pub fn push_streamed(
|
||||
&mut self,
|
||||
au: &mut StreamedAu,
|
||||
chunk: &[u8],
|
||||
coder: &dyn ErasureCoder,
|
||||
mut emit: impl FnMut(&PacketHeader, &[u8]) -> Result<()>,
|
||||
) -> Result<()> {
|
||||
au.total_bytes += chunk.len() as u64;
|
||||
au.pending.extend_from_slice(chunk);
|
||||
let block_bytes = self.fec.max_data_per_block as usize * self.shard_payload;
|
||||
while au.pending.len() > block_bytes {
|
||||
// Room for this sentinel block AND the final block after it, within the peer's
|
||||
// per-frame block ceiling and the u16 wire field.
|
||||
if au.blocks_out as usize + 2 > self.max_blocks.min(u16::MAX as usize) {
|
||||
return Err(PunktfunkError::Unsupported(
|
||||
"streamed AU exceeds the negotiated max_frame_bytes",
|
||||
));
|
||||
}
|
||||
let sof = !au.opened;
|
||||
let (bi, pts, uf) = (au.blocks_out, au.pts_ns, au.user_flags);
|
||||
let fi = au.frame_index;
|
||||
self.emit_streamed_block(
|
||||
fi,
|
||||
pts,
|
||||
uf,
|
||||
bi,
|
||||
&au.pending[..block_bytes],
|
||||
0,
|
||||
0,
|
||||
sof,
|
||||
false,
|
||||
coder,
|
||||
&mut emit,
|
||||
)?;
|
||||
au.pending.drain(..block_bytes);
|
||||
au.blocks_out += 1;
|
||||
au.opened = true;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Close a streamed AU: seal the final block — its headers carry the REAL
|
||||
/// `frame_bytes`/`block_count`, which retro-validate the whole frame at the receiver — with
|
||||
/// `FLAG_EOF` on the last emitted packet. An empty AU degenerates to today's single
|
||||
/// zero-padded-shard frame (`block_count = 1`, never a sentinel).
|
||||
pub fn finish_streamed(
|
||||
&mut self,
|
||||
au: StreamedAu,
|
||||
coder: &dyn ErasureCoder,
|
||||
mut emit: impl FnMut(&PacketHeader, &[u8]) -> Result<()>,
|
||||
) -> Result<()> {
|
||||
let frame_bytes = u32::try_from(au.total_bytes)
|
||||
.map_err(|_| PunktfunkError::Unsupported("streamed AU exceeds u32 bytes"))?;
|
||||
let block_count = au.blocks_out + 1;
|
||||
self.emit_streamed_block(
|
||||
au.frame_index,
|
||||
au.pts_ns,
|
||||
au.user_flags,
|
||||
au.blocks_out,
|
||||
&au.pending,
|
||||
frame_bytes,
|
||||
block_count,
|
||||
!au.opened,
|
||||
true,
|
||||
coder,
|
||||
&mut emit,
|
||||
)
|
||||
}
|
||||
|
||||
/// Seal ONE streamed block (data shards + its parity, in wire order). Sentinel blocks pass
|
||||
/// `frame_bytes = 0` / `block_count = 0`; the final block passes the real totals. `sof`
|
||||
/// marks the frame's very first packet, `eof` its very last.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn emit_streamed_block(
|
||||
&mut self,
|
||||
frame_index: u32,
|
||||
pts_ns: u64,
|
||||
user_flags: u32,
|
||||
block_index: u16,
|
||||
bytes: &[u8],
|
||||
frame_bytes: u32,
|
||||
block_count: u16,
|
||||
sof: bool,
|
||||
eof: bool,
|
||||
coder: &dyn ErasureCoder,
|
||||
emit: &mut impl FnMut(&PacketHeader, &[u8]) -> Result<()>,
|
||||
) -> Result<()> {
|
||||
let payload = self.shard_payload;
|
||||
if payload > u16::MAX as usize {
|
||||
return Err(PunktfunkError::InvalidArg("shard_payload exceeds u16"));
|
||||
}
|
||||
// At least one (zero-padded) data shard even for an empty final block (empty AU).
|
||||
let k = bytes.len().div_ceil(payload).max(1);
|
||||
let m = self
|
||||
.fec
|
||||
.recovery_for(k)
|
||||
.min(self.max_total_shards.saturating_sub(k));
|
||||
if k + m > u16::MAX as usize {
|
||||
return Err(PunktfunkError::Unsupported("block shard count exceeds u16"));
|
||||
}
|
||||
// Stage the one possibly-partial (or empty-frame) shard in the zero-padded scratch.
|
||||
let full_shards = bytes.len() / payload;
|
||||
self.tail.clear();
|
||||
self.tail.resize(payload, 0);
|
||||
let rem = bytes.len() % payload;
|
||||
if rem > 0 {
|
||||
self.tail[..rem].copy_from_slice(&bytes[full_shards * payload..]);
|
||||
}
|
||||
let tail = &self.tail;
|
||||
let shard_at = |s: usize| -> &[u8] {
|
||||
if s < full_shards {
|
||||
&bytes[s * payload..(s + 1) * payload]
|
||||
} else {
|
||||
tail.as_slice()
|
||||
}
|
||||
};
|
||||
let data_shards: Vec<&[u8]> = (0..k).map(shard_at).collect();
|
||||
if self.recovery.is_empty() {
|
||||
self.recovery.push(Vec::new());
|
||||
}
|
||||
coder.encode_into(&data_shards, m, &mut self.recovery[0])?;
|
||||
|
||||
let mut next_seq = self.next_seq;
|
||||
let mut emit_one = |next_seq: &mut u32, shard_index: usize, body: &[u8], flags: u8| {
|
||||
let seq = *next_seq;
|
||||
*next_seq = next_seq.wrapping_add(1);
|
||||
let hdr = PacketHeader {
|
||||
pts_ns,
|
||||
frame_index,
|
||||
stream_seq: seq,
|
||||
frame_bytes,
|
||||
user_flags,
|
||||
block_index,
|
||||
block_count,
|
||||
data_shards: k as u16,
|
||||
recovery_shards: m as u16,
|
||||
shard_index: shard_index as u16,
|
||||
shard_bytes: payload as u16,
|
||||
magic: PUNKTFUNK_MAGIC,
|
||||
version: self.version,
|
||||
fec_scheme: coder.scheme() as u8,
|
||||
flags,
|
||||
};
|
||||
emit(&hdr, body)
|
||||
};
|
||||
for (shard_index, body) in data_shards.iter().enumerate() {
|
||||
let mut flags = FLAG_PIC;
|
||||
if sof && shard_index == 0 {
|
||||
flags |= FLAG_SOF;
|
||||
}
|
||||
if eof && m == 0 && shard_index + 1 == k {
|
||||
flags |= FLAG_EOF;
|
||||
}
|
||||
emit_one(&mut next_seq, shard_index, body, flags)?;
|
||||
}
|
||||
for r in 0..m {
|
||||
let mut flags = FLAG_PIC;
|
||||
if eof && r + 1 == m {
|
||||
flags |= FLAG_EOF;
|
||||
}
|
||||
let body: &[u8] = &self.recovery[0][r];
|
||||
emit_one(&mut next_seq, k + r, body, flags)?;
|
||||
}
|
||||
self.next_seq = next_seq;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -70,7 +70,12 @@ struct BlockState {
|
||||
}
|
||||
|
||||
struct FrameBuf {
|
||||
/// Exact AU size. 0 = unknown: the frame was opened by a streamed-AU SENTINEL packet
|
||||
/// ([`crate::quic::VIDEO_CAP_STREAMED_AU`]) and the final block's real totals haven't
|
||||
/// arrived yet — the frame can't complete before they do (and retro-validate).
|
||||
frame_bytes: usize,
|
||||
/// Block count; 0 = unknown (sentinel-opened, totals not yet pinned). A legacy-opened
|
||||
/// frame always has ≥ 1 here, so 0 doubles as the "unpinned streamed" marker.
|
||||
block_count: usize,
|
||||
pts_ns: u64,
|
||||
user_flags: u32,
|
||||
@@ -277,33 +282,58 @@ impl Reassembler {
|
||||
|| total == 0
|
||||
|| total > lim.max_total_shards
|
||||
|| shard_index >= total
|
||||
|| block_count == 0
|
||||
|| block_count > lim.max_blocks
|
||||
|| hdr.block_index as usize >= block_count
|
||||
|| frame_bytes > lim.max_frame_bytes
|
||||
{
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
// Derived-geometry firewall: every sender (our Packetizer, any version) slices a frame
|
||||
// into consecutive blocks of exactly `max_data_per_block` data shards with only the LAST
|
||||
// block smaller, and stamps the exact `frame_bytes` in every header. That makes every
|
||||
// data shard's final AU offset computable on arrival —
|
||||
// offset = (block_index × max_data_per_block + shard_index) × shard_bytes
|
||||
// — which is what lets shards land straight in the frame buffer below. Enforce the
|
||||
// invariant so a header lying about its geometry is dropped instead of scribbling into
|
||||
// another shard's range.
|
||||
let total_data = frame_bytes.div_ceil(shard_bytes).max(1);
|
||||
let expect_blocks = total_data.div_ceil(lim.max_data_shards).max(1);
|
||||
// Streamed-AU sentinel ([`crate::quic::VIDEO_CAP_STREAMED_AU`]): `block_count == 0` — a
|
||||
// value no legacy sender ever emits — marks a NON-FINAL block of an AU whose total size
|
||||
// doesn't exist yet (the host is still encoding its tail). The exact derived-geometry
|
||||
// check below can't run without a total, so a sentinel is bounded by the negotiated
|
||||
// limits instead — and by construction: full-K exactly (the offset formula needs it),
|
||||
// never the last block the limits allow (the real final block must still fit after it),
|
||||
// and no total to lie about. The frame-wide exact check runs retroactively the moment
|
||||
// the final block's real totals arrive (the pinning arm below).
|
||||
let sentinel = block_count == 0;
|
||||
let block_idx = hdr.block_index as usize;
|
||||
let expect_data_shards = if block_idx + 1 == expect_blocks {
|
||||
total_data - (expect_blocks - 1) * lim.max_data_shards
|
||||
// For a sentinel-opened frame the buffer must hold ANY final geometry the totals may
|
||||
// later pin — the maximum the negotiated limits allow (the design's "allocate at
|
||||
// max_frame_bytes"; the existing in-flight budget bounds the amplification).
|
||||
let total_data_max = lim.max_frame_bytes.div_ceil(shard_bytes).max(1);
|
||||
let total_data = frame_bytes.div_ceil(shard_bytes).max(1);
|
||||
if sentinel {
|
||||
if frame_bytes != 0
|
||||
|| data_shards != lim.max_data_shards
|
||||
|| block_idx + 1 >= lim.max_blocks
|
||||
{
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
} else {
|
||||
lim.max_data_shards
|
||||
};
|
||||
if block_count != expect_blocks || data_shards != expect_data_shards {
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
if block_count > lim.max_blocks || block_idx >= block_count {
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
// Derived-geometry firewall: every sender (our Packetizer, any version) slices a
|
||||
// frame into consecutive blocks of exactly `max_data_per_block` data shards with
|
||||
// only the LAST block smaller, and stamps the exact `frame_bytes` in every
|
||||
// non-sentinel header. That makes every data shard's final AU offset computable on
|
||||
// arrival —
|
||||
// offset = (block_index × max_data_per_block + shard_index) × shard_bytes
|
||||
// — which is what lets shards land straight in the frame buffer below. Enforce the
|
||||
// invariant so a header lying about its geometry is dropped instead of scribbling
|
||||
// into another shard's range.
|
||||
let expect_blocks = total_data.div_ceil(lim.max_data_shards).max(1);
|
||||
let expect_data_shards = if block_idx + 1 == expect_blocks {
|
||||
total_data - (expect_blocks - 1) * lim.max_data_shards
|
||||
} else {
|
||||
lim.max_data_shards
|
||||
};
|
||||
if block_count != expect_blocks || data_shards != expect_data_shards {
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
let body = &pkt[HEADER_LEN..HEADER_LEN + shard_bytes];
|
||||
|
||||
@@ -350,8 +380,13 @@ impl Reassembler {
|
||||
}
|
||||
|
||||
// First packet of a frame allocates its whole (zeroed) buffer, budget-gated; later
|
||||
// packets must agree with its geometry.
|
||||
let buf_len = total_data * shard_bytes;
|
||||
// packets must agree with its geometry. A sentinel-opened (streamed) frame allocates at
|
||||
// the limits' maximum — its real size doesn't exist yet.
|
||||
let buf_len = if sentinel {
|
||||
total_data_max * shard_bytes
|
||||
} else {
|
||||
total_data * shard_bytes
|
||||
};
|
||||
let frame = match win.frames.entry(hdr.frame_index) {
|
||||
std::collections::hash_map::Entry::Occupied(e) => e.into_mut(),
|
||||
std::collections::hash_map::Entry::Vacant(e) => {
|
||||
@@ -374,7 +409,66 @@ impl Reassembler {
|
||||
})
|
||||
}
|
||||
};
|
||||
if frame.block_count != block_count || frame.frame_bytes != frame_bytes {
|
||||
if sentinel {
|
||||
// Sentinel packets carry no totals to cross-check: while the frame is unpinned
|
||||
// (`block_count == 0`) they match by construction, and once the totals are pinned —
|
||||
// whichever packet arrived first, sentinel or final (reorder is normal) — a
|
||||
// sentinel block must still be NON-final under them. Its full-K shape was already
|
||||
// enforced by the firewall, which is exactly what the pinned geometry demands of
|
||||
// every non-final block, and its shard offsets are within the pinned range by
|
||||
// `block_idx + 1 < block_count`.
|
||||
if frame.block_count != 0 && block_idx + 1 >= frame.block_count {
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
} else if frame.block_count == 0 {
|
||||
// A streamed frame meets its FINAL block's real totals: retro-validate every block
|
||||
// the sentinels created against the exact geometry these totals derive (the same
|
||||
// invariant the firewall enforces per-packet for legacy frames), then PIN them. A
|
||||
// header that lies — totals under which an already-received sentinel block is
|
||||
// out of range or not full-K — kills the WHOLE frame: its landed shards can't be
|
||||
// trusted to be at the offsets this geometry means, so delivering any of it would
|
||||
// hand the decoder spliced bytes.
|
||||
let expect_blocks = total_data.div_ceil(lim.max_data_shards).max(1);
|
||||
let final_k = total_data - (expect_blocks - 1) * lim.max_data_shards;
|
||||
let lied = frame.blocks.iter().any(|(&bi, b)| {
|
||||
let bi = bi as usize;
|
||||
bi >= expect_blocks
|
||||
|| (bi + 1 < expect_blocks && b.data_shards != lim.max_data_shards)
|
||||
|| (bi + 1 == expect_blocks && b.data_shards != final_k)
|
||||
});
|
||||
if lied {
|
||||
let mut f = win
|
||||
.frames
|
||||
.remove(&hdr.frame_index)
|
||||
.expect("frame entry exists");
|
||||
*in_flight_bytes -= f.buf.len();
|
||||
// Remember the index (with its late-shard memory, exactly like an aged-out
|
||||
// frame) so stragglers can't resurrect it, reclaim the parity buffers, and
|
||||
// count the loss — the client's recovery request is the right outcome for a
|
||||
// frame that was destroyed by a lying header.
|
||||
win.completed.insert(
|
||||
hdr.frame_index,
|
||||
reconstructed_shards(&f.blocks, lim.max_data_shards),
|
||||
);
|
||||
for block in f.blocks.values_mut() {
|
||||
for slot in block.recovery.iter_mut() {
|
||||
if let Some(rb) = slot.take() {
|
||||
if recovery_pool.len() < RECOVERY_POOL_MAX {
|
||||
recovery_pool.push(rb);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if !is_probe {
|
||||
StatsCounters::add(&stats.frames_dropped, 1);
|
||||
}
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
frame.frame_bytes = frame_bytes;
|
||||
frame.block_count = block_count;
|
||||
} else if frame.block_count != block_count || frame.frame_bytes != frame_bytes {
|
||||
drop(stats);
|
||||
return Ok(None);
|
||||
}
|
||||
@@ -382,6 +476,7 @@ impl Reassembler {
|
||||
buf,
|
||||
blocks,
|
||||
blocks_ok,
|
||||
block_count: frame_block_count,
|
||||
..
|
||||
} = frame;
|
||||
|
||||
@@ -485,8 +580,12 @@ impl Reassembler {
|
||||
}
|
||||
}
|
||||
|
||||
// Whole frame ready?
|
||||
if *blocks_ok == block_count {
|
||||
// Whole frame ready? Judged against the FRAME's pinned block count, not this packet's
|
||||
// header — a streamed frame can complete on a reordered sentinel packet (header says 0)
|
||||
// after the final block already pinned the real totals, and can never complete before
|
||||
// they're pinned (`0` never equals a non-zero `blocks_ok`).
|
||||
let block_count = *frame_block_count;
|
||||
if block_count != 0 && *blocks_ok == block_count {
|
||||
let mut done = win.frames.remove(&hdr.frame_index).unwrap();
|
||||
win.completed.insert(
|
||||
hdr.frame_index,
|
||||
|
||||
@@ -640,3 +640,294 @@ fn adaptive_fec_ramp_keeps_maximal_blocks_within_the_peers_ceiling() {
|
||||
src
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Streamed access units (VIDEO_CAP_STREAMED_AU — nvenc-subframe-slice-output.md Phase 2)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Packetize one streamed AU from `chunks` via begin/push/finish, returning the emitted wire
|
||||
/// packets (header ++ shard) and the concatenated source bytes.
|
||||
fn streamed_packets(
|
||||
scheme: FecScheme,
|
||||
fec_percent: u8,
|
||||
chunks: &[&[u8]],
|
||||
) -> (Vec<Vec<u8>>, Vec<u8>) {
|
||||
let cfg = e2e_config(scheme, fec_percent);
|
||||
let coder = coder_for(scheme);
|
||||
let mut pk = Packetizer::new(&cfg);
|
||||
let mut au = pk.begin_streamed(12345, 0, Some(0));
|
||||
let mut pkts: Vec<Vec<u8>> = Vec::new();
|
||||
let mut src = Vec::new();
|
||||
for c in chunks {
|
||||
src.extend_from_slice(c);
|
||||
pk.push_streamed(&mut au, c, coder.as_ref(), |h: &PacketHeader, b: &[u8]| {
|
||||
let mut p = Vec::with_capacity(HEADER_LEN + b.len());
|
||||
p.extend_from_slice(h.as_bytes());
|
||||
p.extend_from_slice(b);
|
||||
pkts.push(p);
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
pk.finish_streamed(au, coder.as_ref(), |h: &PacketHeader, b: &[u8]| {
|
||||
let mut p = Vec::with_capacity(HEADER_LEN + b.len());
|
||||
p.extend_from_slice(h.as_bytes());
|
||||
p.extend_from_slice(b);
|
||||
pkts.push(p);
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
(pkts, src)
|
||||
}
|
||||
|
||||
/// Deliver a streamed AU's packets (optionally with kills within the FEC budget, reversed
|
||||
/// order, and a duplicate) and assert byte-identical completion. Reversed order is the
|
||||
/// critical case: the FINAL block's real-total headers arrive FIRST, the frame opens
|
||||
/// legacy-shaped, and the sentinels must still be accepted against the pinned totals.
|
||||
fn streamed_roundtrip(scheme: FecScheme, kill: &[usize], reverse: bool) {
|
||||
let chunks: Vec<Vec<u8>> = (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(scheme, 50, &chunk_refs);
|
||||
// 150 B / 16 B shards / 4-shard blocks → sentinel blocks 0,1 (4 data + 2 rec each) +
|
||||
// final block (2 data + 1 rec) = 15 packets.
|
||||
assert_eq!(
|
||||
pkts.len(),
|
||||
15,
|
||||
"expected geometry changed — update the kills"
|
||||
);
|
||||
|
||||
let mut delivery: Vec<Vec<u8>> = pkts
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(i, _)| !kill.contains(i))
|
||||
.map(|(_, p)| p.clone())
|
||||
.collect();
|
||||
if reverse {
|
||||
delivery.reverse();
|
||||
}
|
||||
if let Some(dup) = delivery.first().cloned() {
|
||||
delivery.push(dup);
|
||||
}
|
||||
|
||||
let cfg = e2e_config(scheme, 50);
|
||||
let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg));
|
||||
let coder = coder_for(scheme);
|
||||
let stats = StatsCounters::default();
|
||||
let mut got = None;
|
||||
for p in &delivery {
|
||||
if let Some(f) = r.push(p, coder.as_ref(), &stats).unwrap() {
|
||||
assert!(got.is_none(), "frame must complete exactly once");
|
||||
got = Some(f);
|
||||
}
|
||||
}
|
||||
let f = got.expect("streamed frame must complete within the FEC budget");
|
||||
assert_eq!(
|
||||
f.data, src,
|
||||
"reassembled streamed AU must be byte-identical"
|
||||
);
|
||||
assert_eq!(f.pts_ns, 12345);
|
||||
assert!(f.complete);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn streamed_roundtrip_clean_and_reversed() {
|
||||
streamed_roundtrip(FecScheme::Gf16, &[], false);
|
||||
streamed_roundtrip(FecScheme::Gf16, &[], true);
|
||||
streamed_roundtrip(FecScheme::Gf8, &[], true);
|
||||
}
|
||||
|
||||
/// Loss within each block's FEC budget: one data shard from a sentinel block, one from the
|
||||
/// final block — in both delivery orders.
|
||||
#[test]
|
||||
fn streamed_roundtrip_survives_loss_and_reorder() {
|
||||
// Wire order: blk0 = 0..4 data + 4..6 rec, blk1 = 6..10 data + 10..12 rec,
|
||||
// final = 12..14 data + 14 rec.
|
||||
streamed_roundtrip(FecScheme::Gf16, &[1, 12], false);
|
||||
streamed_roundtrip(FecScheme::Gf16, &[1, 12], true);
|
||||
}
|
||||
|
||||
/// The wire shape of a streamed AU: sentinel headers (block_count = 0, frame_bytes = 0,
|
||||
/// full-K) on every non-final block, real totals + EOF on the final block, SOF on the very
|
||||
/// first packet only.
|
||||
#[test]
|
||||
fn streamed_headers_sentinel_then_final() {
|
||||
let chunks: Vec<Vec<u8>> = (0..3).map(|_| vec![0xA5u8; 50]).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 mut saw_final = false;
|
||||
for (i, p) in pkts.iter().enumerate() {
|
||||
let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap();
|
||||
assert_eq!(
|
||||
h.flags & FLAG_SOF != 0,
|
||||
i == 0,
|
||||
"SOF exactly on the first packet"
|
||||
);
|
||||
assert_eq!(
|
||||
h.flags & FLAG_EOF != 0,
|
||||
i + 1 == pkts.len(),
|
||||
"EOF exactly on the last packet"
|
||||
);
|
||||
if h.block_index < 2 {
|
||||
assert_eq!(
|
||||
h.block_count, 0,
|
||||
"non-final block must ride sentinel headers"
|
||||
);
|
||||
assert_eq!(h.frame_bytes, 0);
|
||||
assert_eq!(h.data_shards, 4, "sentinel blocks are exactly full-K");
|
||||
} else {
|
||||
saw_final = true;
|
||||
assert_eq!(h.block_count, 3, "final block carries the real block count");
|
||||
assert_eq!(h.frame_bytes as usize, src.len(), "and the real AU size");
|
||||
}
|
||||
}
|
||||
assert!(saw_final);
|
||||
}
|
||||
|
||||
/// A streamed AU smaller than one block emits NO sentinels — its single (final) block is
|
||||
/// byte-identical in shape to a legacy frame, so small frames pay zero streaming overhead
|
||||
/// and any receiver accepts them.
|
||||
#[test]
|
||||
fn streamed_small_frame_degenerates_to_legacy() {
|
||||
let (pkts, src) = streamed_packets(FecScheme::Gf16, 50, &[&[0x5Au8; 40]]);
|
||||
for p in &pkts {
|
||||
let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap();
|
||||
assert_eq!(
|
||||
h.block_count, 1,
|
||||
"single-block streamed AU must be legacy-shaped"
|
||||
);
|
||||
assert_eq!(h.frame_bytes as usize, src.len());
|
||||
}
|
||||
}
|
||||
|
||||
/// Sentinel firewall: a sentinel header that is not exactly full-K, or claims a non-zero
|
||||
/// total, or sits where the final block could no longer follow, is dropped before any
|
||||
/// allocation happens.
|
||||
#[test]
|
||||
fn streamed_sentinel_firewall_bounds() {
|
||||
let mut r = Reassembler::new(limits());
|
||||
let coder = coder_for(FecScheme::Gf8);
|
||||
let stats = StatsCounters::default();
|
||||
let sentinel = |f: fn(&mut PacketHeader)| {
|
||||
let mut h = base_header();
|
||||
h.block_count = 0;
|
||||
h.frame_bytes = 0;
|
||||
h.data_shards = 8; // limits().max_data_shards — the only legal sentinel K
|
||||
h.recovery_shards = 0;
|
||||
f(&mut h);
|
||||
h
|
||||
};
|
||||
// Not full-K.
|
||||
let h = sentinel(|h| h.data_shards = 7);
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
// Claims a total.
|
||||
let h = sentinel(|h| h.frame_bytes = 64);
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
// Sits on the last block the limits allow (no room for the final block after it).
|
||||
let h = sentinel(|h| h.block_index = 3); // limits().max_blocks == 4
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
assert_eq!(stats.snapshot().packets_dropped, 3);
|
||||
// A conformant sentinel IS accepted (proves the rejections above weren't vacuous).
|
||||
let h = sentinel(|_| {});
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
assert_eq!(
|
||||
stats.snapshot().packets_dropped,
|
||||
3,
|
||||
"conformant sentinel accepted"
|
||||
);
|
||||
}
|
||||
|
||||
/// Retro-validation: final-block totals under which an already-received sentinel block is
|
||||
/// out of range (or mis-sized) kill the WHOLE frame — no spliced delivery — and the killed
|
||||
/// index cannot be resurrected by stragglers.
|
||||
#[test]
|
||||
fn streamed_lying_final_totals_kill_the_frame_wholesale() {
|
||||
let mut r = Reassembler::new(limits());
|
||||
let coder = coder_for(FecScheme::Gf8);
|
||||
let stats = StatsCounters::default();
|
||||
// Two sentinel blocks (indexes 0 and 1) open the frame and land shards.
|
||||
for bi in 0..2u16 {
|
||||
let mut h = base_header();
|
||||
h.block_count = 0;
|
||||
h.frame_bytes = 0;
|
||||
h.data_shards = 8;
|
||||
h.recovery_shards = 0;
|
||||
h.block_index = bi;
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
}
|
||||
// A "final" header claiming the whole AU is ONE 16-byte shard: geometry-valid on its own
|
||||
// (expect_blocks = 1, K = 1), but it disowns both sentinel blocks → the frame dies.
|
||||
let mut lying = base_header();
|
||||
lying.block_count = 1;
|
||||
lying.frame_bytes = 16;
|
||||
lying.data_shards = 1;
|
||||
lying.recovery_shards = 0;
|
||||
assert!(r
|
||||
.push(&packet(lying), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
let snap = stats.snapshot();
|
||||
assert_eq!(
|
||||
snap.frames_dropped, 1,
|
||||
"the lying frame must be counted lost"
|
||||
);
|
||||
// A straggler sentinel for the killed index must not resurrect it.
|
||||
let mut h = base_header();
|
||||
h.block_count = 0;
|
||||
h.frame_bytes = 0;
|
||||
h.data_shards = 8;
|
||||
h.recovery_shards = 0;
|
||||
let before = stats.snapshot().packets_dropped;
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
assert_eq!(
|
||||
stats.snapshot().packets_dropped,
|
||||
before + 1,
|
||||
"straggler for a killed frame must be dropped, not re-open it"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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]
|
||||
fn streamed_open_amplification_is_budget_bounded() {
|
||||
let mut r = Reassembler::new(limits());
|
||||
let coder = coder_for(FecScheme::Gf8);
|
||||
let stats = StatsCounters::default();
|
||||
// limits(): max_frame_bytes 4096 → each sentinel open commits 4096 B; budget = 4×4096.
|
||||
for fi in 0..5u32 {
|
||||
let mut h = base_header();
|
||||
h.block_count = 0;
|
||||
h.frame_bytes = 0;
|
||||
h.data_shards = 8;
|
||||
h.recovery_shards = 0;
|
||||
h.frame_index = fi;
|
||||
assert!(r
|
||||
.push(&packet(h), coder.as_ref(), &stats)
|
||||
.unwrap()
|
||||
.is_none());
|
||||
}
|
||||
assert_eq!(
|
||||
stats.snapshot().packets_dropped,
|
||||
1,
|
||||
"the fifth max-sized open must be refused by the in-flight budget"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user