GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge FEC: own the GF(2^8) Reed-Solomon math; drop reed-solomon-erasure

Browse source 
Operator: "FEC: Please own the math." rs-gf256-v1 implemented in
media_fec.rs, wire-identical to the crate (1,795 geometries, independent
derivation, 6,520 cross-version recoveries, 60k hostile blocks identical).
lru and instant advisories gone; 13 fewer dependencies. combine uses one
product row per coefficient: encode ~1.0-1.4x, decode ~1.2-1.4x the crate.
Follow-up: k^3 inversion makes 255+1 recovery ~124 ms.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 16:48:18 +02:00
commit 016df4626c
5 changed files with 1765 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/cultnet-rs/Cargo.toml
View file

@ -19,7 +19,6 @@ cultcache-rs = { path = "../cultcache-rs" }
ed25519-dalek = "2.2"
hmac = "0.12"
rand = "0.9"
reed-solomon-erasure = "6"
rmp-serde = "1.3"
rmpv = { version = "1.3", features = ["with-serde"] }
serde = { version = "1", features = ["derive"] }

195
packages/cultnet-rs/src/media_fec.rs
View file

@ -14,11 +14,11 @@
//! the `k + m` shards recover the block (the code is MDS), so any `m` erasures
//! are repaired and any more are reported as [`MediaFecError::BeyondRepair`],
//! never as wrong bytes. The construction is named on every parity record by
//! [`MEDIA_FEC_SCHEME_RS_GF256_V1`] and backed by the `reed-solomon-erasure`
//! crate. The library is not the contract: the construction below is, and
//! `tests/media_fec.rs` pins the parity bytes of known inputs, so a library that
//! changes the matrix fails there instead of on the wire, and a decoder in
//! another runtime has vectors to meet.
//! [`MEDIA_FEC_SCHEME_RS_GF256_V1`] and implemented in this module, from the
//! text below. The construction is the contract: `tests/media_fec.rs` and
//! `tests/fixtures/media_fec_rs_gf256_v1.kat` pin the parity bytes of known
//! inputs, so a change to the math fails there instead of on the wire, and a
//! decoder in another runtime has vectors to meet.
//!
//! # The `rs-gf256-v1` construction
//!
@ -42,8 +42,10 @@
//! * Any `k` surviving shards recover the block: invert the `k x k` submatrix
//! of `E` for the surviving rows and multiply.
//!
//! The known-answer vectors in `tests/media_fec.rs` were checked against an
//! independent implementation of exactly this text.
//! The known-answer vectors were produced by `reed-solomon-erasure` 6.0.0's
//! `galois_8` field, which builds exactly this matrix, before this module
//! replaced it; the smaller ones in `tests/media_fec.rs` were also checked
//! against a second, independent implementation of this text.
//!
//! # What a block is
//!
@ -88,8 +90,6 @@
use std::collections::BTreeMap;
use std::fmt;
use reed_solomon_erasure::galois_8::ReedSolomon;
use crate::media_stream_contracts::{
GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord,
GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord,
@ -102,8 +102,8 @@ use crate::media_stream_wire::{
use crate::rudp::RUDP_FIXED_HEADER_BYTES;
/// Names the construction on every parity record: systematic Reed-Solomon over
/// GF(2^8), as built by `reed-solomon-erasure`'s `galois_8` field. A different
/// matrix is a different scheme id.
/// GF(2^8), as the module docs state it. A different matrix is a different
/// scheme id.
pub const MEDIA_FEC_SCHEME_RS_GF256_V1: &str = "rs-gf256-v1";
/// The largest UDP payload an IPv4 datagram carries on a 1500-byte MTU.
@ -237,40 +237,165 @@ impl fmt::Display for MediaFecError {
impl std::error::Error for MediaFecError {}
// ---------------------------------------------------------------------------
// The code: equal-length shards in, equal-length shards out.
// The code: rs-gf256-v1 exactly as the module docs state it. Equal-length
// shards in, equal-length shards out; every caller has already bounded a block
// to 1..=256 shards of one validated length, so nothing here allocates more
// than the block's shards and three `k x k` matrices.
// ---------------------------------------------------------------------------
fn parity_shards(data: &[Vec<u8>], parity_count: usize) -> Result<Vec<Vec<u8>>, MediaFecError> {
let code = ReedSolomon::new(data.len(), parity_count)
.map_err(|error| MediaFecError::invalid(format!("{error:?}")))?;
let mut parity = vec![vec![0_u8; data[0].len()]; parity_count];
code.encode_sep(data, &mut parity)
.map_err(|error| MediaFecError::invalid(format!("{error:?}")))?;
Ok(parity)
/// `x^8 + x^4 + x^3 + x^2 + 1`, the polynomial the field is reduced by.
const FIELD_POLYNOMIAL: u16 = 0x11D;
/// `EXP[i]` is `2^i`, written out over two periods so the sum of two logarithms
/// indexes it without reduction; `LOG` is its inverse on the non-zero
/// elements. `2` generates the multiplicative group of GF(2^8) under
/// [`FIELD_POLYNOMIAL`], so the tables cover the field.
const FIELD_TABLES: ([u8; 510], [u8; 256]) = {
let mut exp = [0_u8; 510];
let mut log = [0_u8; 256];
let mut power: u16 = 1;
let mut i = 0;
while i < 255 {
exp[i] = power as u8;
exp[i + 255] = power as u8;
log[power as usize] = i as u8;
power <<= 1;
if power & 0x100 != 0 {
power ^= FIELD_POLYNOMIAL;
}
i += 1;
}
(exp, log)
};
fn gf_mul(a: u8, b: u8) -> u8 {
let (exp, log) = &FIELD_TABLES;
if a == 0 || b == 0 {
return 0;
}
exp[usize::from(log[usize::from(a)]) + usize::from(log[usize::from(b)])]
}
/// `1 / a` for non-zero `a`.
fn gf_reciprocal(a: u8) -> u8 {
let (exp, log) = &FIELD_TABLES;
exp[255 - usize::from(log[usize::from(a)])]
}
/// Row `r` of the Vandermonde matrix `V`: `r^0, r^1, .., r^(k-1)`, the row
/// number read as a field element, with `0^0 = 1`.
fn vandermonde_row(r: usize, k: usize) -> Vec<u8> {
let point = r as u8;
let mut row = Vec::with_capacity(k);
let mut power = 1_u8;
for _ in 0..k {
row.push(power);
power = gf_mul(power, point);
}
row
}
/// The inverse of a square matrix, by Gauss-Jordan elimination.
fn invert(mut matrix: Vec<Vec<u8>>) -> Vec<Vec<u8>> {
let n = matrix.len();
let mut inverse: Vec<Vec<u8>> =
(0..n).map(|r| (0..n).map(|c| u8::from(r == c)).collect()).collect();
for col in 0..n {
let pivot = (col..n)
.find(|&r| matrix[r][col] != 0)
.expect("rs-gf256-v1 only inverts matrices whose evaluation points are distinct");
matrix.swap(col, pivot);
inverse.swap(col, pivot);
let scale = gf_reciprocal(matrix[col][col]);
for c in 0..n {
matrix[col][c] = gf_mul(scale, matrix[col][c]);
inverse[col][c] = gf_mul(scale, inverse[col][c]);
}
let (pivot_row, pivot_inverse) = (matrix[col].clone(), inverse[col].clone());
for r in (0..n).filter(|&r| r != col) {
let factor = matrix[r][col];
for c in 0..n {
matrix[r][c] ^= gf_mul(factor, pivot_row[c]);
inverse[r][c] ^= gf_mul(factor, pivot_inverse[c]);
}
}
}
inverse
}
/// The given rows of the encoding matrix `E = V * inverse(top)` for a block of
/// `k` data shards, where `top` is the first `k` rows of `V`.
fn encoding_rows(k: usize, rows: impl Iterator<Item = usize>) -> Vec<Vec<u8>> {
let top_inverse = invert((0..k).map(|r| vandermonde_row(r, k)).collect());
rows.map(|r| {
let v = vandermonde_row(r, k);
(0..k)
.map(|c| (0..k).fold(0, |sum, i| sum ^ gf_mul(v[i], top_inverse[i][c])))
.collect()
})
.collect()
}
/// Byte `j` of the result is the sum over `i` of `coefficients[i] * shards[i][j]`.
///
/// The hot loop of both encode and recovery: each coefficient's 256 products
/// are tabulated once, so a shard byte costs one lookup and one XOR.
fn combine(coefficients: &[u8], shards: &[&[u8]], shard_bytes: usize) -> Vec<u8> {
let mut out = vec![0_u8; shard_bytes];
for (&coefficient, shard) in coefficients.iter().zip(shards) {
match coefficient {
0 => {}
1 => {
for (byte, &input) in out.iter_mut().zip(shard.iter()) {
*byte ^= input;
}
}
_ => {
let products: [u8; 256] = std::array::from_fn(|x| gf_mul(coefficient, x as u8));
for (byte, &input) in out.iter_mut().zip(shard.iter()) {
*byte ^= products[usize::from(input)];
}
}
}
}
out
}
/// The `parity_count` parity shards of `data`: rows `k..k + m` of `E`.
fn parity_shards(data: &[Vec<u8>], parity_count: usize) -> Vec<Vec<u8>> {
let k = data.len();
let inputs: Vec<&[u8]> = data.iter().map(Vec::as_slice).collect();
encoding_rows(k, k..k + parity_count)
.iter()
.map(|row| combine(row, &inputs, data[0].len()))
.collect()
}
/// `shards` holds the block's `k` data slots then `m` parity slots, `None` for
/// each missing. Returns the `k` data shards.
/// each missing. Returns the `k` data shards, recovered from the first `k`
/// present shards through the inverse of their rows of `E`.
fn recover_data_shards(
data_shards: usize,
mut shards: Vec<Option<Vec<u8>>>,
shards: Vec<Option<Vec<u8>>>,
) -> Result<Vec<Vec<u8>>, MediaFecError> {
let parity_shards = shards.len() - data_shards;
let present = shards.iter().flatten().count();
if present < data_shards {
let present: Vec<usize> = (0..shards.len())
.filter(|&slot| shards[slot].is_some())
.take(data_shards)
.collect();
if present.len() < data_shards {
return Err(MediaFecError::BeyondRepair {
shards_present: present,
shards_present: present.len(),
shards_needed: data_shards,
});
}
let code = ReedSolomon::new(data_shards, parity_shards)
.map_err(|error| MediaFecError::invalid(format!("{error:?}")))?;
code.reconstruct_data(&mut shards)
.map_err(|error| MediaFecError::invalid(format!("{error:?}")))?;
Ok(shards
.into_iter()
.take(data_shards)
.map(|shard| shard.expect("reconstruct_data fills every data shard"))
let inputs: Vec<&[u8]> = present.iter().map(|&slot| shards[slot].as_deref().unwrap()).collect();
let shard_bytes = inputs[0].len();
let decode = invert(encoding_rows(data_shards, present.iter().copied()));
Ok((0..data_shards)
.map(|slot| match &shards[slot] {
Some(shard) => shard.clone(),
None => combine(&decode[slot], &inputs, shard_bytes),
})
.collect())
}
@ -393,7 +518,7 @@ pub fn protect_video_frame(
.map(|chunk| padded(&chunk.payload, shard_bytes))
.collect();
let parity_count = policy.video_parity_shards(size);
let parity = parity_shards(&data, usize::from(parity_count))?;
let parity = parity_shards(&data, usize::from(parity_count));
let parity_records: Vec<GameCultMediaWireRecord> = parity
.into_iter()
@ -634,7 +759,7 @@ pub fn protect_audio_block(
let data: Vec<Vec<u8>> = packets.iter().map(|packet| packet.payload.clone()).collect();
let deadline_ticks = packets.iter().map(|packet| packet.deadline_ticks).max().unwrap_or(0);
let parity: Vec<GameCultMediaAudioParityShardRecord> =
parity_shards(&data, usize::from(policy.audio_parity_shards))?
parity_shards(&data, usize::from(policy.audio_parity_shards))
.into_iter()
.enumerate()
.map(|(parity_index, payload)| GameCultMediaAudioParityShardRecord {

1322
packages/cultnet-rs/tests/fixtures/media_fec_rs_gf256_v1.kat vendored Normal file
View file

File diff suppressed because one or more lines are too long

195
packages/cultnet-rs/tests/media_fec.rs
View file

@ -1141,3 +1141,198 @@ fn a_budget_at_the_media_record_ceiling_admits_large_shards_end_to_end() {
.unwrap();
decode_media_wire_record(&wire).unwrap();
}
// ---------------------------------------------------------------------------
// Known answers from the fixture: every standard geometry and the policy's
// edges, byte for byte. The fixture's header states how its data is made.
// ---------------------------------------------------------------------------
/// The data of a fixture case, as the fixture's header defines it.
fn kat_data(k: usize, m: usize, shard_bytes: usize, fill: &str) -> Vec<Vec<u8>> {
(0..k)
.map(|i| match fill {
"lcg" => bytes((k * 65_536 + m * 256 + i) as u32, shard_bytes),
"zero" => vec![0; shard_bytes],
"ff" => vec![0xff; shard_bytes],
other => panic!("unknown fill {other}"),
})
.collect()
}
struct KatCase {
k: usize,
m: usize,
shard_bytes: usize,
fill: String,
parity: Vec<Vec<u8>>,
erasures: Vec<Vec<usize>>,
}
impl KatCase {
fn name(&self) -> String {
format!("k={} m={} {} bytes {}", self.k, self.m, self.shard_bytes, self.fill)
}
/// The case's data as an audio block: the codec path that takes any `(k, m)`.
fn packets(&self) -> Vec<GameCultMediaAudioPacketRecord> {
let mut packets = audio_run(self.k as u64, self.shard_bytes);
for (packet, data) in packets.iter_mut().zip(kat_data(self.k, self.m, self.shard_bytes, &self.fill)) {
packet.payload = data;
}
packets
}
fn policy(&self) -> MediaFecPolicy {
MediaFecPolicy {
audio_data_shards: self.k as u16,
audio_parity_shards: self.m as u16,
max_wire_bytes: GAMECULT_MEDIA_MAX_WIRE_BYTES,
..STANDARD
}
}
}
fn unhex(text: &str) -> Vec<u8> {
(0..text.len())
.step_by(2)
.map(|at| u8::from_str_radix(&text[at..at + 2], 16).expect("fixture hex"))
.collect()
}
fn kat_cases() -> Vec<KatCase> {
let mut cases: Vec<KatCase> = Vec::new();
for line in include_str!("fixtures/media_fec_rs_gf256_v1.kat").lines() {
let fields: Vec<&str> = line.split_whitespace().collect();
match fields.as_slice() {
[] => {}
[comment, ..] if comment.starts_with('#') => {}
["case", k, m, shard_bytes, fill] => cases.push(KatCase {
k: k.parse().unwrap(),
m: m.parse().unwrap(),
shard_bytes: shard_bytes.parse().unwrap(),
fill: fill.to_string(),
parity: Vec::new(),
erasures: Vec::new(),
}),
["parity", index, bytes] => {
let case = cases.last_mut().expect("parity follows a case");
assert_eq!(index.parse::<usize>().unwrap(), case.parity.len(), "{}", case.name());
case.parity.push(unhex(bytes));
}
["erase", slots] => cases
.last_mut()
.expect("erase follows a case")
.erasures
.push(slots.split(',').map(|slot| slot.parse().unwrap()).collect()),
other => panic!("unreadable fixture line {other:?}"),
}
}
cases
}
/// Parity bytes of every fixture case, byte for byte. If these ever change,
/// `rs-gf256-v1` changed meaning: bump the scheme id, do not edit the fixture.
#[test]
fn every_rs_gf256_v1_known_answer_parity_matches_byte_for_byte() {
let cases = kat_cases();
assert_eq!(cases.len(), 35, "the fixture's case count");
for case in &cases {
assert_eq!(case.parity.len(), case.m, "{}", case.name());
let parity = protect_audio_block(&case.packets(), &case.policy(), provenance())
.unwrap_or_else(|error| panic!("{}: {error}", case.name()));
let produced: Vec<String> = parity.iter().map(|shard| hex(&shard.payload)).collect();
let expected: Vec<String> = case.parity.iter().map(|shard| hex(shard)).collect();
assert_eq!(produced, expected, "{}", case.name());
}
}
/// Every fixture erasure pattern, data and parity mixed, recovers exactly
/// the erased data from the fixture's own parity bytes.
#[test]
fn every_rs_gf256_v1_known_erasure_pattern_recovers_the_data() {
for case in kat_cases() {
let packets = case.packets();
let mut parity = protect_audio_block(&packets, &case.policy(), provenance()).unwrap();
for (shard, bytes) in parity.iter_mut().zip(&case.parity) {
shard.payload = bytes.clone(); // decode from the fixture, not from the encoder
}
assert!(!case.erasures.is_empty(), "{}", case.name());
for erased in &case.erasures {
let data: Vec<_> =
(0..case.k).filter(|slot| !erased.contains(slot)).map(|slot| packets[slot].clone()).collect();
let present: Vec<_> = (0..case.m)
.filter(|slot| !erased.contains(&(case.k + slot)))
.map(|slot| parity[slot].clone())
.collect();
let recovered = recover_audio_block(&present, &data)
.unwrap_or_else(|error| panic!("{} erased {erased:?}: {error}", case.name()));
let lost: Vec<usize> = erased.iter().copied().filter(|&slot| slot < case.k).collect();
assert_eq!(recovered.len(), lost.len(), "{} erased {erased:?}", case.name());
for (packet, slot) in recovered.iter().zip(lost) {
assert_eq!(packet.packet_id, packets[slot].packet_id, "{} erased {erased:?}", case.name());
assert_eq!(packet.payload, packets[slot].payload, "{} erased {erased:?}", case.name());
}
}
}
}
/// Seeded random geometries past the fixture's: any `e <= m` erasures, data
/// and parity mixed, recover the data exactly, and `m + 1` are refused.
#[test]
fn random_erasures_up_to_m_recover_exactly_and_m_plus_one_is_refused() {
let mut state = 0x9e37_79b9_u32;
let mut next = |bound: usize| {
state ^= state << 13;
state ^= state >> 17;
state ^= state << 5;
state as usize % bound
};
for trial in 0..300 {
let k = 1 + next(48);
let m = 1 + next(24);
let shard_bytes = 1 + next(40);
let mut packets = audio_run(k as u64, shard_bytes);
for (index, packet) in packets.iter_mut().enumerate() {
packet.payload = bytes((trial * 1_000 + index) as u32, shard_bytes);
}
let policy = MediaFecPolicy {
audio_data_shards: k as u16,
audio_parity_shards: m as u16,
..STANDARD
};
let parity = protect_audio_block(&packets, &policy, provenance()).unwrap();
for erasures in [1 + next(m), m + 1] {
if erasures + 1 > k + m {
continue; // at least one parity shard must arrive to recover with
}
// A random erasure set that leaves at least one parity shard.
let mut slots: Vec<usize> = (0..k + m).collect();
let erased = loop {
for at in 0..slots.len() {
let swap = at + next(slots.len() - at);
slots.swap(at, swap);
}
let erased = slots[..erasures].to_vec();
if (k..k + m).any(|slot| !erased.contains(&slot)) {
break erased;
}
};
let data: Vec<_> =
(0..k).filter(|slot| !erased.contains(slot)).map(|slot| packets[slot].clone()).collect();
let present: Vec<_> =
(0..m).filter(|slot| !erased.contains(&(k + slot))).map(|slot| parity[slot].clone()).collect();
let outcome = recover_audio_block(&present, &data);
let context = format!("trial {trial}: k={k} m={m} {shard_bytes} bytes erased {erased:?}");
if erasures <= m {
let mut all = data;
all.extend(outcome.unwrap_or_else(|error| panic!("{context}: {error}")));
all.sort_by_key(|packet| packet.packet_id);
let payloads: Vec<_> = all.iter().map(|packet| &packet.payload).collect();
let expected: Vec<_> = packets.iter().map(|packet| &packet.payload).collect();
assert_eq!(payloads, expected, "{context}");
} else {
assert!(matches!(outcome, Err(MediaFecError::BeyondRepair { .. })), "{context}: {outcome:?}");
}
}
}
}

88
packages/cultnet-rs/tests/media_fec_bench.rs Normal file
View file

@ -0,0 +1,88 @@
//! Timings for the media FEC codec, without a bench framework.
//!
//! Ignored by default; run in release, where the numbers mean something:
//!
//! ```text
//! cargo test --release --test media_fec_bench -- --ignored --nocapture
//! ```
//!
//! Each line is nanoseconds per call through the public surface, so record
//! validation and cloning are included. The geometries are the ones the codec
//! meets: standard audio, the widest standard video block at a datagram, and
//! the worst recovery the policy admits (255 + 1).
use std::hint::black_box;
use std::time::Instant;
use cultnet_rs::{
GAMECULT_MEDIA_MAX_WIRE_BYTES, GameCultMediaAudioPacketRecord, MediaFecPolicy,
MediaWireProvenance, protect_audio_block, recover_audio_block,
};
fn provenance() -> MediaWireProvenance<'static> {
MediaWireProvenance {
stored_at: "unix:1700000000000",
runtime_id: "bench",
role: "bench.media",
producer: "bench",
}
}
fn packets(k: usize, len: usize) -> Vec<GameCultMediaAudioPacketRecord> {
let mut state = 0x2545_f491_u32;
(0..k as u64)
.map(|index| GameCultMediaAudioPacketRecord {
stream_id: "bench".to_string(),
session_id: "bench".to_string(),
packet_id: index,
codec: "opus".to_string(),
pts_ticks: 960 * index as i64,
duration_ticks: 960,
timebase_num: 1,
timebase_den: 48_000,
deadline_ticks: 960 * index as i64 + 5_000,
payload: (0..len)
.map(|_| {
state ^= state << 13;
state ^= state >> 17;
state ^= state << 5;
state as u8
})
.collect(),
})
.collect()
}
/// Mean nanoseconds per call over at least `min_ms`, after three warm-up calls.
fn time(mut f: impl FnMut(), min_ms: u128) -> f64 {
for _ in 0..3 {
f();
}
let start = Instant::now();
let mut calls = 0_u64;
while start.elapsed().as_millis() < min_ms {
f();
calls += 1;
}
start.elapsed().as_nanos() as f64 / calls as f64
}
#[test]
#[ignore = "timing, not a test: run in release with --ignored --nocapture"]
fn media_fec_timings() {
for (k, m, len) in [(4, 2, 160), (4, 2, 1300), (16, 4, 1200), (255, 1, 1300)] {
let policy = MediaFecPolicy {
audio_data_shards: k as u16,
audio_parity_shards: m as u16,
max_wire_bytes: GAMECULT_MEDIA_MAX_WIRE_BYTES,
..MediaFecPolicy::STANDARD
};
let block = packets(k, len);
let encode = time(|| drop(black_box(protect_audio_block(black_box(&block), &policy, provenance()))), 300);
let parity = protect_audio_block(&block, &policy, provenance()).unwrap();
let lost = m.min(k);
let survivors = &block[lost..];
let decode = time(|| drop(black_box(recover_audio_block(black_box(&parity), survivors))), 300);
println!("k={k:3} m={m:3} len={len:5}: protect {encode:12.0} ns | recover {lost} lost {decode:12.0} ns");
}
}