From bcb2d891cc5ec21769f85445e0c2e631cf3f4b99 Mon Sep 17 00:00:00 2001 From: Celestial Date: Thu, 9 Apr 2026 14:47:12 +0200 Subject: [PATCH 1/2] perf(shred): skip primary fec reparse --- .../src/app/runtime/runloop/packet_workers.rs | 3 +- crates/sof-observer/src/shred/fec/core.rs | 28 ++++++++----------- 2 files changed, 14 insertions(+), 17 deletions(-) diff --git a/crates/sof-observer/src/app/runtime/runloop/packet_workers.rs b/crates/sof-observer/src/app/runtime/runloop/packet_workers.rs index 18f5fcc8..5f560a61 100644 --- a/crates/sof-observer/src/app/runtime/runloop/packet_workers.rs +++ b/crates/sof-observer/src/app/runtime/runloop/packet_workers.rs @@ -510,7 +510,8 @@ where parsed_header_slot(&packet.parsed_header), &mut observed_slot_leaders, ); - let recovered_packets = fec_recoverer.ingest_packet(packet.packet_bytes.as_ref()); + let recovered_packets = fec_recoverer + .ingest_packet(packet.packet_bytes.as_ref(), &packet.parsed_header); push_primary_shred(packet, &mut accepted_shreds); for recovered in recovered_packets { diff --git a/crates/sof-observer/src/shred/fec/core.rs b/crates/sof-observer/src/shred/fec/core.rs index 836ac8a5..cc69e738 100644 --- a/crates/sof-observer/src/shred/fec/core.rs +++ b/crates/sof-observer/src/shred/fec/core.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, hash_map::Entry}; use reed_solomon_erasure::galois_8::ReedSolomon; -use crate::shred::wire::{ParsedShred, ShredVariant, parse_shred}; +use crate::shred::wire::{ParsedShredHeader, ShredVariant}; #[path = "recover.rs"] mod recover; @@ -69,23 +69,19 @@ impl FecRecoverer { } } - pub fn ingest_packet(&mut self, packet: &[u8]) -> Vec> { - let parsed = match parse_shred(packet) { - Ok(parsed) => parsed, - Err(_) => return Vec::new(), - }; + pub fn ingest_packet(&mut self, packet: &[u8], parsed: &ParsedShredHeader) -> Vec> { let signature = match parse_packet_signature(packet) { Some(signature) => signature, None => return Vec::new(), }; - let (slot, fec_set_index, variant) = match &parsed { - ParsedShred::Data(data) => ( + let (slot, fec_set_index, variant) = match parsed { + ParsedShredHeader::Data(data) => ( data.common.slot, data.common.fec_set_index, SetVariant::from(data.common.shred_variant), ), - ParsedShred::Code(code) => ( + ParsedShredHeader::Code(code) => ( code.common.slot, code.common.fec_set_index, SetVariant::from(code.common.shred_variant), @@ -102,13 +98,13 @@ impl FecRecoverer { if !set.accepts_variant(variant) { return Vec::new(); } - set.ingest_packet(&parsed, packet); + set.ingest_packet(parsed, packet); recovered = recover_missing_data(set, fec_set_index, &mut self.reed_solomon_cache) .unwrap_or_default(); should_remove = set.is_data_complete_for_config(fec_set_index); } else { let mut new_set = ErasureSet::new(signature); - new_set.ingest_packet(&parsed, packet); + new_set.ingest_packet(parsed, packet); let _ = self.sets.insert(set_id, new_set); } @@ -154,10 +150,10 @@ impl ErasureSet { self.variant.is_none_or(|existing| existing == incoming) } - fn ingest_packet(&mut self, parsed: &ParsedShred, packet: &[u8]) { + fn ingest_packet(&mut self, parsed: &ParsedShredHeader, packet: &[u8]) { let common_variant = match parsed { - ParsedShred::Data(data) => data.common.shred_variant, - ParsedShred::Code(code) => code.common.shred_variant, + ParsedShredHeader::Data(data) => data.common.shred_variant, + ParsedShredHeader::Code(code) => code.common.shred_variant, }; if self.variant.is_none() { self.variant = Some(SetVariant::from(common_variant)); @@ -167,7 +163,7 @@ impl ErasureSet { }; match parsed { - ParsedShred::Data(data) => { + ParsedShredHeader::Data(data) => { let Some(shard) = extract_data_erasure_shard(packet, shard_len) else { return; }; @@ -179,7 +175,7 @@ impl ErasureSet { } } } - ParsedShred::Code(code) => { + ParsedShredHeader::Code(code) => { let Some(shard) = extract_coding_erasure_shard(packet, shard_len) else { return; }; From aaca6c5236dfea18b615b525289dc1bad351f311 Mon Sep 17 00:00:00 2001 From: Celestial Date: Thu, 9 Apr 2026 15:17:30 +0200 Subject: [PATCH 2/2] fix(fuzz): align shred fec recover target --- .../fuzz/fuzz_targets/shred_fec_recover.rs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/crates/sof-observer/fuzz/fuzz_targets/shred_fec_recover.rs b/crates/sof-observer/fuzz/fuzz_targets/shred_fec_recover.rs index f28cfab7..3b88de36 100644 --- a/crates/sof-observer/fuzz/fuzz_targets/shred_fec_recover.rs +++ b/crates/sof-observer/fuzz/fuzz_targets/shred_fec_recover.rs @@ -13,7 +13,7 @@ use sof::{ }, shred::{ fec::FecRecoverer, - wire::{ParsedShred, parse_shred}, + wire::{ParsedDataShredHeader, ParsedShred, ParsedShredHeader, parse_shred}, }, }; @@ -217,7 +217,20 @@ fuzz_target!(|bytes: &[u8]| { } }; - let recovered = recoverer.ingest_packet(&packet); + let Ok(parsed) = parse_shred(&packet) else { + assert!(recoverer.tracked_sets() <= max_tracked_sets); + continue; + }; + let parsed_header = match parsed { + ParsedShred::Data(data) => ParsedShredHeader::Data(ParsedDataShredHeader { + common: data.common, + data_header: data.data_header, + payload_offset: SIZE_OF_DATA_SHRED_HEADERS, + payload_len: data.payload.len(), + }), + ParsedShred::Code(code) => ParsedShredHeader::Code(code), + }; + let recovered = recoverer.ingest_packet(&packet, &parsed_header); assert!(recoverer.tracked_sets() <= max_tracked_sets); for recovered_packet in recovered {