From 68a863866a59e7f8d811f2d24502bddfdd7640e9 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 10 Jul 2026 15:22:53 +0200 Subject: [PATCH] =?UTF-8?q?perf(core):=20packetize=20straight=20into=20the?= =?UTF-8?q?=20wire=20pool=20=E2=80=94=20zero-alloc=20host=20send=20path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/punktfunk-core/src/packet.rs | 38 +++++-- crates/punktfunk-core/src/session.rs | 152 +++++++++++++++++++++++++-- 2 files changed, 171 insertions(+), 19 deletions(-) diff --git a/crates/punktfunk-core/src/packet.rs b/crates/punktfunk-core/src/packet.rs index 2a64bd2f..eec38c6e 100644 --- a/crates/punktfunk-core/src/packet.rs +++ b/crates/punktfunk-core/src/packet.rs @@ -135,7 +135,9 @@ impl Packetizer { 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( &mut self, frame: &[u8], @@ -143,6 +145,31 @@ impl Packetizer { user_flags: u32, coder: &dyn ErasureCoder, ) -> Result>> { + 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 frame_index = self.next_frame_index; 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 { let first = b * max_block; let last = ((b + 1) * max_block).min(total_data); @@ -234,14 +260,10 @@ impl Packetizer { fec_scheme: coder.scheme() as u8, flags, }; - - 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); + emit(&hdr, body)?; } } - Ok(packets) + Ok(()) } } diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 3d4412b9..85c6b711 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -16,6 +16,7 @@ use crate::input::InputEvent; use crate::packet::{Packetizer, Reassembler, ReassemblerLimits, MAX_DATAGRAM_BYTES}; use crate::stats::{Stats, StatsCounters}; use crate::transport::Transport; +use zerocopy::IntoBytes; /// A reassembled, FEC-recovered access unit, ready to hand to the platform decoder. pub struct Frame { @@ -166,18 +167,57 @@ impl Session { "seal_frame called on a client session", )); } - let packets = self - .packetizer - .packetize(data, pts_ns, user_flags, self.coder.as_ref())?; + // Packetize straight into the pooled wire buffers (reused across frames via + // `reclaim_wires`) and seal each in place: the plaintext `header ++ shard` is written + // once, at its final wire offset — no intermediate per-packet Vec at all. Byte-identical + // to the wrapper (`packetize` + seal) path: same plaintext, same emission order, and the + // nonce counter advances per emitted packet exactly as before (pinned by the + // wire-equivalence tests below). Destructure into disjoint field borrows first — the + // emit closure needs `crypto`/`next_seq`/the pool while `packetizer` is `&mut`. + let Session { + packetizer, + coder, + 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); - // Reuse the wire-buffer pool the caller returns via `reclaim_wires`: one buffer per packet, - // sealed in place — after warmup there is no per-packet ciphertext/wire allocation. (`wires` - // is a local, so `seal_into`'s `&mut self` doesn't alias the `&mut` iteration over it.) - let mut wires = std::mem::take(&mut self.wire_pool); - wires.resize_with(packets.len(), Vec::new); - for (wire, pkt) in wires.iter_mut().zip(packets.iter()) { - self.seal_into(pkt, wire)?; - } 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.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> { + 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![ + 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 { + (0..len).map(|i| (i * 31 + 7) as u8).collect() + } +} + #[cfg(test)] mod replay_tests { use super::*;