From f4f6c5556f1a2e8e04333b286eade3e1a1a8ede7 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Tue, 14 Jul 2026 23:19:21 +0200 Subject: [PATCH] =?UTF-8?q?perf(core):=20FEC=20encoder=20reuse=20=E2=80=94?= =?UTF-8?q?=20cached=20codecs=20+=20pooled=20parity,=20no=20per-block=20se?= =?UTF-8?q?tup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1.4 (throughput-beyond-1gbps.md): the send path built a fresh erasure codec and allocated fresh parity Vecs for every FEC block. New trait method ErasureCoder::encode_into generates parity into caller-pooled buffers; the packetizer keeps one parity pool that grows once to the session's high-water recovery count. - gf16: one cached reed_solomon_simd::ReedSolomonEncoder per coder, re-shaped per block via reset() (reuses its working space) — the old encode() convenience call paid engine CPU-feature detection, FFT planning, and work-buffer allocation per block. - gf8: last-used (k, m) Cauchy codec cached, so the generator-matrix build drops out of steady-state frames; parity buffers shaped without re-zeroing (encode_sep's first-input pass overwrites every row). The GameStream VideoPacketizer now owns a persistent coder so the cache survives frames. - encode() delegates to encode_into — one code path, and the nanors byte-exact parity vector keeps pinning Moonlight wire compatibility. Validated: 145 core + 308 host tests + clippy -D warnings on .21, loss-harness recovery curve identical, pipeline bench +0.6-2.4% thrpt (all configs, p<0.05; the loopback bench is encoder-dominated so the alloc savings mostly land outside it). Co-Authored-By: Claude Fable 5 --- crates/punktfunk-core/src/fec/gf16.rs | 67 +++++++++++++++++-- crates/punktfunk-core/src/fec/gf8.rs | 52 +++++++++++--- crates/punktfunk-core/src/fec/mod.rs | 57 ++++++++++------ crates/punktfunk-core/src/packet.rs | 9 ++- crates/punktfunk-host/src/gamestream/video.rs | 10 ++- 5 files changed, 158 insertions(+), 37 deletions(-) diff --git a/crates/punktfunk-core/src/fec/gf16.rs b/crates/punktfunk-core/src/fec/gf16.rs index 84cbe471..e67ab210 100644 --- a/crates/punktfunk-core/src/fec/gf16.rs +++ b/crates/punktfunk-core/src/fec/gf16.rs @@ -6,8 +6,19 @@ use super::{ validate_block_shape, validate_encode_shape, validate_into_shape, ErasureCoder, FecError, }; use crate::config::FecScheme; +use reed_solomon_simd::ReedSolomonEncoder; +use std::sync::Mutex; -pub struct Gf16Coder; +#[derive(Default)] +pub struct Gf16Coder { + /// Cached Leopard encoder (plan Phase 1.4): `reset()` re-shapes it per block while + /// reusing its working space, so steady-state frames cost no encoder construction (the + /// old `reed_solomon_simd::encode` convenience call built one — engine CPU-feature + /// detection, FFT planning, work-buffer allocs — per block). `Mutex` only to keep the + /// `&self` trait surface; a session's coder is driven by its one send thread, so the + /// lock is uncontended. + enc: Mutex>, +} impl ErasureCoder for Gf16Coder { fn scheme(&self) -> FecScheme { @@ -15,16 +26,62 @@ impl ErasureCoder for Gf16Coder { } fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result>, FecError> { + let mut out = Vec::new(); + self.encode_into(data, recovery_count, &mut out)?; + Ok(out) + } + + fn encode_into( + &self, + data: &[&[u8]], + recovery_count: usize, + out: &mut Vec>, + ) -> Result<(), FecError> { if recovery_count == 0 { - return Ok(Vec::new()); + out.clear(); + return Ok(()); } validate_encode_shape(data)?; let k = data.len(); - if data[0].len() % 2 != 0 { + let shard_len = data[0].len(); + if shard_len % 2 != 0 { return Err(FecError::Config("GF(2^16) shard length must be even")); } - reed_solomon_simd::encode(k, recovery_count, data) - .map_err(|_| FecError::Backend("gf16 encode")) + let mut guard = self.enc.lock().unwrap_or_else(|p| p.into_inner()); + let enc = match guard.as_mut() { + Some(enc) => { + enc.reset(k, recovery_count, shard_len) + .map_err(|_| FecError::Backend("gf16 encoder reset"))?; + enc + } + None => guard.insert( + ReedSolomonEncoder::new(k, recovery_count, shard_len) + .map_err(|_| FecError::Backend("gf16 encoder init"))?, + ), + }; + for shard in data { + enc.add_original_shard(shard) + .map_err(|_| FecError::Backend("gf16 add shard"))?; + } + let result = enc.encode().map_err(|_| FecError::Backend("gf16 encode"))?; + // Copy the parity into the caller's pooled buffers: existing `Vec`s are reused + // (clear keeps capacity), the pool grows once to the session's high-water M. + out.truncate(recovery_count); + let mut parity = result.recovery_iter(); + for buf in out.iter_mut() { + let shard = parity + .next() + .ok_or(FecError::Backend("gf16 parity count"))?; + buf.clear(); + buf.extend_from_slice(shard); + } + for shard in parity { + out.push(shard.to_vec()); + } + if out.len() != recovery_count { + return Err(FecError::Backend("gf16 parity count")); + } + Ok(()) } fn reconstruct( diff --git a/crates/punktfunk-core/src/fec/gf8.rs b/crates/punktfunk-core/src/fec/gf8.rs index 755beaa7..ee22c2fd 100644 --- a/crates/punktfunk-core/src/fec/gf8.rs +++ b/crates/punktfunk-core/src/fec/gf8.rs @@ -9,8 +9,16 @@ use super::{ }; use crate::config::FecScheme; use fec_rs::ReedSolomon; +use std::sync::Mutex; -pub struct Gf8Coder; +#[derive(Default)] +pub struct Gf8Coder { + /// Last-used Cauchy codec, keyed by its `(k, m)` shape (plan Phase 1.4): video blocks + /// keep one shape for long stretches (it only moves with frame size / adaptive-FEC + /// steps), so caching the matrix kills the per-block generator construction. `Mutex` + /// only to keep the `&self` trait surface; uncontended on the one send thread. + rs: Mutex>, +} impl ErasureCoder for Gf8Coder { fn scheme(&self) -> FecScheme { @@ -18,20 +26,46 @@ impl ErasureCoder for Gf8Coder { } fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result>, FecError> { + let mut out = Vec::new(); + self.encode_into(data, recovery_count, &mut out)?; + Ok(out) + } + + fn encode_into( + &self, + data: &[&[u8]], + recovery_count: usize, + out: &mut Vec>, + ) -> Result<(), FecError> { if recovery_count == 0 { - return Ok(Vec::new()); + out.clear(); + return Ok(()); } validate_encode_shape(data)?; let k = data.len(); let shard_len = data[0].len(); - let rs = ReedSolomon::new(k, recovery_count) - .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?; + let mut guard = self.rs.lock().unwrap_or_else(|p| p.into_inner()); + let cached = matches!(&*guard, Some((ck, cm, _)) if *ck == k && *cm == recovery_count); + if !cached { + let rs = ReedSolomon::new(k, recovery_count) + .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?; + *guard = Some((k, recovery_count, rs)); + } + let rs = &guard.as_ref().expect("cache populated above").2; + // Shape the caller's pooled parity buffers without zero-filling: `encode_sep`'s + // first-input pass overwrites every parity row, so stale bytes never survive. + out.truncate(recovery_count); + for buf in out.iter_mut() { + buf.resize(shard_len, 0); + } + while out.len() < recovery_count { + out.push(vec![0u8; shard_len]); + } // `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) + rs.encode_sep(data, out) .map_err(|_| FecError::Backend("gf8 encode"))?; - Ok(parity) + Ok(()) } fn reconstruct( @@ -121,7 +155,7 @@ mod tests { /// these vectors would break and our parity would no longer be Moonlight-decodable. #[test] fn nanors_exact_parity_vectors() { - let coder = Gf8Coder; + let coder = Gf8Coder::default(); // The definitive nanors vector (k=4, m=2): single-byte shards [10,20,30,40] → [136, 0]. let data: [&[u8]; 4] = [&[10u8], &[20], &[30], &[40]]; let parity = coder.encode(&data, 2).unwrap(); @@ -143,7 +177,7 @@ mod tests { /// Round-trip: erase `m` data shards and confirm reconstruction recovers the originals. #[test] fn recovers_erased_data_shards() { - let coder = Gf8Coder; + let coder = Gf8Coder::default(); let data: Vec> = (0..6).map(|i| vec![i as u8; 8]).collect(); let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect(); let parity = coder.encode(&refs, 3).unwrap(); diff --git a/crates/punktfunk-core/src/fec/mod.rs b/crates/punktfunk-core/src/fec/mod.rs index 39922933..e3c86eb5 100644 --- a/crates/punktfunk-core/src/fec/mod.rs +++ b/crates/punktfunk-core/src/fec/mod.rs @@ -34,6 +34,23 @@ pub trait ErasureCoder: Send + Sync { /// buffer instead of copying every data byte into per-shard `Vec`s first. fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result>, FecError>; + /// [`encode`](Self::encode) into caller-pooled parity buffers: on success `out` holds + /// exactly `recovery_count` shards, reusing its existing `Vec` allocations (extras are + /// truncated away, missing ones are grown once to the high-water mark). The per-frame + /// hot path (plan Phase 1.4) — backends also reuse their internal codec state here, so + /// steady-state frames cost no encoder construction and no parity allocations. The + /// default delegates to `encode` (correct, unpooled) for backends without an override. + /// On error `out`'s contents are unspecified and must not be sent. + fn encode_into( + &self, + data: &[&[u8]], + recovery_count: usize, + out: &mut Vec>, + ) -> Result<(), FecError> { + *out = self.encode(data, recovery_count)?; + Ok(()) + } + /// 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. /// Returns the K original shards in order. @@ -67,8 +84,8 @@ pub trait ErasureCoder: Send + Sync { /// Construct the coder for a scheme. pub fn coder_for(scheme: FecScheme) -> Box { match scheme { - FecScheme::Gf8 => Box::new(Gf8Coder), - FecScheme::Gf16 => Box::new(Gf16Coder), + FecScheme::Gf8 => Box::new(Gf8Coder::default()), + FecScheme::Gf16 => Box::new(Gf16Coder::default()), } } @@ -221,15 +238,15 @@ mod tests { #[test] fn gf16_reconstruct_into_fills_only_the_holes() { - roundtrip_into(&Gf16Coder, 16, 4, 256, &[1, 9], &[3]); - roundtrip_into(&Gf16Coder, 4, 2, 16, &[0, 3], &[]); - roundtrip_into(&Gf16Coder, 4, 2, 16, &[], &[0, 1]); // nothing missing, no parity needed + roundtrip_into(&Gf16Coder::default(), 16, 4, 256, &[1, 9], &[3]); + roundtrip_into(&Gf16Coder::default(), 4, 2, 16, &[0, 3], &[]); + roundtrip_into(&Gf16Coder::default(), 4, 2, 16, &[], &[0, 1]); // nothing missing, no parity needed } #[test] fn gf8_reconstruct_into_fills_only_the_holes() { - roundtrip_into(&Gf8Coder, 16, 4, 256, &[0, 7], &[1]); - roundtrip_into(&Gf8Coder, 4, 2, 16, &[2], &[1]); + roundtrip_into(&Gf8Coder::default(), 16, 4, 256, &[0, 7], &[1]); + roundtrip_into(&Gf8Coder::default(), 4, 2, 16, &[2], &[1]); } #[test] @@ -238,24 +255,24 @@ mod tests { // Too few shards: 2 of 4 data present, no recovery. let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect(); let have = [true, true, false, false]; - assert!(Gf16Coder + assert!(Gf16Coder::default() .reconstruct_into(2, &mut slots, &have, &[]) .is_err()); // Recovery index out of the declared range. let parity = [0u8; 8]; let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect(); - assert!(Gf16Coder + assert!(Gf16Coder::default() .reconstruct_into(2, &mut slots, &have, &[(2, &parity), (3, &parity)]) .is_err()); // Mismatched recovery shard length. let short = [0u8; 6]; let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect(); - assert!(Gf8Coder + assert!(Gf8Coder::default() .reconstruct_into(2, &mut slots, &have, &[(0, &short), (1, &parity)]) .is_err()); // `have` length disagreeing with `data`. let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect(); - assert!(Gf8Coder + assert!(Gf8Coder::default() .reconstruct_into(2, &mut slots, &[true; 3], &[(0, &parity)]) .is_err()); } @@ -263,19 +280,19 @@ mod tests { #[test] fn gf8_recovers_within_budget() { // 16 data + 4 recovery; lose 2 data + 2 recovery (== budget). - roundtrip(&Gf8Coder, 16, 4, 256, &[0, 7, 16, 19]); + roundtrip(&Gf8Coder::default(), 16, 4, 256, &[0, 7, 16, 19]); } #[test] fn gf16_recovers_within_budget() { - roundtrip(&Gf16Coder, 16, 4, 256, &[1, 9, 17, 18]); + roundtrip(&Gf16Coder::default(), 16, 4, 256, &[1, 9, 17, 18]); } #[test] fn gf8_too_much_loss_errors() { let data: Vec> = (0..8).map(|_| vec![0u8; 64]).collect(); let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect(); - let recovery = Gf8Coder.encode(&refs, 2).unwrap(); + let recovery = Gf8Coder::default().encode(&refs, 2).unwrap(); let mut received: Vec>> = data .iter() .cloned() @@ -286,8 +303,8 @@ mod tests { received[0] = None; received[1] = None; received[2] = None; - assert!(Gf16Coder.scheme() == FecScheme::Gf16); - let err = Gf8Coder.reconstruct(8, 2, &mut received); + assert!(Gf16Coder::default().scheme() == FecScheme::Gf16); + let err = Gf8Coder::default().reconstruct(8, 2, &mut received); assert!(err.is_err()); } @@ -296,9 +313,9 @@ mod tests { // data=2, recovery=2 expects a 4-element slice; a 3-element one must error, not // panic on the recovery-slice index (both backends). let mut recv: Vec>> = vec![Some(vec![0u8; 8]), None, Some(vec![0u8; 8])]; - assert!(Gf16Coder.reconstruct(2, 2, &mut recv).is_err()); + assert!(Gf16Coder::default().reconstruct(2, 2, &mut recv).is_err()); let mut recv: Vec>> = vec![Some(vec![0u8; 8]), None, Some(vec![0u8; 8])]; - assert!(Gf8Coder.reconstruct(2, 2, &mut recv).is_err()); + assert!(Gf8Coder::default().reconstruct(2, 2, &mut recv).is_err()); } #[test] @@ -306,9 +323,9 @@ mod tests { // The GF16 fast path used to clone shards verbatim without a length check. let mut recv: Vec>> = vec![Some(vec![0u8; 8]), Some(vec![0u8; 6]), None, None]; - assert!(Gf16Coder.reconstruct(2, 2, &mut recv).is_err()); + assert!(Gf16Coder::default().reconstruct(2, 2, &mut recv).is_err()); let mut recv: Vec>> = vec![Some(vec![0u8; 8]), Some(vec![0u8; 6]), None, None]; - assert!(Gf8Coder.reconstruct(2, 2, &mut recv).is_err()); + assert!(Gf8Coder::default().reconstruct(2, 2, &mut recv).is_err()); } } diff --git a/crates/punktfunk-core/src/packet.rs b/crates/punktfunk-core/src/packet.rs index 9386a246..fb23e505 100644 --- a/crates/punktfunk-core/src/packet.rs +++ b/crates/punktfunk-core/src/packet.rs @@ -147,6 +147,10 @@ pub struct Packetizer { /// 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, + /// Reusable parity buffers for [`ErasureCoder::encode_into`] (plan Phase 1.4): grows once + /// to the session's high-water recovery count, then every block's parity is generated + /// into it with zero allocations. + recovery: Vec>, } impl Packetizer { @@ -159,6 +163,7 @@ impl Packetizer { fec: config.fec, version: config.phase as u8, tail: Vec::new(), + recovery: Vec::new(), } } @@ -262,6 +267,7 @@ impl Packetizer { self.tail[..rem].copy_from_slice(&frame[full_shards * payload..]); } let tail = &self.tail; + let recovery_pool = &mut self.recovery; let shard_at = |s: usize| -> &[u8] { if s < full_shards { &frame[s * payload..(s + 1) * payload] @@ -279,7 +285,8 @@ impl Packetizer { 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)?; + coder.encode_into(&data_shards, recovery_count, recovery_pool)?; + let recovery = &*recovery_pool; let total_shards = block_data_count + recovery_count; if total_shards > u16::MAX as usize { return Err(PunktfunkError::Unsupported("block shard count exceeds u16")); diff --git a/crates/punktfunk-host/src/gamestream/video.rs b/crates/punktfunk-host/src/gamestream/video.rs index bb917cb3..6acbe2db 100644 --- a/crates/punktfunk-host/src/gamestream/video.rs +++ b/crates/punktfunk-host/src/gamestream/video.rs @@ -49,6 +49,9 @@ pub struct VideoPacketizer { frame_index: u32, /// Monotonic per-stream packet counter (the RTP sequence / streamPacketIndex source). seq: u32, + /// Persistent GF(2⁸) coder so its `(k, m)` Cauchy-matrix cache survives across frames + /// (plan Phase 1.4) — a stream's block shape only moves with frame size. + coder: Gf8Coder, } impl VideoPacketizer { @@ -65,6 +68,7 @@ impl VideoPacketizer { min_fec: min_fec as usize, frame_index: 0, seq: 0, + coder: Gf8Coder::default(), } } @@ -158,7 +162,7 @@ impl VideoPacketizer { let wire_pct = if m > 0 { (100 * m) / k } else { 0 }; let parity = if m > 0 { let refs: Vec<&[u8]> = shards.iter().map(|s| s.as_slice()).collect(); - Gf8Coder.encode(&refs, m).unwrap_or_default() + self.coder.encode(&refs, m).unwrap_or_default() } else { Vec::new() }; @@ -328,7 +332,9 @@ mod tests { // Drop data shard 1; reconstruct from the rest via the same Cauchy coder. let mut received: Vec>> = pkts.iter().map(|p| Some(p.clone())).collect(); received[1] = None; - let recovered = Gf8Coder.reconstruct(k, m, &mut received).unwrap(); + let recovered = Gf8Coder::default() + .reconstruct(k, m, &mut received) + .unwrap(); // The recovered shard equals the original data shard's RS-covered bytes: its flags // byte (offset 24) is PIC (middle shard), proving the NV header recovers correctly. assert_eq!(recovered[1][24], FLAG_PIC);