bench: latency harness + paired latency fixtures
The latency_harness.py harness spawns the bench binary on a PTY, records monotonic_ns() per stdout line, and reports inter-arrival gap distribution (median / p99 / p99.9 / max). Tail latency is Decision 10's central real-time claim; total wall-time and RSS are the wrong metrics for that question. Paired fixtures: bench_latency_implicit (Boehm-fair, no mode annotations, leaks under --alloc=rc) and bench_latency_explicit (mode-annotated hot path, what RC was built for). Both use a depth-19 balanced tree (~16 MB) as the persistent live working set, plus per-op IntList build+sum churn forcing GC pressure under Boehm. Authored by ailang-bencher; ships evidence, not features.
This commit is contained in:
Executable
+172
@@ -0,0 +1,172 @@
|
||||
#!/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 <binary> [--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())
|
||||
Reference in New Issue
Block a user