perf(core): packetize straight into the wire pool — zero-alloc host send path

Stage B of the zero-copy host packetize path (networking-audit deferred
plan §1): Packetizer::packetize_each yields (header, shard) pairs in exact
wire order; Session::seal_frame writes seq(8) ‖ header(40) ‖ shard ‖ tag
scratch directly into the pooled wire buffer and seals [8..] in place. The
per-packet intermediate Vec (header ++ body) and its extra memcpy are gone
— with Stage A, every data byte is now copied once (frame → wire) instead
of three times, and the ~2 transient allocs/packet on the send thread are
zero after pool warmup (~180k allocs/s at 1 Gbps rates).

packetize() stays as a thin wrapper over packetize_each — the reference
implementation used by tests and the loss harness.

- wire-equivalence test: pooled path vs wrapper path byte-identical across
  multi-block/partial-tail/exact-multiple/empty frames, fec 0%/50%, both
  schemes, crypto on/off
- loss-harness sweep: recovery rates identical to the pre-item-1 baseline
- bench pipeline (end-to-end incl. client half) vs pre-item-1 baseline,
  stages A+B cumulative: gf16/64K -3.6%, gf16/1M -3.2%; gf8 cases are
  Cauchy-math-bound and unchanged within noise
- cargo ndk check (arm64-v8a) green

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-10 15:22:53 +02:00
parent cdbdc078d6
commit 68a863866a
2 changed files with 171 additions and 19 deletions
+30 -8
View File
@@ -135,7 +135,9 @@ impl Packetizer {
self.fec.fec_percent self.fec.fec_percent
} }
/// Packetize one access unit into wire packets (header + shard payload each). /// Packetize one access unit into owned wire packets (header ++ shard payload each).
/// Thin wrapper over [`packetize_each`](Self::packetize_each) — the allocation-free
/// streaming path's reference implementation (tests and the loss harness use this).
pub fn packetize( pub fn packetize(
&mut self, &mut self,
frame: &[u8], frame: &[u8],
@@ -143,6 +145,31 @@ impl Packetizer {
user_flags: u32, user_flags: u32,
coder: &dyn ErasureCoder, coder: &dyn ErasureCoder,
) -> Result<Vec<Vec<u8>>> { ) -> Result<Vec<Vec<u8>>> {
let mut packets = Vec::new();
self.packetize_each(frame, pts_ns, user_flags, coder, |hdr, body| {
let mut pkt = Vec::with_capacity(HEADER_LEN + body.len());
pkt.extend_from_slice(hdr.as_bytes());
pkt.extend_from_slice(body);
packets.push(pkt);
Ok(())
})?;
Ok(packets)
}
/// Packetize one access unit, yielding each packet to `emit` as a `(header, shard bytes)`
/// pair — in exact wire order, which is also the order the session's nonce counter
/// advances. No per-packet allocation happens here, so the caller can write header and
/// shard straight into a pooled wire buffer and seal in place
/// ([`Session::seal_frame`](crate::session::Session::seal_frame)). An `emit` error aborts
/// the frame mid-way (packet numbering has already advanced — callers treat it as fatal).
pub fn packetize_each(
&mut self,
frame: &[u8],
pts_ns: u64,
user_flags: u32,
coder: &dyn ErasureCoder,
mut emit: impl FnMut(&PacketHeader, &[u8]) -> Result<()>,
) -> Result<()> {
let payload = self.shard_payload; let payload = self.shard_payload;
let frame_index = self.next_frame_index; let frame_index = self.next_frame_index;
self.next_frame_index = self.next_frame_index.wrapping_add(1); self.next_frame_index = self.next_frame_index.wrapping_add(1);
@@ -183,7 +210,6 @@ impl Packetizer {
} }
}; };
let mut packets = Vec::new();
for b in 0..block_count { for b in 0..block_count {
let first = b * max_block; let first = b * max_block;
let last = ((b + 1) * max_block).min(total_data); let last = ((b + 1) * max_block).min(total_data);
@@ -234,14 +260,10 @@ impl Packetizer {
fec_scheme: coder.scheme() as u8, fec_scheme: coder.scheme() as u8,
flags, flags,
}; };
emit(&hdr, body)?;
let mut pkt = Vec::with_capacity(HEADER_LEN + body.len());
pkt.extend_from_slice(hdr.as_bytes());
pkt.extend_from_slice(body);
packets.push(pkt);
} }
} }
Ok(packets) Ok(())
} }
} }
+141 -11
View File
@@ -16,6 +16,7 @@ use crate::input::InputEvent;
use crate::packet::{Packetizer, Reassembler, ReassemblerLimits, MAX_DATAGRAM_BYTES}; use crate::packet::{Packetizer, Reassembler, ReassemblerLimits, MAX_DATAGRAM_BYTES};
use crate::stats::{Stats, StatsCounters}; use crate::stats::{Stats, StatsCounters};
use crate::transport::Transport; use crate::transport::Transport;
use zerocopy::IntoBytes;
/// A reassembled, FEC-recovered access unit, ready to hand to the platform decoder. /// A reassembled, FEC-recovered access unit, ready to hand to the platform decoder.
pub struct Frame { pub struct Frame {
@@ -166,18 +167,57 @@ impl Session {
"seal_frame called on a client session", "seal_frame called on a client session",
)); ));
} }
let packets = self // Packetize straight into the pooled wire buffers (reused across frames via
.packetizer // `reclaim_wires`) and seal each in place: the plaintext `header ++ shard` is written
.packetize(data, pts_ns, user_flags, self.coder.as_ref())?; // once, at its final wire offset — no intermediate per-packet Vec at all. Byte-identical
StatsCounters::add(&self.stats.frames_submitted, 1); // to the wrapper (`packetize` + seal) path: same plaintext, same emission order, and the
// Reuse the wire-buffer pool the caller returns via `reclaim_wires`: one buffer per packet, // nonce counter advances per emitted packet exactly as before (pinned by the
// sealed in place — after warmup there is no per-packet ciphertext/wire allocation. (`wires` // wire-equivalence tests below). Destructure into disjoint field borrows first — the
// is a local, so `seal_into`'s `&mut self` doesn't alias the `&mut` iteration over it.) // emit closure needs `crypto`/`next_seq`/the pool while `packetizer` is `&mut`.
let mut wires = std::mem::take(&mut self.wire_pool); let Session {
wires.resize_with(packets.len(), Vec::new); packetizer,
for (wire, pkt) in wires.iter_mut().zip(packets.iter()) { coder,
self.seal_into(pkt, wire)?; crypto,
next_seq,
wire_pool,
..
} = self;
let mut wires = std::mem::take(wire_pool);
let mut used = 0usize;
let result = packetizer.packetize_each(data, pts_ns, user_flags, coder.as_ref(), {
let wires = &mut wires;
let used = &mut used;
move |hdr, body| {
if *used == wires.len() {
wires.push(Vec::new());
} }
let wire = &mut wires[*used];
*used += 1;
let seq = *next_seq;
*next_seq = next_seq.wrapping_add(1);
wire.clear();
match crypto {
Some(c) => {
// seq(8) ‖ header(40) ‖ shard ‖ tag scratch(16), sealed over [8..].
wire.extend_from_slice(&seq.to_be_bytes());
wire.extend_from_slice(hdr.as_bytes());
wire.extend_from_slice(body);
wire.resize(wire.len() + crate::crypto::TAG_LEN, 0);
c.seal_in_place(seq, &mut wire[8..])?;
}
None => {
wire.extend_from_slice(hdr.as_bytes());
wire.extend_from_slice(body);
}
}
Ok(())
}
});
result?;
// A smaller frame uses fewer buffers than the pool holds: drop the unused tail, same
// as the previous `resize_with(packets.len(), ..)` did.
wires.truncate(used);
StatsCounters::add(&self.stats.frames_submitted, 1);
let bytes: u64 = wires.iter().map(|w| w.len() as u64).sum(); let bytes: u64 = wires.iter().map(|w| w.len() as u64).sum();
StatsCounters::add(&self.stats.packets_sent, wires.len() as u64); StatsCounters::add(&self.stats.packets_sent, wires.len() as u64);
StatsCounters::add(&self.stats.bytes_sent, bytes); StatsCounters::add(&self.stats.bytes_sent, bytes);
@@ -491,6 +531,96 @@ impl ReplayWindow {
} }
} }
#[cfg(test)]
mod wire_equivalence_tests {
use super::*;
use crate::config::{FecConfig, FecScheme, ProtocolPhase};
use crate::transport::loopback_pair;
fn host_cfg(scheme: FecScheme, fec_percent: u8, encrypt: bool) -> Config {
Config {
role: Role::Host,
phase: match scheme {
FecScheme::Gf8 => ProtocolPhase::P1GameStream,
FecScheme::Gf16 => ProtocolPhase::P2Punktfunk,
},
fec: FecConfig {
scheme,
fec_percent,
max_data_per_block: 8,
},
shard_payload: 64,
max_frame_bytes: 8 * 1024 * 1024,
encrypt,
key: [7u8; 16],
salt: [3, 1, 4, 1],
loopback_drop_period: 0,
}
}
fn host_session(cfg: Config) -> Session {
let (h, _c) = loopback_pair(0, 0);
Session::new(cfg, Box::new(h)).unwrap()
}
/// The reference wire path: build owned packets via the `packetize` wrapper, then seal
/// each into its own buffer — the pre-zero-copy implementation of `seal_frame`, spelled
/// out with the session's own private pieces so the two paths share nothing but state.
fn seal_via_wrapper(sess: &mut Session, frame: &[u8], pts_ns: u64, flags: u32) -> Vec<Vec<u8>> {
let packets = sess
.packetizer
.packetize(frame, pts_ns, flags, sess.coder.as_ref())
.unwrap();
let mut wires = Vec::new();
for pkt in &packets {
let mut wire = Vec::new();
sess.seal_into(pkt, &mut wire).unwrap();
wires.push(wire);
}
wires
}
/// `seal_frame`'s packetize-straight-into-the-wire-pool path must produce byte-identical
/// sealed output to the wrapper path (same plaintext = header ++ shard, same nonce
/// sequence) — for multi-block frames, partial tail shards, exact-multiple frames, the
/// empty frame, fec 0%/50%, both schemes, crypto on and off (plan §1.4).
#[test]
fn zero_copy_seal_matches_wrapper_path() {
for scheme in [FecScheme::Gf8, FecScheme::Gf16] {
for fec_percent in [0u8, 50] {
for encrypt in [true, false] {
let mut opt = host_session(host_cfg(scheme, fec_percent, encrypt));
let mut refr = host_session(host_cfg(scheme, fec_percent, encrypt));
// shard_payload 64 × max_data_per_block 8: >512 bytes spans FEC blocks.
let frames: Vec<Vec<u8>> = vec![
pattern(3000), // multi-block + partial tail shard
pattern(1024), // exact multiple (2 full blocks)
pattern(100), // single block, partial tail
Vec::new(), // empty frame → 1 zeroed shard
pattern(64), // exactly one full shard
];
for (i, frame) in frames.iter().enumerate() {
let got = opt.seal_frame(frame, 1000 * i as u64, i as u32).unwrap();
let want = seal_via_wrapper(&mut refr, frame, 1000 * i as u64, i as u32);
assert_eq!(
got, want,
"wire mismatch: scheme={scheme:?} fec={fec_percent}% encrypt={encrypt} frame#{i}"
);
// Return the buffers so later frames exercise the pooled-reuse path
// (including a bigger frame after a smaller one and vice versa).
opt.reclaim_wires(got);
}
}
}
}
}
fn pattern(len: usize) -> Vec<u8> {
(0..len).map(|i| (i * 31 + 7) as u8).collect()
}
}
#[cfg(test)] #[cfg(test)]
mod replay_tests { mod replay_tests {
use super::*; use super::*;