From 1ea38ec71543ac339f5c8dd080d00a8135d571ea Mon Sep 17 00:00:00 2001 From: lbl <1791778603@qq.com> Date: Thu, 20 Aug 2026 22:32:41 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20FEC=20=E5=8F=98=E9=95=BF?= =?UTF-8?q?=E6=89=B9=E6=AC=A1=E8=A7=A3=E7=A0=81=E5=BF=85=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E4=B8=94=E8=BF=9E=E5=B8=A6=E4=B8=A2=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 编码端把所有数据 shard 填充到 max_len 再做 RS 编码，解码端此前按 实际长度存放数据 shard，批次内包长不一致时 reconstruct 必报 IncorrectShardSize。现在记录 shard_size（校验包长度）， 无论校验包先到还是后到都把数据 shard 补齐到统一长度 - 解码失败不再向上传播错误：坏 group 直接移除， 当前完好包与后续数据包正常透传 - 补充变长批次恢复、校验包先到恢复、解码失败不吞好包三个测试 --- vnt-core/src/fec/decoder.rs | 232 +++++++++++++++++++++++++++++++++++- vnt-core/src/fec/encoder.rs | 2 +- 2 files changed, 232 insertions(+), 2 deletions(-) diff --git a/vnt-core/src/fec/decoder.rs b/vnt-core/src/fec/decoder.rs index 576a4f5..6a35e3f 100644 --- a/vnt-core/src/fec/decoder.rs +++ b/vnt-core/src/fec/decoder.rs @@ -26,6 +26,8 @@ struct FecDecoderInner { struct FecGroup { data_shards: usize, parity_shards: usize, + /// shard 统一长度（校验包到达后才能确定，等于编码端填充后的 max_len） + shard_size: usize, received_original_count: usize, received_shards: Vec>>, last_update: Instant, @@ -44,6 +46,7 @@ impl Default for FecGroup { Self { data_shards: 0, parity_shards: 0, + shard_size: 0, received_original_count: 0, received_shards: Vec::with_capacity(16), last_update: Instant::now(), @@ -120,12 +123,21 @@ impl FecDecoder { } group.data_shards = data_shards; group.parity_shards = parity_shards; + // 校验包在线上即编码端填充后的 shard 统一长度 + group.shard_size = payload.len(); if group.received_shards.len() < data_shards + parity_shards { group .received_shards .resize(data_shards + parity_shards, None); } + // RS 要求所有 shard 等长：把先到的数据 shard 补齐到 shard_size + // （编码端发送前把所有数据 shard 填充到 max_len） + for shard in group.received_shards[..data_shards].iter_mut().flatten() { + if shard.len() < group.shard_size { + shard.resize(group.shard_size, 0); + } + } group.received_shards[packet_index] = Some(payload); } else { let buffer = TransmissionBytes::zeroed(HEAD_LENGTH + payload.len()); @@ -147,6 +159,10 @@ impl FecDecoder { batch_data[1] = flags_byte; batch_data[2..4].copy_from_slice(&(payload.len() as u16).to_be_bytes()); batch_data[4..].copy_from_slice(&payload); + // 校验包先到时 shard 尺寸已知，补齐保持所有 shard 等长 + if group.shard_size != 0 && batch_data.len() < group.shard_size { + batch_data.resize(group.shard_size, 0); + } group.received_shards[packet_index] = Some(batch_data); group.received_original_count += 1; } @@ -161,7 +177,26 @@ impl FecDecoder { return Ok(packet.map(|v| vec![v])); } - let result = Self::try_decode(group, (src_ip, group_id), &net_packet)?; + // 解码失败（如对端构造的异常 group）只放弃该 group， + // 不能向上传播错误——否则当前完好接收的包会被连带丢弃 + let decode_failed; + let result = match Self::try_decode(group, (src_ip, group_id), &net_packet) { + Ok(result) => { + decode_failed = false; + result + } + Err(e) => { + log::warn!( + "fec decode failed, drop group, src={src_ip},group_id={group_id}: {e:?}" + ); + decode_failed = true; + None + } + }; + if decode_failed { + // 移除坏 group，后续该 group 的完好数据包按新 group 正常透传 + inner.groups.remove(&(src_ip, group_id)); + } if inner.last_cleanup.elapsed() > Duration::from_secs(1) { Self::cleanup_old_groups(&mut inner.groups); @@ -300,3 +335,198 @@ impl FecDecoder { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::fec::encoder::ParityData; + + const SRC: u32 = 0x0A000001; + const DST: u32 = 0x0A000002; + + /// 构造线上的 FEC 数据包（payload 为 prost 编码的 FecPacket） + fn build_data_packet( + group_id: u64, + packet_index: u32, + type_byte: u8, + flags_byte: u8, + payload: &[u8], + ) -> NetPacket { + let fec = FecPacket { + group_id, + packet_index, + payload: payload.to_vec(), + parity_data: None, + }; + build_packet(fec, type_byte, flags_byte) + } + + fn build_parity_packet( + group_id: u64, + packet_index: u32, + payload: Vec, + data_shards: u32, + parity_shards: u32, + ) -> NetPacket { + let fec = FecPacket { + group_id, + packet_index, + payload, + parity_data: Some(ParityData { + data_shards, + parity_shards, + }), + }; + // 校验包头部的 type/flags 不参与数据重建，固定取值 + build_packet(fec, 0x81, 0) + } + + fn build_packet( + fec: FecPacket, + type_byte: u8, + flags_byte: u8, + ) -> NetPacket { + let fec_payload = fec.encode_to_vec(); + let buffer = TransmissionBytes::zeroed(HEAD_LENGTH + fec_payload.len()); + let mut pkt = NetPacket::new(buffer).unwrap(); + pkt.head_mut()[0] = type_byte; + pkt.head_mut()[2] = flags_byte; + pkt.set_src_id(SRC); + pkt.set_dest_id(DST); + pkt.set_ttl(5); + pkt.set_fec_flag(true); + pkt.set_payload(&fec_payload).unwrap(); + pkt + } + + /// 按编码端格式组装 shard：[type_byte, flags_byte, len(u16), payload...]，填充到 max_len + fn make_shard(type_byte: u8, flags_byte: u8, payload: &[u8], max_len: usize) -> Vec { + let mut shard = vec![0u8; max_len]; + shard[0] = type_byte; + shard[1] = flags_byte; + shard[2..4].copy_from_slice(&(payload.len() as u16).to_be_bytes()); + shard[4..4 + payload.len()].copy_from_slice(payload); + shard + } + + /// 变长批次：丢一个数据包，靠校验包必须能恢复（修复前必报 IncorrectShardSize） + #[test] + fn test_reconstruct_variable_length_batch() { + let decoder = FecDecoder::new(); + let group_id = 1u64; + + let payloads: [&[u8]; 3] = [&[0xAA; 10], &[0xBB; 30], &[0xCC; 20]]; + let type_bytes = [0x81u8, 0x82, 0x83]; + let max_len = 4 + 30; // 编码端按最长 shard 填充 + + // 编码端：3 数据 + 1 校验 + let mut shards: Vec> = payloads + .iter() + .zip(type_bytes) + .map(|(p, t)| make_shard(t, 0, p, max_len)) + .collect(); + shards.push(vec![0u8; max_len]); + let rs = ReedSolomon::new(3, 1).unwrap(); + { + let mut refs: Vec<&mut [u8]> = shards.iter_mut().map(|s| s.as_mut()).collect(); + rs.encode(&mut refs).unwrap(); + } + let parity = shards[3].clone(); + + // 收到 pkt0、pkt1，pkt2 丢失，随后收到校验包 + assert!( + decoder + .receive(build_data_packet(group_id, 0, type_bytes[0], 0, payloads[0])) + .unwrap() + .is_some() + ); + assert!( + decoder + .receive(build_data_packet(group_id, 1, type_bytes[1], 0, payloads[1])) + .unwrap() + .is_some() + ); + let recovered = decoder + .receive(build_parity_packet(group_id, 3, parity, 3, 1)) + .unwrap() + .expect("should reconstruct the lost packet"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].payload(), &payloads[2][..]); + assert_eq!(recovered[0].head()[0], type_bytes[2]); + assert_eq!(recovered[0].src_id(), SRC); + assert_eq!(recovered[0].dest_id(), DST); + assert!(!recovered[0].is_fec()); + } + + /// 校验包先到时，后到的数据 shard 也要补齐（等长）后能恢复 + #[test] + fn test_reconstruct_when_parity_arrives_first() { + let decoder = FecDecoder::new(); + let group_id = 2u64; + + let payloads: [&[u8]; 2] = [&[0x11; 8], &[0x22; 24]]; + let max_len = 4 + 24; + + let mut shards = vec![ + make_shard(0x81, 0, payloads[0], max_len), + make_shard(0x82, 0, payloads[1], max_len), + vec![0u8; max_len], + ]; + let rs = ReedSolomon::new(2, 1).unwrap(); + { + let mut refs: Vec<&mut [u8]> = shards.iter_mut().map(|s| s.as_mut()).collect(); + rs.encode(&mut refs).unwrap(); + } + let parity = shards[2].clone(); + + // 校验包先到（此时丢失 pkt0 还未知），再收到 pkt1 + assert!( + decoder + .receive(build_parity_packet(group_id, 2, parity, 2, 1)) + .unwrap() + .is_none() + ); + let recovered = decoder + .receive(build_data_packet(group_id, 1, 0x82, 0, payloads[1])) + .unwrap() + .expect("should reconstruct pkt0 after parity-first"); + + // 校验包先到时 pkt1 会透传，恢复的 pkt0 也应在返回列表中 + let recovered_pkt0 = recovered + .iter() + .find(|p| p.payload() == &payloads[0][..]) + .expect("recovered pkt0 missing"); + assert_eq!(recovered_pkt0.head()[0], 0x81); + } + + /// 异常 group（校验 shard 比数据 shard 短）导致重建失败时： + /// 当前包不被连带丢弃，坏 group 被移除，后续数据包正常透传 + #[test] + fn test_decode_failure_does_not_swallow_good_packets() { + let decoder = FecDecoder::new(); + let group_id = 3u64; + + // 先到一个较大的数据包 + assert!( + decoder + .receive(build_data_packet(group_id, 0, 0x81, 0, &[0xAA; 40])) + .unwrap() + .is_some() + ); + // 异常校验包：shard 只有 4 字节，比已存数据 shard 短，reconstruct 必失败 + assert!( + decoder + .receive(build_parity_packet(group_id, 2, vec![0u8; 4], 2, 1)) + .unwrap() + .is_none() + ); + // 坏 group 已被移除，后续数据包按新 group 正常透传 + let passed = decoder + .receive(build_data_packet(group_id, 1, 0x82, 0, &[0xBB; 8])) + .unwrap() + .expect("good packet must not be swallowed"); + assert_eq!(passed.len(), 1); + assert_eq!(passed[0].payload(), &[0xBB; 8][..]); + } +} diff --git a/vnt-core/src/fec/encoder.rs b/vnt-core/src/fec/encoder.rs index 0f7eeab..23ebb33 100644 --- a/vnt-core/src/fec/encoder.rs +++ b/vnt-core/src/fec/encoder.rs @@ -15,7 +15,7 @@ mod fec_proto { include!(concat!(env!("OUT_DIR"), "/protocol.fec.rs")); } use crate::utils::task_control::TaskGroup; -pub use fec_proto::FecPacket; +pub use fec_proto::{FecPacket, ParityData}; const BATCH_SIZE: usize = 10; const REDUNDANCY_RATE: f32 = 0.2;