#!/usr/bin/env python3 """ Per-operation latency harness for the bench_latency_{implicit,explicit} fixtures. The fixtures emit a stdout marker line every PRINT_K bench operations. We spawn the binary with its stdout connected to a PTY so libc's printf path is line-buffered (rather than block-buffered as it would be on a pipe), record `time.monotonic_ns()` at every line read, and report the inter-arrival distribution. The headline numbers are median, p99, p99.9, max of the inter-arrival gaps (in microseconds). Tail latency is the question; total wall time is not. We do NOT count the gap between READY (8888) and the first sample — that gap includes the build_tree(19) cost and is not a steady-state operation. We also drop the very last gap (DONE 9999) because the program exit path can include final teardown that is not part of the loop budget. Usage: python3 bench/latency_harness.py [--label LABEL] """ from __future__ import annotations import argparse import os import pty import resource import select import sys import time def run_pty(binary: str, timeout_s: float = 600.0) -> tuple[list[int], int, int, int, float]: """Run `binary` with stdout on a PTY, return (timestamps_ns, lines, exit_status, max_rss_kb, wall_s).""" pid, fd = pty.fork() if pid == 0: # Child: replace argv[0] with the binary and exec. try: os.execvp(binary, [binary]) except FileNotFoundError as e: sys.stderr.write(f"exec failed: {e}\n") os._exit(127) # Parent: read from the PTY master, timestamp every newline. timestamps: list[int] = [] line_count = 0 buf = bytearray() deadline = time.monotonic() + timeout_s t_start = time.monotonic() try: while True: remaining = deadline - time.monotonic() if remaining <= 0: os.kill(pid, 9) raise TimeoutError(f"binary exceeded {timeout_s:.0f}s") r, _, _ = select.select([fd], [], [], min(remaining, 1.0)) if not r: continue try: chunk = os.read(fd, 4096) except OSError: # PTY master read after child exit -> EIO on Linux. break if not chunk: break now = time.monotonic_ns() buf.extend(chunk) # Walk the buffer; record `now` for each newline encountered. # Multiple newlines in one read share the same monotonic_ns # — that's a known quantization artefact of bursty stdio # delivery, NOT a per-op timing artefact. We surface it in # the report. nl = buf.find(b"\n") while nl != -1: timestamps.append(now) line_count += 1 del buf[: nl + 1] nl = buf.find(b"\n") finally: try: os.close(fd) except OSError: pass # Wait for child and capture rusage. _, status = os.waitpid(pid, 0) ru = resource.getrusage(resource.RUSAGE_CHILDREN) t_end = time.monotonic() return timestamps, line_count, status, ru.ru_maxrss, t_end - t_start def percentile(xs: list[float], p: float) -> float: """Nearest-rank percentile (xs assumed sorted).""" if not xs: return float("nan") k = max(0, min(len(xs) - 1, int(round((p / 100.0) * (len(xs) - 1))))) return xs[k] def report(label: str, timestamps_ns: list[int], lines: int, status: int, rss_kb: int, wall_s: float) -> None: # The fixtures emit: # line 1 : 8888 (READY, after build_tree) # line 2..1001 : per-chunk markers (1000 timing samples) # line 1002 : 9999 (DONE) # Inter-arrival gaps: between consecutive timestamps. # We drop the first gap (READY -> first chunk includes a partial # chunk's setup but is dominated by chunk work; still informative) # and the last gap (last chunk -> DONE includes program exit). # Per the orchestrator's brief we want the steady-state distribution, # so we trim both ends. if lines < 5: print(f"[{label}] FAIL: only {lines} lines, status={status}") return gaps_ns: list[int] = [] for i in range(1, len(timestamps_ns)): gaps_ns.append(timestamps_ns[i] - timestamps_ns[i - 1]) # Trim the first gap (READY -> first chunk; includes the loop's # first chunk's full work) and the last gap (last chunk -> DONE; # includes program teardown). Everything between is steady-state. trimmed = gaps_ns[1:-1] if len(gaps_ns) >= 3 else gaps_ns gaps_us = sorted(g / 1000.0 for g in trimmed) # Quantization detection: how many gaps are exactly 0 ns? That # means two newlines arrived in the same os.read. zero_gaps = sum(1 for g in trimmed if g == 0) median = percentile(gaps_us, 50) p90 = percentile(gaps_us, 90) p99 = percentile(gaps_us, 99) p999 = percentile(gaps_us, 99.9) p_max = max(gaps_us) if gaps_us else float("nan") p_min = min(gaps_us) if gaps_us else float("nan") mean = sum(gaps_us) / len(gaps_us) if gaps_us else float("nan") print(f"=== {label} ===") print(f" lines: {lines} exit_status: {status} wall: {wall_s:.3f}s max_rss: {rss_kb} KB") print(f" inter-arrival gaps (steady-state, n={len(trimmed)}, trimmed first+last):") print(f" min: {p_min:10.1f} us") print(f" median: {median:10.1f} us") print(f" mean: {mean:10.1f} us") print(f" p90: {p90:10.1f} us") print(f" p99: {p99:10.1f} us") print(f" p99.9: {p999:10.1f} us") print(f" max: {p_max:10.1f} us") print(f" p99/median: {p99 / median if median else float('nan'):.2f}x") print(f" max/median: {p_max / median if median else float('nan'):.2f}x") print(f" zero-ns gaps (read coalescing): {zero_gaps} / {len(trimmed)}") def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("binary", help="path to the bench binary") ap.add_argument("--label", default=None, help="label for the report") ap.add_argument("--timeout", type=float, default=600.0) args = ap.parse_args() label = args.label or os.path.basename(args.binary) timestamps, lines, status, rss_kb, wall_s = run_pty(args.binary, args.timeout) report(label, timestamps, lines, status, rss_kb, wall_s) return 0 if __name__ == "__main__": sys.exit(main())