perf(core): FEC encoder reuse — cached codecs + pooled parity, no per-block setup
ci / docs-site (push) Successful in 1m0s
ci / web (push) Successful in 1m0s
decky / build-publish (push) Successful in 19s
docker / build-push (--build-arg FEDORA_VERSION=44, ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm) (push) Successful in 10s
docker / build-push (., web/Dockerfile, punktfunk-web) (push) Successful in 7s
docker / build-push (ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Successful in 8s
docker / build-push (ci, ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Successful in 8s
docker / build-push (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Successful in 8s
ci / bench (push) Successful in 6m4s
windows-host / package (push) Successful in 9m3s
flatpak / build-publish (push) Failing after 8m2s
docker / deploy-docs (push) Successful in 24s
deb / build-publish (push) Successful in 12m19s
android / android (push) Successful in 12m46s
windows-msix / package (arm64, C:\Users\Public\ffmpeg-arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Successful in 4m22s
arch / build-publish (push) Successful in 14m45s
windows-msix / package (x64, C:\Users\Public\ffmpeg, , x86_64-pc-windows-msvc, C:\t) (push) Successful in 3m37s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Successful in 13m42s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Successful in 17m34s
windows / build (aarch64-pc-windows-msvc) (push) Successful in 4m57s
ci / rust (push) Successful in 23m21s
windows / build (x86_64-pc-windows-msvc) (push) Successful in 5m57s
release / apple (push) Has been cancelled
apple / screenshots (push) Has been cancelled
apple / swift (push) Has been cancelled
ci / docs-site (push) Successful in 1m0s
ci / web (push) Successful in 1m0s
decky / build-publish (push) Successful in 19s
docker / build-push (--build-arg FEDORA_VERSION=44, ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm) (push) Successful in 10s
docker / build-push (., web/Dockerfile, punktfunk-web) (push) Successful in 7s
docker / build-push (ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Successful in 8s
docker / build-push (ci, ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Successful in 8s
docker / build-push (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Successful in 8s
ci / bench (push) Successful in 6m4s
windows-host / package (push) Successful in 9m3s
flatpak / build-publish (push) Failing after 8m2s
docker / deploy-docs (push) Successful in 24s
deb / build-publish (push) Successful in 12m19s
android / android (push) Successful in 12m46s
windows-msix / package (arm64, C:\Users\Public\ffmpeg-arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Successful in 4m22s
arch / build-publish (push) Successful in 14m45s
windows-msix / package (x64, C:\Users\Public\ffmpeg, , x86_64-pc-windows-msvc, C:\t) (push) Successful in 3m37s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Successful in 13m42s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Successful in 17m34s
windows / build (aarch64-pc-windows-msvc) (push) Successful in 4m57s
ci / rust (push) Successful in 23m21s
windows / build (x86_64-pc-windows-msvc) (push) Successful in 5m57s
release / apple (push) Has been cancelled
apple / screenshots (push) Has been cancelled
apple / swift (push) Has been cancelled
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 <noreply@anthropic.com>
This commit is contained in:
@@ -6,8 +6,19 @@ use super::{
|
|||||||
validate_block_shape, validate_encode_shape, validate_into_shape, ErasureCoder, FecError,
|
validate_block_shape, validate_encode_shape, validate_into_shape, ErasureCoder, FecError,
|
||||||
};
|
};
|
||||||
use crate::config::FecScheme;
|
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<Option<ReedSolomonEncoder>>,
|
||||||
|
}
|
||||||
|
|
||||||
impl ErasureCoder for Gf16Coder {
|
impl ErasureCoder for Gf16Coder {
|
||||||
fn scheme(&self) -> FecScheme {
|
fn scheme(&self) -> FecScheme {
|
||||||
@@ -15,16 +26,62 @@ impl ErasureCoder for Gf16Coder {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> {
|
fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, 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<Vec<u8>>,
|
||||||
|
) -> Result<(), FecError> {
|
||||||
if recovery_count == 0 {
|
if recovery_count == 0 {
|
||||||
return Ok(Vec::new());
|
out.clear();
|
||||||
|
return Ok(());
|
||||||
}
|
}
|
||||||
validate_encode_shape(data)?;
|
validate_encode_shape(data)?;
|
||||||
let k = data.len();
|
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"));
|
return Err(FecError::Config("GF(2^16) shard length must be even"));
|
||||||
}
|
}
|
||||||
reed_solomon_simd::encode(k, recovery_count, data)
|
let mut guard = self.enc.lock().unwrap_or_else(|p| p.into_inner());
|
||||||
.map_err(|_| FecError::Backend("gf16 encode"))
|
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(
|
fn reconstruct(
|
||||||
|
|||||||
@@ -9,8 +9,16 @@ use super::{
|
|||||||
};
|
};
|
||||||
use crate::config::FecScheme;
|
use crate::config::FecScheme;
|
||||||
use fec_rs::ReedSolomon;
|
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<Option<(usize, usize, ReedSolomon)>>,
|
||||||
|
}
|
||||||
|
|
||||||
impl ErasureCoder for Gf8Coder {
|
impl ErasureCoder for Gf8Coder {
|
||||||
fn scheme(&self) -> FecScheme {
|
fn scheme(&self) -> FecScheme {
|
||||||
@@ -18,20 +26,46 @@ impl ErasureCoder for Gf8Coder {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, FecError> {
|
fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, 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<Vec<u8>>,
|
||||||
|
) -> Result<(), FecError> {
|
||||||
if recovery_count == 0 {
|
if recovery_count == 0 {
|
||||||
return Ok(Vec::new());
|
out.clear();
|
||||||
|
return Ok(());
|
||||||
}
|
}
|
||||||
validate_encode_shape(data)?;
|
validate_encode_shape(data)?;
|
||||||
let k = data.len();
|
let k = data.len();
|
||||||
let shard_len = data[0].len();
|
let shard_len = data[0].len();
|
||||||
let rs = ReedSolomon::new(k, recovery_count)
|
let mut guard = self.rs.lock().unwrap_or_else(|p| p.into_inner());
|
||||||
.map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
|
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 —
|
// `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.
|
// same Cauchy codec as `encode`, without copying the data into a shards scratch.
|
||||||
let mut parity: Vec<Vec<u8>> = (0..recovery_count).map(|_| vec![0u8; shard_len]).collect();
|
rs.encode_sep(data, out)
|
||||||
rs.encode_sep(data, &mut parity)
|
|
||||||
.map_err(|_| FecError::Backend("gf8 encode"))?;
|
.map_err(|_| FecError::Backend("gf8 encode"))?;
|
||||||
Ok(parity)
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn reconstruct(
|
fn reconstruct(
|
||||||
@@ -121,7 +155,7 @@ mod tests {
|
|||||||
/// these vectors would break and our parity would no longer be Moonlight-decodable.
|
/// these vectors would break and our parity would no longer be Moonlight-decodable.
|
||||||
#[test]
|
#[test]
|
||||||
fn nanors_exact_parity_vectors() {
|
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].
|
// 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 data: [&[u8]; 4] = [&[10u8], &[20], &[30], &[40]];
|
||||||
let parity = coder.encode(&data, 2).unwrap();
|
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.
|
/// Round-trip: erase `m` data shards and confirm reconstruction recovers the originals.
|
||||||
#[test]
|
#[test]
|
||||||
fn recovers_erased_data_shards() {
|
fn recovers_erased_data_shards() {
|
||||||
let coder = Gf8Coder;
|
let coder = Gf8Coder::default();
|
||||||
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 refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect();
|
let refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).collect();
|
||||||
let parity = coder.encode(&refs, 3).unwrap();
|
let parity = coder.encode(&refs, 3).unwrap();
|
||||||
|
|||||||
@@ -34,6 +34,23 @@ pub trait ErasureCoder: Send + Sync {
|
|||||||
/// buffer instead of copying every data byte into per-shard `Vec`s first.
|
/// 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>;
|
fn encode(&self, data: &[&[u8]], recovery_count: usize) -> Result<Vec<Vec<u8>>, 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<Vec<u8>>,
|
||||||
|
) -> Result<(), FecError> {
|
||||||
|
*out = self.encode(data, recovery_count)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// 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.
|
||||||
/// Returns the K original shards in order.
|
/// Returns the K original shards in order.
|
||||||
@@ -67,8 +84,8 @@ pub trait ErasureCoder: Send + Sync {
|
|||||||
/// Construct the coder for a scheme.
|
/// Construct the coder for a scheme.
|
||||||
pub fn coder_for(scheme: FecScheme) -> Box<dyn ErasureCoder> {
|
pub fn coder_for(scheme: FecScheme) -> Box<dyn ErasureCoder> {
|
||||||
match scheme {
|
match scheme {
|
||||||
FecScheme::Gf8 => Box::new(Gf8Coder),
|
FecScheme::Gf8 => Box::new(Gf8Coder::default()),
|
||||||
FecScheme::Gf16 => Box::new(Gf16Coder),
|
FecScheme::Gf16 => Box::new(Gf16Coder::default()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -221,15 +238,15 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn gf16_reconstruct_into_fills_only_the_holes() {
|
fn gf16_reconstruct_into_fills_only_the_holes() {
|
||||||
roundtrip_into(&Gf16Coder, 16, 4, 256, &[1, 9], &[3]);
|
roundtrip_into(&Gf16Coder::default(), 16, 4, 256, &[1, 9], &[3]);
|
||||||
roundtrip_into(&Gf16Coder, 4, 2, 16, &[0, 3], &[]);
|
roundtrip_into(&Gf16Coder::default(), 4, 2, 16, &[0, 3], &[]);
|
||||||
roundtrip_into(&Gf16Coder, 4, 2, 16, &[], &[0, 1]); // nothing missing, no parity needed
|
roundtrip_into(&Gf16Coder::default(), 4, 2, 16, &[], &[0, 1]); // nothing missing, no parity needed
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn gf8_reconstruct_into_fills_only_the_holes() {
|
fn gf8_reconstruct_into_fills_only_the_holes() {
|
||||||
roundtrip_into(&Gf8Coder, 16, 4, 256, &[0, 7], &[1]);
|
roundtrip_into(&Gf8Coder::default(), 16, 4, 256, &[0, 7], &[1]);
|
||||||
roundtrip_into(&Gf8Coder, 4, 2, 16, &[2], &[1]);
|
roundtrip_into(&Gf8Coder::default(), 4, 2, 16, &[2], &[1]);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -238,24 +255,24 @@ mod tests {
|
|||||||
// Too few shards: 2 of 4 data present, no recovery.
|
// Too few shards: 2 of 4 data present, no recovery.
|
||||||
let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect();
|
let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect();
|
||||||
let have = [true, true, false, false];
|
let have = [true, true, false, false];
|
||||||
assert!(Gf16Coder
|
assert!(Gf16Coder::default()
|
||||||
.reconstruct_into(2, &mut slots, &have, &[])
|
.reconstruct_into(2, &mut slots, &have, &[])
|
||||||
.is_err());
|
.is_err());
|
||||||
// Recovery index out of the declared range.
|
// Recovery index out of the declared range.
|
||||||
let parity = [0u8; 8];
|
let parity = [0u8; 8];
|
||||||
let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect();
|
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)])
|
.reconstruct_into(2, &mut slots, &have, &[(2, &parity), (3, &parity)])
|
||||||
.is_err());
|
.is_err());
|
||||||
// Mismatched recovery shard length.
|
// Mismatched recovery shard length.
|
||||||
let short = [0u8; 6];
|
let short = [0u8; 6];
|
||||||
let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect();
|
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)])
|
.reconstruct_into(2, &mut slots, &have, &[(0, &short), (1, &parity)])
|
||||||
.is_err());
|
.is_err());
|
||||||
// `have` length disagreeing with `data`.
|
// `have` length disagreeing with `data`.
|
||||||
let mut slots: Vec<&mut [u8]> = buf.chunks_mut(8).collect();
|
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)])
|
.reconstruct_into(2, &mut slots, &[true; 3], &[(0, &parity)])
|
||||||
.is_err());
|
.is_err());
|
||||||
}
|
}
|
||||||
@@ -263,19 +280,19 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn gf8_recovers_within_budget() {
|
fn gf8_recovers_within_budget() {
|
||||||
// 16 data + 4 recovery; lose 2 data + 2 recovery (== 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]
|
#[test]
|
||||||
fn gf16_recovers_within_budget() {
|
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]
|
#[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 refs: Vec<&[u8]> = data.iter().map(|s| s.as_slice()).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<Option<Vec<u8>>> = data
|
let mut received: Vec<Option<Vec<u8>>> = data
|
||||||
.iter()
|
.iter()
|
||||||
.cloned()
|
.cloned()
|
||||||
@@ -286,8 +303,8 @@ mod tests {
|
|||||||
received[0] = None;
|
received[0] = None;
|
||||||
received[1] = None;
|
received[1] = None;
|
||||||
received[2] = None;
|
received[2] = None;
|
||||||
assert!(Gf16Coder.scheme() == FecScheme::Gf16);
|
assert!(Gf16Coder::default().scheme() == FecScheme::Gf16);
|
||||||
let err = Gf8Coder.reconstruct(8, 2, &mut received);
|
let err = Gf8Coder::default().reconstruct(8, 2, &mut received);
|
||||||
assert!(err.is_err());
|
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
|
// data=2, recovery=2 expects a 4-element slice; a 3-element one must error, not
|
||||||
// panic on the recovery-slice index (both backends).
|
// panic on the recovery-slice index (both backends).
|
||||||
let mut recv: Vec<Option<Vec<u8>>> = vec![Some(vec![0u8; 8]), None, Some(vec![0u8; 8])];
|
let mut recv: Vec<Option<Vec<u8>>> = 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<Option<Vec<u8>>> = vec![Some(vec![0u8; 8]), None, Some(vec![0u8; 8])];
|
let mut recv: Vec<Option<Vec<u8>>> = 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]
|
#[test]
|
||||||
@@ -306,9 +323,9 @@ mod tests {
|
|||||||
// The GF16 fast path used to clone shards verbatim without a length check.
|
// The GF16 fast path used to clone shards verbatim without a length check.
|
||||||
let mut recv: Vec<Option<Vec<u8>>> =
|
let mut recv: Vec<Option<Vec<u8>>> =
|
||||||
vec![Some(vec![0u8; 8]), Some(vec![0u8; 6]), None, None];
|
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<Option<Vec<u8>>> =
|
let mut recv: Vec<Option<Vec<u8>>> =
|
||||||
vec![Some(vec![0u8; 8]), Some(vec![0u8; 6]), None, None];
|
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());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -147,6 +147,10 @@ pub struct Packetizer {
|
|||||||
/// Every other data shard is a `shard_payload`-sized slice straight into the frame buffer —
|
/// 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.
|
/// blocks are consecutive shard ranges, so only the frame's last shard can be partial.
|
||||||
tail: Vec<u8>,
|
tail: Vec<u8>,
|
||||||
|
/// 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<Vec<u8>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Packetizer {
|
impl Packetizer {
|
||||||
@@ -159,6 +163,7 @@ impl Packetizer {
|
|||||||
fec: config.fec,
|
fec: config.fec,
|
||||||
version: config.phase as u8,
|
version: config.phase as u8,
|
||||||
tail: Vec::new(),
|
tail: Vec::new(),
|
||||||
|
recovery: Vec::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -262,6 +267,7 @@ impl Packetizer {
|
|||||||
self.tail[..rem].copy_from_slice(&frame[full_shards * payload..]);
|
self.tail[..rem].copy_from_slice(&frame[full_shards * payload..]);
|
||||||
}
|
}
|
||||||
let tail = &self.tail;
|
let tail = &self.tail;
|
||||||
|
let recovery_pool = &mut self.recovery;
|
||||||
let shard_at = |s: usize| -> &[u8] {
|
let shard_at = |s: usize| -> &[u8] {
|
||||||
if s < full_shards {
|
if s < full_shards {
|
||||||
&frame[s * payload..(s + 1) * payload]
|
&frame[s * payload..(s + 1) * payload]
|
||||||
@@ -279,7 +285,8 @@ impl Packetizer {
|
|||||||
let data_shards: Vec<&[u8]> = (first..last).map(shard_at).collect();
|
let data_shards: Vec<&[u8]> = (first..last).map(shard_at).collect();
|
||||||
|
|
||||||
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)?;
|
coder.encode_into(&data_shards, recovery_count, recovery_pool)?;
|
||||||
|
let recovery = &*recovery_pool;
|
||||||
let total_shards = block_data_count + recovery_count;
|
let total_shards = block_data_count + recovery_count;
|
||||||
if total_shards > u16::MAX as usize {
|
if total_shards > u16::MAX as usize {
|
||||||
return Err(PunktfunkError::Unsupported("block shard count exceeds u16"));
|
return Err(PunktfunkError::Unsupported("block shard count exceeds u16"));
|
||||||
|
|||||||
@@ -49,6 +49,9 @@ pub struct VideoPacketizer {
|
|||||||
frame_index: u32,
|
frame_index: u32,
|
||||||
/// Monotonic per-stream packet counter (the RTP sequence / streamPacketIndex source).
|
/// Monotonic per-stream packet counter (the RTP sequence / streamPacketIndex source).
|
||||||
seq: u32,
|
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 {
|
impl VideoPacketizer {
|
||||||
@@ -65,6 +68,7 @@ impl VideoPacketizer {
|
|||||||
min_fec: min_fec as usize,
|
min_fec: min_fec as usize,
|
||||||
frame_index: 0,
|
frame_index: 0,
|
||||||
seq: 0,
|
seq: 0,
|
||||||
|
coder: Gf8Coder::default(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -158,7 +162,7 @@ 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 {
|
||||||
let refs: Vec<&[u8]> = shards.iter().map(|s| s.as_slice()).collect();
|
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 {
|
} else {
|
||||||
Vec::new()
|
Vec::new()
|
||||||
};
|
};
|
||||||
@@ -328,7 +332,9 @@ mod tests {
|
|||||||
// Drop data shard 1; reconstruct from the rest via the same Cauchy coder.
|
// Drop data shard 1; reconstruct from the rest via the same Cauchy coder.
|
||||||
let mut received: Vec<Option<Vec<u8>>> = pkts.iter().map(|p| Some(p.clone())).collect();
|
let mut received: Vec<Option<Vec<u8>>> = pkts.iter().map(|p| Some(p.clone())).collect();
|
||||||
received[1] = None;
|
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
|
// 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.
|
// byte (offset 24) is PIC (middle shard), proving the NV header recovers correctly.
|
||||||
assert_eq!(recovered[1][24], FLAG_PIC);
|
assert_eq!(recovered[1][24], FLAG_PIC);
|
||||||
|
|||||||
Reference in New Issue
Block a user