From e4905643165073230a09b9a08d7d486301c7af58 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Tue, 7 Jul 2026 07:34:24 +0200 Subject: [PATCH] fix(core,host): make the native data plane survive real Wi-Fi links MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-caused live on a phone at 100 Mbps (stream stuck seconds behind, then oscillating): a stack of transport defects, each amplifying the next. - MTU-safe shards: shard_payload 1452 overshot the IPv4/1500 budget (the old math forgot the 40 B header + 24 B crypto ride inside the UDP payload and counted IP+UDP as 8 B) — the kernel silently split EVERY video datagram into two IP fragments, doubling per-datagram loss on Wi-Fi. New config::mtu1500_shard_payload() = 1408 (1472 sealed = the exact ceiling), negotiated in the Welcome, pinned by a unit test. - Android batched I/O: recv/send batching was cfg(linux); Android is target_os="android" and silently fell back to a syscall per datagram. The libc crate binds neither recvmmsg/sendmmsg nor mmsghdr for Android, so a local bionic extern binding provides them (API 21+, floor is 28); cbindgen excludes them from the C header. The pump/runtime threads also get the Apple-QoS analogue on Android: nice −8 (below the decode thread's −10). - Latency-bounded receive: packets are consumed strictly in order at exactly the arrival rate, so a standing queue (Wi-Fi stall, power-save clumping) NEVER drains — observed as a stream permanently 6-7 s behind with both 32 MB socket buffers full. The pump now flushes the entire backlog (Session::flush_backlog: discard ring + kernel queue at memcpy speed, reset the reassembler) and requests a keyframe when frames keep completing > 400 ms behind the skew-corrected capture clock (30 consecutive, 2 s cooldown, logged). - Time-based loss window: the reassembler declared an incomplete frame lost a fixed 4 INDICES behind the newest — 33 ms at 120 fps, inside normal Wi-Fi retry/reorder timescales, so merely-late frames were pruned every few seconds, each costing a recovery-IDR burst + an inflated loss report. Now 120 ms of capture time (LOSS_WINDOW_NS), same fuse at every refresh rate, with a 64-index hard cap bounding memory against hostile pts. - Adaptive-FEC hysteresis: the controller was memoryless — one clean 750 ms report dropped FEC from 8 % straight back to the 1 % floor, so periodic burst loss (Wi-Fi scan / BT coexistence beats) always hit an unprotected stream and ping-ponged 1↔8 % with a frozen frame per cycle (observed in the host log as alternating loss_ppm=0/50000). Attack stays instant; decay is now one point per clean report. Verified: full core suite (incl. new flush + time-window tests) on macOS + Linux, host release build, arm64 cargo-ndk build, and a 30 s wired probe run at 2800x1260@120 — 3559/3559 frames, zero loss, capture→received p50 5.3 ms (host 5.1 + network 0.3). Co-Authored-By: Claude Fable 5 --- crates/punktfunk-core/benches/pipeline.rs | 2 +- crates/punktfunk-core/cbindgen.toml | 9 +- crates/punktfunk-core/src/client.rs | 84 ++++++++++- crates/punktfunk-core/src/config.rs | 26 ++++ crates/punktfunk-core/src/packet.rs | 168 ++++++++++++++++++--- crates/punktfunk-core/src/session.rs | 39 +++++ crates/punktfunk-core/src/transport/udp.rs | 69 +++++++-- crates/punktfunk-core/tests/loopback.rs | 42 ++++++ crates/punktfunk-host/src/punktfunk1.rs | 31 +++- 9 files changed, 418 insertions(+), 52 deletions(-) diff --git a/crates/punktfunk-core/benches/pipeline.rs b/crates/punktfunk-core/benches/pipeline.rs index 4abc929d..4fc48bf0 100644 --- a/crates/punktfunk-core/benches/pipeline.rs +++ b/crates/punktfunk-core/benches/pipeline.rs @@ -16,7 +16,7 @@ use punktfunk_core::session::Session; use punktfunk_core::transport::loopback_pair; const TAG_LEN: usize = 16; // AES-GCM authentication tag -const SHARD: usize = 1452; // ~one MTU-sized data shard +const SHARD: usize = punktfunk_core::config::mtu1500_shard_payload(); // one MTU-safe data shard fn cfg(role: Role, scheme: FecScheme) -> Config { Config { diff --git a/crates/punktfunk-core/cbindgen.toml b/crates/punktfunk-core/cbindgen.toml index 4bfbb531..0e263da0 100644 --- a/crates/punktfunk-core/cbindgen.toml +++ b/crates/punktfunk-core/cbindgen.toml @@ -13,10 +13,11 @@ documentation_style = "c99" parse_deps = false [export] -# Internal Apple-only FFI (transport/udp.rs `recvmsg_x` batched recv + its `MsghdrX`) — NOT part of -# the C ABI. cbindgen otherwise sweeps the foreign import and its #[repr(C)] struct into the header, -# where socklen_t/ssize_t/iovec are undefined and the C harness fails to compile. -exclude = ["MsghdrX", "recvmsg_x"] +# Internal platform-only FFI — NOT part of the C ABI. cbindgen otherwise sweeps the foreign +# imports and their #[repr(C)] structs into the header, where socklen_t/ssize_t/iovec/msghdr are +# undefined and the C harness fails to compile: the Apple batched recv (transport/udp.rs +# `recvmsg_x` + `MsghdrX`) and the Android bionic mmsg bindings (`android_mmsg` module). +exclude = ["MsghdrX", "recvmsg_x", "mmsghdr", "sendmmsg", "recvmmsg"] [export.rename] "InputEvent" = "PunktfunkInputEvent" diff --git a/crates/punktfunk-core/src/client.rs b/crates/punktfunk-core/src/client.rs index 5f560932..7141c1fd 100644 --- a/crates/punktfunk-core/src/client.rs +++ b/crates/punktfunk-core/src/client.rs @@ -123,6 +123,24 @@ pub struct ProbeOutcome { /// (display freshness over completeness — FEC/keyframes recover). const FRAME_QUEUE: usize = 16; +/// Backlog latency bound: when completed frames keep arriving further than this behind the host's +/// capture clock (skew-corrected), the pump flushes the receive backlog +/// ([`Session::flush_backlog`]) and requests a keyframe instead of playing that far behind +/// forever. Deliberately generous — an interactive stream is unusable well before 400 ms, but the +/// bound must sit safely above the skew handshake's own error (≈ RTT/2) plus normal delivery +/// jitter so a healthy stream can never trip it. +const FLUSH_LATENCY: Duration = Duration::from_millis(400); + +/// How many CONSECUTIVE over-bound frames arm a flush (~0.5 s at 60 fps). A genuine standing queue +/// puts EVERY frame over the bound; a one-off burst (an IDR, a Wi-Fi scan blip) clears within a +/// frame or two and never reaches the count. +const FLUSH_AFTER_FRAMES: u32 = 30; + +/// Minimum spacing between backlog flushes, so a bottleneck that instantly rebuilds the queue (a +/// link that can't sustain the bitrate at all) degrades into a periodic skip + a logged warning +/// instead of a continuous flush/keyframe storm. +const FLUSH_COOLDOWN: Duration = Duration::from_secs(2); + /// Audio packets buffered for the embedder: 64 × 5 ms = 320 ms of slack. A lagging /// embedder drops the newest packet (the audio renderer conceals the gap). const AUDIO_QUEUE: usize = 64; @@ -248,8 +266,9 @@ pub struct NativeClient { /// std channels these worker threads feed; if the producers run at the default QoS, the /// kernel sees a high-QoS thread parked waiting on a lower-QoS one and the Thread Performance /// Checker flags a priority inversion. Matching the producers to the consumers' QoS removes -/// the inversion without slowing the Swift side. No-op off Apple (the Linux client/host don't -/// run a QoS scheduler, and `punktfunk-probe` doesn't care). +/// the inversion without slowing the Swift side. Android gets a nice-level analogue (see the +/// android arm below); a no-op elsewhere (the Linux client/host don't run a QoS scheduler, and +/// `punktfunk-probe` doesn't care). #[cfg(target_vendor = "apple")] fn pin_thread_user_interactive() { // SAFETY: sets only the current thread's QoS class — always valid to call. @@ -257,9 +276,33 @@ fn pin_thread_user_interactive() { libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_USER_INTERACTIVE, 0); } } -#[cfg(not(target_vendor = "apple"))] +/// Android analogue of the Apple QoS pin: raise the calling thread to nice −8 (the framework's +/// URGENT_DISPLAY band — apps may set negative nice on their own threads). At default nice 0 the +/// EAS scheduler happily parks the data-plane pump (UDP receive + decrypt + FEC — a thread that +/// sleeps between bursts) on a down-clocked little core, and a few ms of scheduling delay during a +/// keyframe burst overflows the socket receive buffer → wire loss the link never saw. −8 keeps the +/// pipeline below the decode thread's −10 (the display path still wins). Best-effort, like Apple's. +#[cfg(target_os = "android")] +fn pin_thread_user_interactive() { + // SAFETY: `gettid`/`setpriority` on the calling thread are always-safe syscalls; a refusal is + // reported via the return value (ignored — a missed boost, not an error on the data path). + unsafe { + let tid = libc::gettid(); + let _ = libc::setpriority(libc::PRIO_PROCESS, tid as libc::id_t, -8); + } +} +#[cfg(not(any(target_vendor = "apple", target_os = "android")))] fn pin_thread_user_interactive() {} +/// Wall-clock now in nanoseconds (CLOCK_REALTIME basis), to compare against the host-stamped +/// capture `pts_ns` after the skew offset is applied — the same latency math the stats HUDs use. +fn now_realtime_ns() -> i128 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as i128) + .unwrap_or(0) +} + /// The calling thread's kernel id, for hot-thread performance hints (the Android client's ADPF /// session today; the consumer is platform-specific). Linux/Android expose `gettid`; elsewhere /// there's nothing to hint with, so registration is a no-op. @@ -1196,6 +1239,11 @@ 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); + // Backlog latency bound (see FLUSH_LATENCY): consecutive over-bound frames + the last + // flush, for the cooldown. Armed only when the skew handshake succeeded (offset ≠ 0) — + // without it the host and client clocks aren't comparable and the bound would misfire. + let mut stale_frames: u32 = 0; + let mut last_flush: Option = None; while !pump_shutdown.load(Ordering::SeqCst) { // Mirror the reassembler's unrecoverable-drop count for the client's keyframe-recovery // loop, and (during a speed test) the packet-level receive counters for the throughput @@ -1230,6 +1278,36 @@ async fn worker_main(args: WorkerArgs) { if frame.flags & FLAG_PROBE as u32 != 0 { continue; // speed-test filler, not video — measured via the counters above } + // Latency bound: a standing receive queue (pump transiently outpaced, a Wi-Fi + // stall, power-save clumping) never drains by itself — the pump consumes at + // exactly the arrival rate, so once behind, the stream stays behind for good + // (observed live: stuck 6–7 s). When frames keep completing over the bound, + // discard the whole backlog and ask for a keyframe: one visible skip instead of + // a permanently unusable stream. Suspended during a speed test (the probe + // MEASURES a saturated queue; flushing would corrupt its receive counters). + if clock_offset_ns != 0 && !probe_active { + let lat_ns = + now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128; + if lat_ns > FLUSH_LATENCY.as_nanos() as i128 { + stale_frames += 1; + } else { + stale_frames = 0; + } + if stale_frames >= FLUSH_AFTER_FRAMES + && last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN) + { + stale_frames = 0; + last_flush = Some(Instant::now()); + let flushed = session.flush_backlog().unwrap_or(0); + let _ = ctrl_tx.send(CtrlRequest::Keyframe); + tracing::warn!( + behind_ms = lat_ns / 1_000_000, + flushed_datagrams = flushed, + "receive backlog exceeded the latency bound — flushed to live" + ); + continue; // this frame is part of the stale past — don't render it + } + } let _ = frame_tx.try_send(frame); } Err(PunktfunkError::NoFrame) => { diff --git a/crates/punktfunk-core/src/config.rs b/crates/punktfunk-core/src/config.rs index ad6ea7e9..cbe623ed 100644 --- a/crates/punktfunk-core/src/config.rs +++ b/crates/punktfunk-core/src/config.rs @@ -256,6 +256,19 @@ pub const fn max_shard_payload() -> usize { MAX_DATAGRAM_BYTES - HEADER_LEN - CRYPTO_OVERHEAD } +/// Largest **even** shard payload whose sealed wire datagram still fits an unfragmented IPv4/UDP +/// packet on a standard 1500-byte MTU: `1500 − 20 (IPv4) − 8 (UDP) − HEADER_LEN − CRYPTO_OVERHEAD` +/// = 1408. Hosts should default `shard_payload` to this: one byte more and the kernel silently +/// splits EVERY video datagram into two IP fragments (a full frame plus a runt) — either fragment +/// lost = the datagram lost, roughly doubling per-datagram loss on Wi-Fi and eating straight into +/// FEC's recovery margin, plus per-pair kernel reassembly and runt airtime at line rate. (Exactly +/// what the previous hardcoded 1452 did: its MTU math forgot the punktfunk header + crypto ride +/// inside the UDP payload and counted the IP+UDP headers as 8 bytes instead of 28.) +pub const fn mtu1500_shard_payload() -> usize { + let p = 1500 - 20 - 8 - HEADER_LEN - CRYPTO_OVERHEAD; + p - p % 2 // FEC requires even shards +} + /// Everything needed to construct a [`Session`](crate::session::Session). /// /// `Debug` is implemented by hand to redact `key`/`salt`, and `key`/`salt` are zeroized @@ -392,6 +405,19 @@ mod tests { assert!(c.validate().is_err()); } + /// Pin the 1500-MTU wire math: the sealed datagram (header + shard + crypto) at the MTU-safe + /// shard payload must be ≤ 1472 (1500 − IPv4 20 − UDP 8), and one shard-step (+2) above must + /// not — the regression that shipped as 1452 and IP-fragmented every video datagram. + #[test] + fn mtu1500_shard_payload_never_fragments() { + let p = mtu1500_shard_payload(); + assert_eq!(p % 2, 0, "FEC requires even shards"); + assert!(p <= max_shard_payload()); + let wire = HEADER_LEN + p + CRYPTO_OVERHEAD; + assert!(wire <= 1472, "sealed datagram {wire} B would IP-fragment"); + assert!(HEADER_LEN + (p + 2) + CRYPTO_OVERHEAD > 1472, "not maximal"); + } + #[test] fn rejects_block_exceeding_scheme_ceiling() { let mut c = Config::p1_defaults(Role::Host); // Gf8, ceiling 255 diff --git a/crates/punktfunk-core/src/packet.rs b/crates/punktfunk-core/src/packet.rs index 78cd1de2..a2c712df 100644 --- a/crates/punktfunk-core/src/packet.rs +++ b/crates/punktfunk-core/src/packet.rs @@ -43,8 +43,29 @@ pub const CRYPTO_OVERHEAD: usize = 8 + crate::crypto::TAG_LEN; /// `shard_payload` so `HEADER_LEN + shard_payload + CRYPTO_OVERHEAD ≤ MAX_DATAGRAM_BYTES`. pub const MAX_DATAGRAM_BYTES: usize = 2048; -/// How many frames behind the newest the reassembler keeps before pruning stragglers. -const REORDER_WINDOW: u32 = 16; +/// How far behind the newest frame's capture pts an INCOMPLETE frame may sit before it is +/// declared lost (counted in `frames_dropped`, which triggers the client's recovery-keyframe +/// request). TIME-based, not frame-count-based, so the fuse is the same at every refresh rate: a +/// fixed index window is refresh-relative (4 frames = 66 ms at 60 fps but only 33 ms at 120 fps — +/// inside normal Wi-Fi retry/block-ack reorder timescales, where a delayed-not-lost shard can +/// trail newer frames). Observed live at 120 fps: the too-tight fuse declared merely-late frames +/// dead every few seconds, and each false loss cost a recovery-IDR burst + an inflated loss report +/// (FEC churn) — a self-sustaining latency/bitrate oscillation. 120 ms rides safely above radio +/// retry jitter while still detecting a real loss ~2× faster than the original 16-frame window did +/// at 60 fps. +const LOSS_WINDOW_NS: u64 = 120_000_000; + +/// Hard cap on how many frame INDICES behind the newest an incomplete frame may sit, whatever its +/// pts claims — bounds the reassembler's memory against a corrupt/hostile pts (which +/// [`LOSS_WINDOW_NS`] alone would trust) and against pathologically high frame rates. At 120 fps, +/// 120 ms ≈ 14 indices, so 64 leaves ample slack up to ~500 fps. +const HARD_LOSS_WINDOW: u32 = 64; + +/// How many frames behind the newest the reassembler remembers emitted/abandoned frame indices +/// (`completed`), so a straggler shard can neither resurrect an abandoned frame nor re-open an +/// emitted one. Must cover at least [`HARD_LOSS_WINDOW`]: stragglers can trickle in later than the +/// loss verdict. +const REORDER_WINDOW: u32 = 64; /// Fixed per-packet header. `#[repr(C)]`, no padding, zero-copy (de)serializable. #[repr(C)] @@ -274,7 +295,10 @@ pub struct Reassembler { /// Recently-emitted frames, so stray/late shards can't resurrect them. Pruned to /// the reorder window alongside `frames`. completed: HashSet, - newest_frame: Option, + /// The newest frame seen, as `(frame_index, capture pts)` — the loss-window anchor: an + /// incomplete frame is declared lost once it sits [`LOSS_WINDOW_NS`] behind this pts (or + /// [`HARD_LOSS_WINDOW`] indices, whichever trips first). + newest_frame: Option<(u32, u64)>, } impl Reassembler { @@ -344,12 +368,12 @@ impl Reassembler { } let payload = pkt[HEADER_LEN..HEADER_LEN + shard_bytes].to_vec(); - self.advance_window(hdr.frame_index, stats); + self.advance_window(hdr.frame_index, hdr.pts_ns, stats); // Drop shards for frames we've already emitted (e.g. the recovery shards of a // frame that completed early via the all-originals-present fast path) or that - // have fallen out of the reorder window. - if self.completed.contains(&hdr.frame_index) || self.is_stale(hdr.frame_index) { + // have fallen out of the loss window. + if self.completed.contains(&hdr.frame_index) || self.is_stale(hdr.frame_index, hdr.pts_ns) { drop(stats); return Ok(None); } @@ -461,19 +485,31 @@ impl Reassembler { Ok(None) } - /// Track the newest frame and prune stragglers that fell out of the reorder window - /// (counting them as dropped). - fn advance_window(&mut self, frame_index: u32, stats: &StatsCounters) { - let newest = match self.newest_frame { + /// Track the newest frame, declare incomplete frames that fell out of the loss window + /// ([`LOSS_WINDOW_NS`] behind the newest pts, or [`HARD_LOSS_WINDOW`] indices) lost — counting + /// them dropped, which is what drives the client's recovery-keyframe request — and prune the + /// completed-index memory to [`REORDER_WINDOW`]. + fn advance_window(&mut self, frame_index: u32, pts_ns: u64, stats: &StatsCounters) { + let (newest, newest_pts) = match self.newest_frame { // `frame_index` is newer iff it's within the forward half of the index space. - Some(n) if frame_index.wrapping_sub(n) > u32::MAX / 2 => n, - _ => frame_index, + Some((n, p)) if frame_index.wrapping_sub(n) > u32::MAX / 2 => (n, p), + _ => (frame_index, pts_ns), }; - self.newest_frame = Some(newest); + self.newest_frame = Some((newest, newest_pts)); let before = self.frames.len(); - self.frames - .retain(|&idx, _| newest.wrapping_sub(idx) <= REORDER_WINDOW); + let completed = &mut self.completed; + self.frames.retain(|&idx, f| { + let keep = newest.wrapping_sub(idx) <= HARD_LOSS_WINDOW + && newest_pts.saturating_sub(f.pts_ns) <= LOSS_WINDOW_NS; + if !keep { + // Remember the abandoned index so a straggler shard is dropped (below, and in + // `push`) instead of resurrecting the frame — which would re-allocate its buffers + // and double-count the drop when it aged out again. + completed.insert(idx); + } + keep + }); let pruned = before - self.frames.len(); if pruned > 0 { StatsCounters::add(&stats.frames_dropped, pruned as u64); @@ -482,13 +518,29 @@ impl Reassembler { .retain(|&idx| newest.wrapping_sub(idx) <= REORDER_WINDOW); } - /// True if `frame_index` lies behind the newest frame by more than the reorder - /// window (so its shards arrive too late to be useful). - fn is_stale(&self, frame_index: u32) -> bool { + /// Drop all in-flight state — every partially-assembled frame and the completed/abandoned + /// index memory — as if the session just started. Used by the client's backlog flush + /// ([`Session::flush_backlog`](crate::session::Session::flush_backlog)): after the socket + /// backlog is discarded wholesale, the partial frames here can never complete (their remaining + /// shards were just thrown away) and the window anchor (`newest_frame`) points into the + /// discarded past. + pub fn reset(&mut self) { + self.frames.clear(); + self.completed.clear(); + self.newest_frame = None; + } + + /// True if this packet's frame lies outside the loss window (behind the newest frame by more + /// than [`LOSS_WINDOW_NS`] of capture time or [`HARD_LOSS_WINDOW`] indices) — its shards + /// arrive too late to be useful, and accepting one would only create a frame buffer the next + /// [`advance_window`] immediately declares lost. + fn is_stale(&self, frame_index: u32, pts_ns: u64) -> bool { match self.newest_frame { - Some(n) => { + Some((n, newest_pts)) => { let behind = n.wrapping_sub(frame_index); - behind > REORDER_WINDOW && behind <= u32::MAX / 2 + behind <= u32::MAX / 2 + && (behind > HARD_LOSS_WINDOW + || newest_pts.saturating_sub(pts_ns) > LOSS_WINDOW_NS) } None => false, } @@ -585,6 +637,82 @@ mod tests { assert_eq!(stats.snapshot().packets_dropped, 1); } + /// The loss window is TIME-based: an incomplete frame survives newer frames arriving within + /// [`LOSS_WINDOW_NS`] of its capture pts (a 33 ms-late shard at 120 fps is late, not lost — + /// the old 4-INDEX window wrongly killed it), is declared lost once the newest pts moves past + /// the window (`frames_dropped`), and a straggler shard can't resurrect it afterwards. + #[test] + fn incomplete_frames_age_out_by_capture_time_not_frame_count() { + let mut r = Reassembler::new(limits()); + let coder = coder_for(FecScheme::Gf8); + let stats = StatsCounters::default(); + const FRAME_NS: u64 = 8_333_333; // 120 fps + + // Frame 0: one of its two shards arrives — incomplete. + let mut h = base_header(); + h.data_shards = 2; + h.frame_bytes = 32; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + + // Frames 1..=8 complete around it (well past the old 4-index window, inside 120 ms): + // frame 0 must still be alive — no drop counted. + for i in 1..=8u32 { + let mut h = base_header(); + h.frame_index = i; + h.pts_ns = i as u64 * FRAME_NS; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_some()); + } + assert_eq!(stats.snapshot().frames_dropped, 0); + + // Frame 0's second shard arrives 8 frames late (~66 ms at 120 fps) — completes fine. + let mut h = base_header(); + h.data_shards = 2; + h.frame_bytes = 32; + h.shard_index = 1; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_some()); + + // Frame 20: incomplete again; then a frame lands past the 120 ms window → declared lost. + let mut h = base_header(); + h.frame_index = 20; + h.pts_ns = 20 * FRAME_NS; + h.data_shards = 2; + h.frame_bytes = 32; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + let mut h = base_header(); + h.frame_index = 21; + h.pts_ns = 20 * FRAME_NS + LOSS_WINDOW_NS + 1; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_some()); + assert_eq!(stats.snapshot().frames_dropped, 1); + + // A straggler shard for the abandoned frame 20 is dropped, never resurrected. + let mut h = base_header(); + h.frame_index = 20; + h.pts_ns = 20 * FRAME_NS; + h.data_shards = 2; + h.frame_bytes = 32; + h.shard_index = 1; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + assert_eq!(stats.snapshot().frames_dropped, 1, "no double-count"); + } + #[test] fn rejects_wrong_shard_bytes_and_oversized_frame() { let coder = coder_for(FecScheme::Gf8); diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 450f25ca..feafa28f 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -290,6 +290,45 @@ impl Session { } } + /// Client: discard the ENTIRE pending receive backlog — the current recv ring plus everything + /// queued in the kernel socket buffer — and reset the reassembler. Returns how many datagrams + /// were thrown away (counted into `packets_dropped`). + /// + /// This is the latency-bound escape hatch: the receive path has no other way to skip ahead. + /// Packets arrive strictly in order, so once a standing queue forms (the pump transiently + /// slower than the wire, a Wi-Fi stall, power-save delivery clumping), the client plays that + /// far behind FOREVER — it consumes at exactly the arrival rate, so the backlog never shrinks + /// (observed live: a stream stuck 6–7 s behind, socket buffers full end to end). Discarding + /// is memcpy-speed (no decrypt/reassembly/allocation), so this empties even a 32 MB buffer in + /// milliseconds; the caller then requests a keyframe and the stream resumes live. The iteration + /// cap (4096 batches ≈ 128k datagrams ≈ 190 MB) only guards against a line-rate sender + /// outpacing the discard loop indefinitely. + pub fn flush_backlog(&mut self) -> Result { + if self.config.role != Role::Client { + return Err(PunktfunkError::InvalidArg( + "flush_backlog called on a host session", + )); + } + // The undelivered tail of the current ring is backlog too. + let mut flushed = self.recv_count.saturating_sub(self.recv_idx) as u64; + self.recv_count = 0; + self.recv_idx = 0; + if !self.recv_scratch.is_empty() { + for _ in 0..4096 { + let n = self + .transport + .recv_batch(&mut self.recv_scratch, &mut self.recv_lens)?; + if n == 0 { + break; + } + flushed += n as u64; + } + } + self.reassembler.reset(); + StatsCounters::add(&self.stats.packets_dropped, flushed); + Ok(flushed) + } + /// Client: serialize and send one input event to the host. pub fn send_input(&mut self, event: &InputEvent) -> Result<()> { if self.config.role != Role::Client { diff --git a/crates/punktfunk-core/src/transport/udp.rs b/crates/punktfunk-core/src/transport/udp.rs index 0589edd0..0c6b48ce 100644 --- a/crates/punktfunk-core/src/transport/udp.rs +++ b/crates/punktfunk-core/src/transport/udp.rs @@ -1,10 +1,12 @@ //! Real UDP datagram transport — native sockets, no async runtime. //! //! Send is batched via `sendmmsg` ([`Transport::send_batch`], ≤64/syscall) and recv via `recvmmsg` -//! ([`Transport::recv_batch`], ≤32/syscall into a reused ring) — the 1 Gbps+ syscall lever -//! (~125k → a few-k syscalls/sec at line rate). The host additionally paces each frame's send -//! across the frame interval (see `punktfunk1.rs::paced_submit`) so a real NIC doesn't drop a line-rate -//! burst. All three layer on this same [`Transport`] seam (scalar fallbacks for loopback/non-Linux). +//! ([`Transport::recv_batch`], ≤32/syscall into a reused ring) on Linux AND Android (which is +//! `target_os = "android"`, not `"linux"` — it needs its own bionic binding, see [`android_mmsg`]) +//! — the 1 Gbps+ syscall lever (~125k → a few-k syscalls/sec at line rate). The host additionally +//! paces each frame's send across the frame interval (see `punktfunk1.rs::paced_submit`) so a real +//! NIC doesn't drop a line-rate burst. All three layer on this same [`Transport`] seam (scalar +//! fallbacks for loopback and the remaining targets). use super::Transport; use crate::packet::MAX_DATAGRAM_BYTES; @@ -57,16 +59,51 @@ fn is_transient_io(e: &std::io::Error) -> bool { } } +/// `sendmmsg`/`recvmmsg` + `mmsghdr` for Android, where the `libc` crate binds only the syscall +/// number (`SYS_recvmmsg`) and neither the wrapper functions nor the struct — even though bionic +/// has exported both since API 21 (below our API-28 floor), and Rust's `target_os = "android"` is +/// NOT `"linux"`, so the batched paths below silently excluded Android and the client fell back to +/// one syscall per datagram. The struct layout is stable kernel ABI (`struct mmsghdr` in +/// `linux/socket.h`): a `msghdr` followed by the received byte count. +#[cfg(target_os = "android")] +mod android_mmsg { + #[repr(C)] + #[allow(non_camel_case_types)] + pub struct mmsghdr { + pub msg_hdr: libc::msghdr, + pub msg_len: libc::c_uint, + } + extern "C" { + pub fn sendmmsg( + sockfd: libc::c_int, + msgvec: *mut mmsghdr, + vlen: libc::c_uint, + flags: libc::c_int, + ) -> libc::c_int; + pub fn recvmmsg( + sockfd: libc::c_int, + msgvec: *mut mmsghdr, + vlen: libc::c_uint, + flags: libc::c_int, + timeout: *mut libc::timespec, + ) -> libc::c_int; + } +} +#[cfg(target_os = "android")] +use android_mmsg::{mmsghdr, recvmmsg, sendmmsg}; +#[cfg(target_os = "linux")] +use libc::{mmsghdr, recvmmsg, sendmmsg}; + /// Build one `mmsghdr` per `iovec` (each a single-buffer message) for `sendmmsg`/`recvmmsg`. Shared /// by `send_batch` + `recv_batch` so the raw-pointer scaffolding lives in exactly one place. /// /// SAFETY (caller's): each returned header holds a raw pointer into `iovs`; the caller MUST keep /// `iovs` alive and unmoved for as long as the headers are passed to the syscall. -#[cfg(target_os = "linux")] -fn mmsghdrs(iovs: &mut [libc::iovec]) -> Vec { +#[cfg(any(target_os = "linux", target_os = "android"))] +fn mmsghdrs(iovs: &mut [libc::iovec]) -> Vec { iovs.iter_mut() .map(|iov| { - let mut h: libc::mmsghdr = unsafe { std::mem::zeroed() }; + let mut h: mmsghdr = unsafe { std::mem::zeroed() }; h.msg_hdr.msg_iov = iov; h.msg_hdr.msg_iovlen = 1; h @@ -575,9 +612,9 @@ impl Transport for UdpTransport { /// no per-message address. The socket is non-blocking, so a full send buffer surfaces as a /// short count (or `EAGAIN` with nothing sent); we stop and report what went out rather than /// block or retry — the data plane is lossy + FEC-protected, and blocking would queue stale - /// frames + add latency. Ports the proven GameStream `sendmmsg_all`. Non-Linux falls back to - /// the trait's scalar `send` loop (no `sendmmsg`). - #[cfg(target_os = "linux")] + /// frames + add latency. Ports the proven GameStream `sendmmsg_all`. Other targets fall back + /// to the trait's scalar `send` loop (no `sendmmsg`). + #[cfg(any(target_os = "linux", target_os = "android"))] fn send_batch(&self, packets: &[&[u8]]) -> std::io::Result { use std::os::fd::AsRawFd; const CHUNK: usize = 64; @@ -593,7 +630,7 @@ impl Transport for UdpTransport { }) .collect(); let mut hdrs = mmsghdrs(&mut iovs); - let n = unsafe { libc::sendmmsg(fd, hdrs.as_mut_ptr(), hdrs.len() as libc::c_uint, 0) }; + let n = unsafe { sendmmsg(fd, hdrs.as_mut_ptr(), hdrs.len() as libc::c_uint, 0) }; if n < 0 { let err = std::io::Error::last_os_error(); // Nothing fit in the send buffer (or a stale ICMP from a connected-socket blip) — @@ -723,9 +760,9 @@ impl Transport for UdpTransport { /// caller's reused buffers (no per-packet allocation). `MSG_DONTWAIT` keeps it non-blocking /// (the socket already is); `EAGAIN` → `0`. A datagram larger than a buffer is truncated and /// `lens[i]` reaches the buffer size — the reassembler then rejects it as malformed, matching - /// `recv`'s oversized-drop. Apple/BSD use the `recv`-loop override below; other non-unix the - /// trait's scalar default. - #[cfg(target_os = "linux")] + /// `recv`'s oversized-drop. Android uses the local bionic binding (see [`android_mmsg`]). + /// Apple/BSD use the `recv`-loop override below; other non-unix the trait's scalar default. + #[cfg(any(target_os = "linux", target_os = "android"))] fn recv_batch(&self, out: &mut [Vec], lens: &mut [usize]) -> std::io::Result { use std::os::fd::AsRawFd; let fd = self.socket.as_raw_fd(); @@ -743,7 +780,7 @@ impl Transport for UdpTransport { .collect(); let mut hdrs = mmsghdrs(&mut iovs); let n = unsafe { - libc::recvmmsg( + recvmmsg( fd, hdrs.as_mut_ptr(), n_bufs as libc::c_uint, @@ -772,7 +809,7 @@ impl Transport for UdpTransport { /// batches; our client per-packet-allocated). It is still one syscall per datagram (a future /// `recvmsg_x` batch would cut that too); `EAGAIN` ends the drain. Oversized datagrams set /// `lens[i] == buf.len()` and the caller (`poll_frame`) drops them — same contract as `recvmmsg`. - #[cfg(all(unix, not(target_os = "linux")))] + #[cfg(all(unix, not(any(target_os = "linux", target_os = "android"))))] fn recv_batch(&self, out: &mut [Vec], lens: &mut [usize]) -> std::io::Result { // Apple: prefer the batched `recvmsg_x` syscall when enabled; a surprise error disables it // and falls through to the always-correct scalar loop below. diff --git a/crates/punktfunk-core/tests/loopback.rs b/crates/punktfunk-core/tests/loopback.rs index df8cabec..17e9aa77 100644 --- a/crates/punktfunk-core/tests/loopback.rs +++ b/crates/punktfunk-core/tests/loopback.rs @@ -112,6 +112,48 @@ fn lossless_stream_is_exact() { ); } +/// The client's latency-bound escape hatch: `flush_backlog` must discard every queued datagram +/// (counting them dropped), reset the reassembler so half-assembled frames from the flushed past +/// can't linger, and leave the session healthy — the next submitted frame recovers byte-exact. +#[test] +fn flush_backlog_discards_queue_and_recovers() { + let (host_tp, client_tp) = loopback_pair(0, 0); + let mut host = Session::new( + config(Role::Host, FecScheme::Gf16, false, 0), + Box::new(host_tp), + ) + .unwrap(); + let mut client = Session::new( + config(Role::Client, FecScheme::Gf16, false, 0), + Box::new(client_tp), + ) + .unwrap(); + + let frames = sample_frames(); + // Read one frame first so the client's recv ring exists and may hold an undelivered tail. + host.submit_frame(&frames[0], 0, 0).unwrap(); + client.poll_frame().unwrap(); + // Queue a multi-frame backlog, then flush it: everything pending is discarded. + for (i, f) in frames.iter().enumerate().skip(1) { + host.submit_frame(f, i as u64 * 1_000_000, 0).unwrap(); + } + let flushed = client.flush_backlog().unwrap(); + assert!(flushed > 0, "a queued backlog must be discarded"); + assert_eq!(client.stats().packets_dropped, flushed); + assert!( + matches!( + client.poll_frame(), + Err(punktfunk_core::PunktfunkError::NoFrame) + ), + "nothing pending after a flush" + ); + // The stream resumes cleanly: the next frame (the "recovery keyframe") completes byte-exact. + let recovery = vec![0xA5u8; 100_000]; + host.submit_frame(&recovery, 99_000_000, 0).unwrap(); + let got = client.poll_frame().expect("post-flush frame completes"); + assert_eq!(got.data, recovery); +} + #[test] fn input_round_trips_client_to_host() { let (host_tp, client_tp) = loopback_pair(0, 0); diff --git a/crates/punktfunk-host/src/punktfunk1.rs b/crates/punktfunk-host/src/punktfunk1.rs index 96ee03cb..9e8fca14 100644 --- a/crates/punktfunk-host/src/punktfunk1.rs +++ b/crates/punktfunk-host/src/punktfunk1.rs @@ -26,7 +26,9 @@ #![deny(clippy::undocumented_unsafe_blocks)] use anyhow::{anyhow, Context, Result}; -use punktfunk_core::config::{CompositorPref, FecConfig, FecScheme, GamepadPref, Role}; +use punktfunk_core::config::{ + mtu1500_shard_payload, CompositorPref, FecConfig, FecScheme, GamepadPref, Role, +}; use punktfunk_core::input::{InputEvent, InputKind}; use punktfunk_core::packet::{FLAG_PIC, FLAG_PROBE, FLAG_SOF}; use punktfunk_core::quic::{ @@ -969,11 +971,14 @@ async fn serve_session( fec_percent: fec_static_override().unwrap_or(FEC_ADAPTIVE_START), max_data_per_block: 4096, }, - // ~1452-byte payload keeps the IP datagram within a 1500 MTU (1452 + 40 header + 24 - // crypto + 8 IP/UDP ≈ 1500), vs the old 1200 — ~17% fewer packets for free, and an even - // size (FEC requires even shards). Negotiated, so the client follows. Jumbo (≈8900) is a - // future negotiated bump (needs MAX_DATAGRAM_BYTES raised + end-to-end 9000 MTU). - shard_payload: 1452, + // The largest even payload whose sealed datagram (header + shard + crypto) fits an + // unfragmented IPv4/UDP packet on a 1500 MTU — 1408, giving 1472 = the exact ceiling. + // The previous 1452 overshot it (its math forgot the header/crypto ride inside the UDP + // payload) and silently IP-fragmented EVERY video datagram, doubling per-datagram loss + // on Wi-Fi — the "100 Mbps badly fails on the phone" root cause. Negotiated, so the + // client follows. Jumbo (≈8900) is a future negotiated bump (needs MAX_DATAGRAM_BYTES + // raised + end-to-end 9000 MTU). + shard_payload: mtu1500_shard_payload() as u16, encrypt: true, key, salt: *b"pkf1", @@ -1092,8 +1097,18 @@ async fn serve_session( // send loop reads `fec_target_ctl` and applies it per frame. Ignored when FEC // is pinned via PUNKTFUNK_FEC_PCT. if adaptive_fec { - let target = adapt_fec(rep.loss_ppm); - let prev = fec_target_ctl.swap(target, Ordering::Relaxed); + // Fast attack, slow decay: jump straight to what the reported loss + // needs, but come DOWN only one point per clean report (~750 ms). The + // memoryless controller ping-ponged on periodic burst loss (Wi-Fi + // scans / BT coexistence, a burst every few seconds): a single clean + // window dropped FEC back to the floor, so every next burst hit an + // unprotected stream — an unrecoverable frame, a freeze, and a + // recovery-IDR burst, once per cycle. Decaying over ~10 windows keeps + // the stream covered across the gap while still converging to FEC_MIN + // on a genuinely clean link. + let prev = fec_target_ctl.load(Ordering::Relaxed); + let target = adapt_fec(rep.loss_ppm).max(prev.saturating_sub(1)); + fec_target_ctl.store(target, Ordering::Relaxed); if prev != target { tracing::info!( loss_ppm = rep.loss_ppm,