From f2fa7828d6f88c5f03d738a5854b75a1f8866fef Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Tue, 14 Jul 2026 19:07:57 +0200 Subject: [PATCH] fix(probe,scripts): make speed-test sweeps work headless and tell the truth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three bugs found running the owed throughput sweeps (all three conspired to make yesterday's 'transport does 1G+' numbers fabrications): - the probe never advertised VIDEO_CAP_PROBE_SEQ, so every host DECLINED its speed tests; the zeroed decline reply divided a settle-window sliver by 1 ms and printed plausible-looking garbage. Advertise the cap (the shared-core reassembler windows probe-space frames) and detect the all-zero decline explicitly. - an idle virtual desktop publishes no frames on damage-driven capture (Windows IDD-push), so the pipeline build timed out before the burst could run. The probe now injects a ±2 px cursor wiggle over the wire during --speed-test warmup — injected host-side into the right session, works headless everywhere. - throughput-sweep.py: tracing emits ANSI color into pipes, which broke the key=value parser (crash on the first point); strip it, guard half-parsed lines, and surface host declines as a flag. Also logs the whole-run receive stage split (PUNKTFUNK_PERF) at stream end — the probe is the measurement tool for the client-pump wall. Co-Authored-By: Claude Fable 5 --- clients/probe/src/main.rs | 54 ++++++++++++++++++++++++++++++++++--- scripts/throughput-sweep.py | 14 ++++++++++ 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/clients/probe/src/main.rs b/clients/probe/src/main.rs index a5b372ff..7d587712 100644 --- a/clients/probe/src/main.rs +++ b/clients/probe/src/main.rs @@ -470,7 +470,10 @@ async fn session(args: Args) -> Result<()> { video_caps: { // Always ask for per-AU host timings (0xCF) — this is a measurement tool, and the // host/network split is exactly what it exists to report. Old hosts ignore the bit. - let mut caps = punktfunk_core::quic::VIDEO_CAP_HOST_TIMING; + // PROBE_SEQ: the shared-core reassembler windows probe-space frames, so the probe + // qualifies for `--speed-test` bursts; without the bit the host declines them. + let mut caps = punktfunk_core::quic::VIDEO_CAP_HOST_TIMING + | punktfunk_core::quic::VIDEO_CAP_PROBE_SEQ; if std::env::var_os("PUNKTFUNK_CLIENT_10BIT").is_some() { caps |= punktfunk_core::quic::VIDEO_CAP_10BIT; } @@ -639,11 +642,28 @@ async fn session(args: Args) -> Result<()> { } else { 0 }; + let conn2 = conn.clone(); tokio::spawn(async move { use std::sync::atomic::Ordering::Relaxed; - tokio::time::sleep(std::time::Duration::from_secs(2)).await; // let the stream warm up - // Baseline the packet-level counters right before the burst (video is paused during it, - // so the delta is pure probe traffic plus a sliver of resumed video in the settle). + // Warm up the stream — and generate desktop activity while doing so. Damage-driven + // capture paths (Windows IDD-push, a static headless desktop anywhere) publish NO + // frame until something composes, and the host's pipeline build waits for a first + // frame — so an idle virtual display would time the whole speed test out. A ±2 px + // cursor wiggle over the wire is injected host-side into the right session/desktop. + for i in 0..20u32 { + let mv = InputEvent { + kind: InputKind::MouseMove, + _pad: [0; 3], + code: 0, + x: if i % 2 == 0 { 2 } else { -2 }, + y: 0, + flags: 0, + }; + let _ = conn2.send_datagram(mv.encode().to_vec().into()); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + // Baseline the packet-level counters right before the burst (video is paused during it, + // so the delta is pure probe traffic plus a sliver of resumed video in the settle). let base_pkts = rxp.load(Relaxed); let base_bytes = rxb.load(Relaxed); tracing::info!(target_kbps, duration_ms, "requesting speed-test probe"); @@ -668,6 +688,15 @@ async fn session(args: Args) -> Result<()> { return; } }; + // A declined burst comes back all-zero (duration_ms = 0) — e.g. the host predates + // speed tests. Say so instead of dividing a settle-window sliver by 1 ms. + if res.duration_ms == 0 { + tracing::error!( + "SPEED TEST declined by host (all-zero ProbeResult) — host too old, or it \ + rejected the request; check the host log" + ); + return; + } // The reliable result can beat the last UDP shards — let the tail arrive before reading. // Keep this short: video resumes the instant the burst ends, so a long settle counts // resumed-video packets against the probe (inflating recv past the host's wire count). @@ -1242,6 +1271,23 @@ async fn session(args: Args) -> Result<()> { s.flush().ok(); } + // PUNKTFUNK_PERF: cumulative receive-path stage split for the whole run — where the + // receive core's time went (kernel drain vs AES-GCM open vs reassembly+FEC). This is + // the measurement tool's view of the client-pump wall the 2026-07-14 sweeps pinned. + if let Some(p) = session.take_pump_perf() { + let per_pkt = |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, + pkts_per_batch = p.packets.checked_div(p.batches.max(1)).unwrap_or(0), + decrypt_ns_pkt = per_pkt(p.decrypt_ns), + reasm_ns_pkt = per_pkt(p.reasm_ns), + "receive stage split (whole run, PUNKTFUNK_PERF)" + ); + } + latencies_us.sort_unstable(); let pct = |p: f64| -> u64 { if latencies_us.is_empty() { diff --git a/scripts/throughput-sweep.py b/scripts/throughput-sweep.py index 51207949..59789a93 100755 --- a/scripts/throughput-sweep.py +++ b/scripts/throughput-sweep.py @@ -78,6 +78,14 @@ SPEED_LINE = "SPEED TEST complete" # ---- output parsing ------------------------------------------------------- +ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + + +def strip_ansi(text: str) -> str: + """tracing emits ANSI color even into a pipe; strip it or `key=value` never matches.""" + return ANSI_RE.sub("", text) + + def _field(line: str, key: str) -> str | None: """Extract `key=value` or `key="value"` from a tracing log line (order-agnostic).""" m = re.search(rf'\b{re.escape(key)}=(?:"([^"]*)"|(\S+))', line) @@ -90,6 +98,7 @@ def parse_speed_line(text: str) -> dict | None: """Find the 'SPEED TEST complete' line in probe output and pull its fields. Field names mirror clients/probe/src/main.rs:698-708 exactly.""" + text = strip_ansi(text) line = next((ln for ln in text.splitlines() if SPEED_LINE in ln), None) if line is None: return None @@ -116,7 +125,10 @@ def parse_speed_line(text: str) -> dict | None: def scan_warnings(text: str) -> list[str]: """Surface client-side red flags that change interpretation of the numbers.""" + text = strip_ansi(text) flags = [] + if "SPEED TEST declined" in text: + flags.append("host declined the speed test (old host build?) — check the host log") if "UDP socket buffer capped" in text: flags.append("client SO_RCVBUF capped below target (raise kern.ipc.maxsockbuf)") if "falling back to per-packet sends" in text or "USO unsupported" in text: @@ -314,6 +326,8 @@ def main() -> int: print(f" -> {t} Mbps ...", end="", flush=True) parsed, flags, raw = run_point(bin_path, host, t, args) all_flags.update(flags) + if parsed is not None and parsed.get("delivered_mbps") is None: + parsed = None # SPEED TEST line present but unparseable — treat as a failed point if parsed is None: print(" FAIL") # surface why, briefly