From cdbdc078d603d8025fa2a567194bfc34b65e01b0 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 10 Jul 2026 15:16:07 +0200 Subject: [PATCH] =?UTF-8?q?perf(core):=20ref-based=20FEC=20encode=20?= =?UTF-8?q?=E2=80=94=20packetize=20shards=20reference=20the=20frame=20in?= =?UTF-8?q?=20place?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage A of the zero-copy host packetize path (networking-audit deferred plan §1): ErasureCoder::encode now takes &[&[u8]], so Packetizer::packetize builds each block's data shards as slices straight into the frame buffer instead of allocating + copying a Vec per data shard. Only the frame's final (possibly partial) shard is staged in a reusable zero-padded scratch; blocks are consecutive shard ranges, so every other shard is a full payload-sized slice. - gf8: encode_sep() over the same Cauchy codec — parity byte-identical to nanors/Moonlight (nanors_exact_parity_vectors unchanged and green) - gf16: reed_solomon_simd::encode is already generic over AsRef<[u8]> - loss-harness sweep: recovery rates identical before/after - bench pipeline (end-to-end, host+client): gf8/64K -3.0%, gf16/64K -2.2%, gf16/1M -3.4%, gf8/1M -0.7% Co-Authored-By: Claude Fable 5 --- crates/punktfunk-core/src/fec/gf16.rs | 2 +- crates/punktfunk-core/src/fec/gf8.rs | 18 ++++----- crates/punktfunk-core/src/fec/mod.rs | 12 ++++-- crates/punktfunk-core/src/packet.rs | 39 +++++++++++++------ crates/punktfunk-core/tests/loopback.rs | 3 +- crates/punktfunk-host/src/gamestream/video.rs | 3 +- 6 files changed, 49 insertions(+), 28 deletions(-) diff --git a/crates/punktfunk-core/src/fec/gf16.rs b/crates/punktfunk-core/src/fec/gf16.rs index c6d742f2..b0e0bc32 100644 --- a/crates/punktfunk-core/src/fec/gf16.rs +++ b/crates/punktfunk-core/src/fec/gf16.rs @@ -12,7 +12,7 @@ impl ErasureCoder for Gf16Coder { FecScheme::Gf16 } - fn encode(&self, data: &[Vec], recovery_count: usize) -> Result>, FecError> { + fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result>, FecError> { if recovery_count == 0 { return Ok(Vec::new()); } diff --git a/crates/punktfunk-core/src/fec/gf8.rs b/crates/punktfunk-core/src/fec/gf8.rs index 8792ea65..d8858e14 100644 --- a/crates/punktfunk-core/src/fec/gf8.rs +++ b/crates/punktfunk-core/src/fec/gf8.rs @@ -15,7 +15,7 @@ impl ErasureCoder for Gf8Coder { FecScheme::Gf8 } - fn encode(&self, data: &[Vec], recovery_count: usize) -> Result>, FecError> { + fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result>, FecError> { if recovery_count == 0 { return Ok(Vec::new()); } @@ -24,13 +24,12 @@ impl ErasureCoder for Gf8Coder { let shard_len = data[0].len(); let rs = ReedSolomon::new(k, recovery_count) .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?; - // fec-rs fills parity in place: shards = data || zeroed parity. - let mut shards: Vec> = Vec::with_capacity(k + recovery_count); - shards.extend_from_slice(data); - shards.resize_with(k + recovery_count, || vec![0u8; shard_len]); - rs.encode(&mut shards) + // `encode_sep` reads the data shards by reference and fills the parity in place — + // same Cauchy codec as `encode`, without copying the data into a shards scratch. + let mut parity: Vec> = (0..recovery_count).map(|_| vec![0u8; shard_len]).collect(); + rs.encode_sep(data, &mut parity) .map_err(|_| FecError::Backend("gf8 encode"))?; - Ok(shards.split_off(k)) + Ok(parity) } fn reconstruct( @@ -84,7 +83,7 @@ mod tests { fn nanors_exact_parity_vectors() { let coder = Gf8Coder; // The definitive nanors vector (k=4, m=2): single-byte shards [10,20,30,40] → [136, 0]. - let data = vec![vec![10u8], vec![20], vec![30], vec![40]]; + let data: [&[u8]; 4] = [&[10u8], &[20], &[30], &[40]]; let parity = coder.encode(&data, 2).unwrap(); assert_eq!(parity, vec![vec![136u8], vec![0u8]]); @@ -106,7 +105,8 @@ mod tests { fn recovers_erased_data_shards() { let coder = Gf8Coder; let data: Vec> = (0..6).map(|i| vec![i as u8; 8]).collect(); - let parity = coder.encode(&data, 3).unwrap(); + let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect(); + let parity = coder.encode(&refs, 3).unwrap(); let mut received: Vec>> = data .iter() .cloned() diff --git a/crates/punktfunk-core/src/fec/mod.rs b/crates/punktfunk-core/src/fec/mod.rs index 353d7ddf..ed7767eb 100644 --- a/crates/punktfunk-core/src/fec/mod.rs +++ b/crates/punktfunk-core/src/fec/mod.rs @@ -30,7 +30,9 @@ pub trait ErasureCoder: Send + Sync { /// Encode `data` (K original shards) into `recovery_count` (M) parity shards. /// Returns the M recovery shards. `recovery_count == 0` returns an empty `Vec`. - fn encode(&self, data: &[Vec], recovery_count: usize) -> Result>, FecError>; + /// Takes shard *references* so the packetizer can point straight into the frame + /// buffer instead of copying every data byte into per-shard `Vec`s first. + fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result>, FecError>; /// Reconstruct the K original shards. `received` has length K+M: indices `0..K` are /// originals, `K..K+M` are recovery shards; `Some` = present, `None` = lost. @@ -79,7 +81,7 @@ pub(crate) fn validate_block_shape( } /// Validate `encode` inputs: at least one data shard, all of equal length. -pub(crate) fn validate_encode_shape(data: &[Vec]) -> Result<(), FecError> { +pub(crate) fn validate_encode_shape(data: &[&[u8]]) -> Result<(), FecError> { let first = data .first() .ok_or(FecError::Config("no data shards"))? @@ -100,7 +102,8 @@ mod tests { let data: Vec> = (0..k) .map(|i| (0..shard_len).map(|b| (i * 31 + b * 7) as u8).collect()) .collect(); - let recovery = coder.encode(&data, m).unwrap(); + let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect(); + let recovery = coder.encode(&refs, m).unwrap(); assert_eq!(recovery.len(), m); let mut received: Vec>> = Vec::with_capacity(k + m); @@ -128,7 +131,8 @@ mod tests { #[test] fn gf8_too_much_loss_errors() { let data: Vec> = (0..8).map(|_| vec![0u8; 64]).collect(); - let recovery = Gf8Coder.encode(&data, 2).unwrap(); + let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect(); + let recovery = Gf8Coder.encode(&refs, 2).unwrap(); let mut received: Vec>> = data .iter() .cloned() diff --git a/crates/punktfunk-core/src/packet.rs b/crates/punktfunk-core/src/packet.rs index a2c712df..2a64bd2f 100644 --- a/crates/punktfunk-core/src/packet.rs +++ b/crates/punktfunk-core/src/packet.rs @@ -104,6 +104,11 @@ pub struct Packetizer { shard_payload: usize, fec: crate::config::FecConfig, version: u8, + /// Reusable zero-padded scratch for the frame's final data shard when the frame isn't an + /// exact `shard_payload` multiple (and for the single all-zero shard of an empty frame). + /// Every other data shard is a `shard_payload`-sized slice straight into the frame buffer — + /// blocks are consecutive shard ranges, so only the frame's last shard can be partial. + tail: Vec, } impl Packetizer { @@ -114,6 +119,7 @@ impl Packetizer { shard_payload: config.shard_payload, fec: config.fec, version: config.phase as u8, + tail: Vec::new(), } } @@ -159,23 +165,32 @@ impl Packetizer { )); } + // Stage the frame's one possibly-partial shard (the last) in the reusable + // zero-padded scratch; every full shard is referenced in place below. + let full_shards = frame.len() / payload; + self.tail.clear(); + self.tail.resize(payload, 0); + let rem = frame.len() % payload; + if rem > 0 { + self.tail[..rem].copy_from_slice(&frame[full_shards * payload..]); + } + let tail = &self.tail; + let shard_at = |s: usize| -> &[u8] { + if s < full_shards { + &frame[s * payload..(s + 1) * payload] + } else { + tail.as_slice() + } + }; + let mut packets = Vec::new(); for b in 0..block_count { let first = b * max_block; let last = ((b + 1) * max_block).min(total_data); let block_data_count = last - first; - // Build this block's data shards (each `payload` bytes, last zero-padded). - let mut data_shards: Vec> = Vec::with_capacity(block_data_count); - for s in first..last { - let start = s * payload; - let end = (start + payload).min(frame.len()); - let mut shard = vec![0u8; payload]; - if start < frame.len() { - shard[..end - start].copy_from_slice(&frame[start..end]); - } - data_shards.push(shard); - } + // This block's data shards: references into `frame` (plus the staged tail). + let data_shards: Vec<&[u8]> = (first..last).map(shard_at).collect(); let recovery_count = self.fec.recovery_for(block_data_count); let recovery = coder.encode(&data_shards, recovery_count)?; @@ -186,7 +201,7 @@ impl Packetizer { for shard_index in 0..total_shards { let body: &[u8] = if shard_index < block_data_count { - &data_shards[shard_index] + data_shards[shard_index] } else { &recovery[shard_index - block_data_count] }; diff --git a/crates/punktfunk-core/tests/loopback.rs b/crates/punktfunk-core/tests/loopback.rs index 17e9aa77..7be7e9ee 100644 --- a/crates/punktfunk-core/tests/loopback.rs +++ b/crates/punktfunk-core/tests/loopback.rs @@ -205,7 +205,8 @@ proptest! { let data: Vec> = (0..k) .map(|i| (0..shard_len).map(|b| (i ^ b).wrapping_add(seed as usize) as u8).collect()) .collect(); - let recovery = coder.encode(&data, m).unwrap(); + let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect(); + let recovery = coder.encode(&refs, m).unwrap(); let mut received: Vec>> = data.iter().cloned().map(Some).chain(recovery.into_iter().map(Some)).collect(); diff --git a/crates/punktfunk-host/src/gamestream/video.rs b/crates/punktfunk-host/src/gamestream/video.rs index 2c644631..561296e2 100644 --- a/crates/punktfunk-host/src/gamestream/video.rs +++ b/crates/punktfunk-host/src/gamestream/video.rs @@ -149,7 +149,8 @@ impl VideoPacketizer { }; let wire_pct = if m > 0 { (100 * m) / k } else { 0 }; let parity = if m > 0 { - Gf8Coder.encode(&shards, m).unwrap_or_default() + let refs: Vec<&[u8]> = shards.iter().map(|s| s.as_slice()).collect(); + Gf8Coder.encode(&refs, m).unwrap_or_default() } else { Vec::new() };