perf(core): ref-based FEC encode — packetize shards reference the frame in place

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 <noreply@anthropic.com>
This commit is contained in:
2026-07-10 15:16:07 +02:00
parent 204577c7ce
commit cdbdc078d6
6 changed files with 49 additions and 28 deletions
+1 -1
View File
@@ -12,7 +12,7 @@ impl ErasureCoder for Gf16Coder {
FecScheme::Gf16 FecScheme::Gf16
} }
fn encode(&self, data: &[Vec<u8>], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> { fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> {
if recovery_count == 0 { if recovery_count == 0 {
return Ok(Vec::new()); return Ok(Vec::new());
} }
+9 -9
View File
@@ -15,7 +15,7 @@ impl ErasureCoder for Gf8Coder {
FecScheme::Gf8 FecScheme::Gf8
} }
fn encode(&self, data: &[Vec<u8>], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> { fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> {
if recovery_count == 0 { if recovery_count == 0 {
return Ok(Vec::new()); return Ok(Vec::new());
} }
@@ -24,13 +24,12 @@ impl ErasureCoder for Gf8Coder {
let shard_len = data[0].len(); let shard_len = data[0].len();
let rs = ReedSolomon::new(k, recovery_count) let rs = ReedSolomon::new(k, recovery_count)
.map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?; .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
// fec-rs fills parity in place: shards = data || zeroed parity. // `encode_sep` reads the data shards by reference and fills the parity in place —
let mut shards: Vec<Vec<u8>> = Vec::with_capacity(k + recovery_count); // same Cauchy codec as `encode`, without copying the data into a shards scratch.
shards.extend_from_slice(data); let mut parity: Vec<Vec<u8>> = (0..recovery_count).map(|_| vec![0u8; shard_len]).collect();
shards.resize_with(k + recovery_count, || vec![0u8; shard_len]); rs.encode_sep(data, &mut parity)
rs.encode(&mut shards)
.map_err(|_| FecError::Backend("gf8 encode"))?; .map_err(|_| FecError::Backend("gf8 encode"))?;
Ok(shards.split_off(k)) Ok(parity)
} }
fn reconstruct( fn reconstruct(
@@ -84,7 +83,7 @@ mod tests {
fn nanors_exact_parity_vectors() { fn nanors_exact_parity_vectors() {
let coder = Gf8Coder; let coder = Gf8Coder;
// The definitive nanors vector (k=4, m=2): single-byte shards [10,20,30,40] → [136, 0]. // 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(); let parity = coder.encode(&data, 2).unwrap();
assert_eq!(parity, vec![vec![136u8], vec![0u8]]); assert_eq!(parity, vec![vec![136u8], vec![0u8]]);
@@ -106,7 +105,8 @@ mod tests {
fn recovers_erased_data_shards() { fn recovers_erased_data_shards() {
let coder = Gf8Coder; let coder = Gf8Coder;
let data: Vec<Vec<u8>> = (0..6).map(|i| vec![i as u8; 8]).collect(); let data: Vec<Vec<u8>> = (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<Option<Vec<u8>>> = data let mut received: Vec<Option<Vec<u8>>> = data
.iter() .iter()
.cloned() .cloned()
+8 -4
View File
@@ -30,7 +30,9 @@ pub trait ErasureCoder: Send + Sync {
/// Encode `data` (K original shards) into `recovery_count` (M) parity shards. /// Encode `data` (K original shards) into `recovery_count` (M) parity shards.
/// Returns the M recovery shards. `recovery_count == 0` returns an empty `Vec`. /// Returns the M recovery shards. `recovery_count == 0` returns an empty `Vec`.
fn encode(&self, data: &[Vec<u8>], recovery_count: usize) -> Result<Vec<Vec<u8>>, 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<Vec<Vec<u8>>, FecError>;
/// Reconstruct the K original shards. `received` has length K+M: indices `0..K` are /// 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. /// 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. /// Validate `encode` inputs: at least one data shard, all of equal length.
pub(crate) fn validate_encode_shape(data: &[Vec<u8>]) -> Result<(), FecError> { pub(crate) fn validate_encode_shape(data: &[&[u8]]) -> Result<(), FecError> {
let first = data let first = data
.first() .first()
.ok_or(FecError::Config("no data shards"))? .ok_or(FecError::Config("no data shards"))?
@@ -100,7 +102,8 @@ mod tests {
let data: Vec<Vec<u8>> = (0..k) let data: Vec<Vec<u8>> = (0..k)
.map(|i| (0..shard_len).map(|b| (i * 31 + b * 7) as u8).collect()) .map(|i| (0..shard_len).map(|b| (i * 31 + b * 7) as u8).collect())
.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); assert_eq!(recovery.len(), m);
let mut received: Vec<Option<Vec<u8>>> = Vec::with_capacity(k + m); let mut received: Vec<Option<Vec<u8>>> = Vec::with_capacity(k + m);
@@ -128,7 +131,8 @@ mod tests {
#[test] #[test]
fn gf8_too_much_loss_errors() { fn gf8_too_much_loss_errors() {
let data: Vec<Vec<u8>> = (0..8).map(|_| vec![0u8; 64]).collect(); let data: Vec<Vec<u8>> = (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<Option<Vec<u8>>> = data let mut received: Vec<Option<Vec<u8>>> = data
.iter() .iter()
.cloned() .cloned()
+27 -12
View File
@@ -104,6 +104,11 @@ pub struct Packetizer {
shard_payload: usize, shard_payload: usize,
fec: crate::config::FecConfig, fec: crate::config::FecConfig,
version: u8, 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<u8>,
} }
impl Packetizer { impl Packetizer {
@@ -114,6 +119,7 @@ impl Packetizer {
shard_payload: config.shard_payload, shard_payload: config.shard_payload,
fec: config.fec, fec: config.fec,
version: config.phase as u8, 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(); let mut packets = Vec::new();
for b in 0..block_count { for b in 0..block_count {
let first = b * max_block; let first = b * max_block;
let last = ((b + 1) * max_block).min(total_data); let last = ((b + 1) * max_block).min(total_data);
let block_data_count = last - first; let block_data_count = last - first;
// Build this block's data shards (each `payload` bytes, last zero-padded). // This block's data shards: references into `frame` (plus the staged tail).
let mut data_shards: Vec<Vec<u8>> = Vec::with_capacity(block_data_count); let data_shards: Vec<&[u8]> = (first..last).map(shard_at).collect();
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);
}
let recovery_count = self.fec.recovery_for(block_data_count); let recovery_count = self.fec.recovery_for(block_data_count);
let recovery = coder.encode(&data_shards, recovery_count)?; let recovery = coder.encode(&data_shards, recovery_count)?;
@@ -186,7 +201,7 @@ impl Packetizer {
for shard_index in 0..total_shards { for shard_index in 0..total_shards {
let body: &[u8] = if shard_index < block_data_count { let body: &[u8] = if shard_index < block_data_count {
&data_shards[shard_index] data_shards[shard_index]
} else { } else {
&recovery[shard_index - block_data_count] &recovery[shard_index - block_data_count]
}; };
+2 -1
View File
@@ -205,7 +205,8 @@ proptest! {
let data: Vec<Vec<u8>> = (0..k) let data: Vec<Vec<u8>> = (0..k)
.map(|i| (0..shard_len).map(|b| (i ^ b).wrapping_add(seed as usize) as u8).collect()) .map(|i| (0..shard_len).map(|b| (i ^ b).wrapping_add(seed as usize) as u8).collect())
.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<Option<Vec<u8>>> = let mut received: Vec<Option<Vec<u8>>> =
data.iter().cloned().map(Some).chain(recovery.into_iter().map(Some)).collect(); data.iter().cloned().map(Some).chain(recovery.into_iter().map(Some)).collect();
@@ -149,7 +149,8 @@ impl VideoPacketizer {
}; };
let wire_pct = if m > 0 { (100 * m) / k } else { 0 }; let wire_pct = if m > 0 { (100 * m) / k } else { 0 };
let parity = if m > 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 { } else {
Vec::new() Vec::new()
}; };