From 0058f624a236b2beb53cd45c367bec9f9defaf22 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Tue, 14 Jul 2026 19:07:10 +0200 Subject: [PATCH] feat(core): receive-path stage timing + frame-jitter observability (PUNKTFUNK_PERF) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session::poll_frame accumulates per-stage ns (recv_batch syscall, AES-GCM open, Reassembler::push incl. FEC) into a PumpPerf drained via take_pump_perf(); the client pump logs the split plus completed-AU inter-arrival jitter (p50/p95/max + late count) every report window. Gated on PUNKTFUNK_PERF — one branch per stage when off. Smoothness previously had no metric at all (jump-to-live counters fire seconds late), and the receive core had no attribution. First live use pinned the 1.57 Gbps client wall on software AES-GCM (7 µs/pkt) vs 0.4 µs reassembly — see punktfunk-planning/design/throughput-beyond-1gbps.md. Co-Authored-By: Claude Fable 5 --- crates/punktfunk-core/src/client.rs | 50 +++++++++++++++++++++++++ crates/punktfunk-core/src/session.rs | 55 ++++++++++++++++++++++++++-- 2 files changed, 102 insertions(+), 3 deletions(-) diff --git a/crates/punktfunk-core/src/client.rs b/crates/punktfunk-core/src/client.rs index 59631ea4..b02776ab 100644 --- a/crates/punktfunk-core/src/client.rs +++ b/crates/punktfunk-core/src/client.rs @@ -1935,6 +1935,13 @@ async fn worker_main(args: WorkerArgs) { const ADAPT_REPORT_INTERVAL: Duration = Duration::from_millis(750); let mut last_report = Instant::now(); let (mut last_recovered, mut last_received, mut last_dropped) = (0u64, 0u64, 0u64); + // PUNKTFUNK_PERF: per-window pump observability — the Session's receive stage split + // (recv / decrypt / reassemble+FEC, see `Session::take_pump_perf`) and completed-AU + // inter-arrival jitter. Smoothness has no metric otherwise: jump-to-live counters only + // fire after the stream is already seconds behind. + let pump_perf_on = std::env::var("PUNKTFUNK_PERF").is_ok_and(|v| v != "0"); + let mut arrivals_us: Vec = Vec::new(); + let mut last_arrival: Option = None; // Adaptive bitrate (see `crate::abr`): armed only when the embedder asked for Automatic // (`bitrate_kbps == 0`) and the host echoed the rate it actually configured (an old host // echoes 0 → controller stays permanently off). Fed once per report window with the same @@ -2037,12 +2044,55 @@ async fn worker_main(args: WorkerArgs) { last_recovered = st.fec_recovered_shards; last_received = st.packets_received; last_dropped = st.frames_dropped; + if pump_perf_on { + if let Some(p) = session.take_pump_perf() { + let per_pkt_ns = |ns: u64| ns.checked_div(p.packets).unwrap_or(0); + tracing::info!( + recv_ms = p.recv_ns / 1_000_000, + decrypt_ms = p.decrypt_ns / 1_000_000, + reasm_ms = p.reasm_ns / 1_000_000, + packets = p.packets, + batches = p.batches, + pkts_per_batch = p.packets.checked_div(p.batches).unwrap_or(0), + decrypt_ns_pkt = per_pkt_ns(p.decrypt_ns), + reasm_ns_pkt = per_pkt_ns(p.reasm_ns), + "pump stage split (window)" + ); + } + // Inter-arrival jitter over the window's completed AUs. `late` counts gaps + // over 2× the window median — the "a frame arrived visibly off-beat" tally. + if arrivals_us.len() >= 8 { + arrivals_us.sort_unstable(); + let pct = |q: usize| arrivals_us[(arrivals_us.len() - 1) * q / 100]; + let (p50, p95) = (pct(50), pct(95)); + let late = arrivals_us.iter().filter(|&&d| d > p50 * 2).count(); + tracing::info!( + frames = arrivals_us.len() + 1, + arrival_p50_us = p50, + arrival_p95_us = p95, + arrival_max_us = arrivals_us.last().copied().unwrap_or(0), + late, + "frame inter-arrival jitter (window)" + ); + } + arrivals_us.clear(); + } } match session.poll_frame() { Ok(frame) => { if frame.flags & FLAG_PROBE as u32 != 0 { continue; // speed-test filler, not video — measured via the counters above } + if pump_perf_on { + let now = Instant::now(); + if let Some(prev) = last_arrival.replace(now) { + // 4096 ≈ 17 s at 240 fps — a stuck window can't grow it unbounded. + if arrivals_us.len() < 4096 { + arrivals_us.push((now - prev).as_micros().min(u32::MAX as u128) + as u32); + } + } + } // Jump-to-live guard. A standing receive/hand-off queue never drains by itself — // the pump consumes strictly in order at the arrival rate, so once behind, the // stream stays behind for good (observed live: stuck 6–7 s). Pre-decode AUs are diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 73a3e5f2..4c1b1004 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -59,6 +59,28 @@ pub struct Session { /// then returns them via [`reclaim_wires`](Self::reclaim_wires)). After warmup each buffer keeps /// its capacity, so the per-packet ciphertext + wire `Vec` allocations vanish from the hot path. wire_pool: Vec>, + /// Receive-path stage timing (`PUNKTFUNK_PERF`), read+reset via [`take_pump_perf`] + /// (Self::take_pump_perf). `None` when disabled — the hot path then pays one branch per stage. + perf: Option, +} + +/// Accumulated client receive-path stage timings since the last [`Session::take_pump_perf`]. +/// Answers "where does the pump core go" at line rate: kernel drain (`recv_ns`) vs AES-GCM +/// (`decrypt_ns`) vs reassembly+FEC (`reasm_ns`, the `Reassembler::push` round-trip including +/// shard copies and block reconstruction). 2026-07-14 sweep context: the pump pegs one core at +/// ~1.5 Gbps wire, ~85% of it userspace — this split is what Phase 2.1 (pooled reassembly) is +/// validated against. +#[derive(Debug, Default, Clone, Copy)] +pub struct PumpPerf { + /// ns inside `recv_batch` (recvmmsg / recvmsg_x), i.e. syscall + kernel copy. + pub recv_ns: u64, + /// ns inside `open_in_place` across all datagrams (AES-128-GCM + replay-window upkeep). + pub decrypt_ns: u64, + /// ns inside `Reassembler::push` (header parse, shard copy, FEC reconstruct, AU assembly). + pub reasm_ns: u64, + /// recv_batch calls (batches) and datagrams processed over the accumulation window. + pub batches: u64, + pub packets: u64, } /// Datagrams drained per `recvmmsg` syscall on the client (the reused ring's size). At ~125k @@ -91,10 +113,20 @@ impl Session { recv_count: 0, recv_idx: 0, wire_pool: Vec::new(), + // Same opt-in the host's stage logs use; read once — set it before connecting. + perf: std::env::var("PUNKTFUNK_PERF") + .is_ok_and(|v| v != "0") + .then(PumpPerf::default), config, }) } + /// Drain the receive-path stage timings accumulated since the last call (window semantics — + /// the pump reads this once per report interval). `None` when `PUNKTFUNK_PERF` is off. + pub fn take_pump_perf(&mut self) -> Option { + self.perf.as_mut().map(std::mem::take) + } + pub fn role(&self) -> Role { self.config.role } @@ -362,9 +394,14 @@ impl Session { loop { // Refill the ring with one `recvmmsg` batch when the current one is drained. if self.recv_idx >= self.recv_count { + let t0 = self.perf.is_some().then(std::time::Instant::now); self.recv_count = self .transport .recv_batch(&mut self.recv_scratch, &mut self.recv_lens)?; + if let (Some(p), Some(t0)) = (self.perf.as_mut(), t0) { + p.recv_ns += t0.elapsed().as_nanos() as u64; + p.batches += 1; + } self.recv_idx = 0; if self.recv_count == 0 { return Err(PunktfunkError::NoFrame); @@ -384,6 +421,9 @@ impl Session { // one). The plaintext lands at [8..8+n] of the sealed wire (behind the seq prefix); an // unencrypted (probe) datagram IS the packet. Field-precise borrows keep the slice into // `recv_scratch` alive across the replay/reassembler calls below. + // Perf note: the two `continue`s below (short / undecryptable noise) skip the decrypt + // accounting — they are the exception path, not line-rate traffic. + let t_dec = self.perf.is_some().then(std::time::Instant::now); let (pkt_range, seq) = match &self.crypto { Some(c) => { // A sealed datagram is at least seq prefix + tag; anything shorter is noise. @@ -398,6 +438,9 @@ impl Session { } None => (0..len, None), }; + if let (Some(p), Some(t)) = (self.perf.as_mut(), t_dec) { + p.decrypt_ns += t.elapsed().as_nanos() as u64; + } // Anti-replay (same rationale as poll_input): reject a datagram whose authenticated // sequence was already seen. Video also dedups per-frame downstream, but filtering here // is uniform and cheap. @@ -412,10 +455,16 @@ impl Session { StatsCounters::add(&self.stats.bytes_received, pkt.len() as u64); // The reassembler validates the packet via its parsed header (`magic`), // ignoring anything that isn't a well-formed video packet. - if let Some(frame) = self + let t_push = self.perf.is_some().then(std::time::Instant::now); + let pushed = self .reassembler - .push(pkt, self.coder.as_ref(), &self.stats)? - { + .push(pkt, self.coder.as_ref(), &self.stats)?; + if let (Some(p), Some(t)) = (self.perf.as_mut(), t_push) { + p.reasm_ns += t.elapsed().as_nanos() as u64; + // Counts datagrams that reached the reassembler (replay-rejected ones don't). + p.packets += 1; + } + if let Some(frame) = pushed { StatsCounters::add(&self.stats.frames_completed, 1); return Ok(frame); }