Nils Andreas Svee
170f229e7b
Update README with comprehensive build/verification instructions and modify the client comparison script to consume CSV output instead of the legacy machine format.
1138 lines
38 KiB
Python
Executable File
1138 lines
38 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Run paired upstream irtt and irtt-rs client comparisons.
|
|
|
|
This is a black-box interoperability harness. It never reads upstream source; it
|
|
only invokes the upstream `irtt` executable and compares its captured output with
|
|
the local Rust client output.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import datetime as dt
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import shlex
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from statistics import mean, median
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_REMOTE_TARGETS = [
|
|
("netperf-eu.bufferbloat.net", "15s", "250ms", "remote-eu"),
|
|
("45.128.222.138", "15s", "250ms", "remote-singapore"),
|
|
]
|
|
DEFAULT_LOCAL_TARGET = ("127.0.0.1", "10s", "100ms", "local")
|
|
DEFAULT_OUTPUT_ROOT = Path("target/interop-comparisons")
|
|
DEFAULT_IRTT_RS_COMMAND = "target/debug/irtt-cli"
|
|
|
|
|
|
@dataclass
|
|
class CommandResult:
|
|
name: str
|
|
command: list[str]
|
|
started_at: str
|
|
finished_at: str
|
|
elapsed_seconds: float
|
|
returncode: int
|
|
stdout_path: str
|
|
stderr_path: str
|
|
extra_paths: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class PairResult:
|
|
upstream: CommandResult
|
|
rust: CommandResult
|
|
start_skew_ms: float
|
|
|
|
|
|
@dataclass
|
|
class Metrics:
|
|
source: str
|
|
sent: int | None = None
|
|
received: int | None = None
|
|
lost: int | None = None
|
|
loss_percent: float | None = None
|
|
rtt_min_us: float | None = None
|
|
rtt_mean_us: float | None = None
|
|
rtt_median_us: float | None = None
|
|
rtt_max_us: float | None = None
|
|
raw_rtt_min_us: float | None = None
|
|
raw_rtt_mean_us: float | None = None
|
|
raw_rtt_median_us: float | None = None
|
|
raw_rtt_max_us: float | None = None
|
|
adjusted_rtt_min_us: float | None = None
|
|
adjusted_rtt_mean_us: float | None = None
|
|
adjusted_rtt_median_us: float | None = None
|
|
adjusted_rtt_max_us: float | None = None
|
|
samples_path: str | None = None
|
|
first_send_wall_ns: int | None = None
|
|
parse_notes: list[str] = field(default_factory=list)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
key: value
|
|
for key, value in self.__dict__.items()
|
|
if value is not None and value != []
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class TestCase:
|
|
name: str
|
|
target: str
|
|
duration: str
|
|
interval: str
|
|
is_local: bool = False
|
|
|
|
|
|
def utc_now() -> str:
|
|
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="milliseconds")
|
|
|
|
|
|
def safe_name(value: str) -> str:
|
|
return re.sub(r"[^A-Za-z0-9_.-]+", "_", value).strip("_") or "target"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Compare upstream irtt and irtt-rs clients with concurrent runs."
|
|
)
|
|
parser.add_argument(
|
|
"--target",
|
|
action="append",
|
|
help=(
|
|
"Target host or host:port. Repeat to run multiple custom targets. "
|
|
"When omitted, the default remote targets plus localhost are used."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--duration",
|
|
help="Override all test durations, for example 10s or 1500ms.",
|
|
)
|
|
parser.add_argument(
|
|
"--interval",
|
|
help="Override all probe intervals, for example 250ms.",
|
|
)
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
default=str(DEFAULT_OUTPUT_ROOT),
|
|
help=f"Directory for run artifacts (default: {DEFAULT_OUTPUT_ROOT}).",
|
|
)
|
|
parser.add_argument(
|
|
"--remote-only",
|
|
action="store_true",
|
|
help="Run only remote/default or explicitly supplied non-local targets.",
|
|
)
|
|
parser.add_argument(
|
|
"--include-local",
|
|
action="store_true",
|
|
help="Also run the localhost test when custom --target values are supplied.",
|
|
)
|
|
parser.add_argument(
|
|
"--upstream-irtt",
|
|
default="irtt",
|
|
help="Path to the upstream irtt executable (default: irtt).",
|
|
)
|
|
parser.add_argument(
|
|
"--irtt-rs-command",
|
|
default=DEFAULT_IRTT_RS_COMMAND,
|
|
help=(
|
|
"Command prefix or direct path for the Rust client. Arguments for "
|
|
"target, duration, interval, and CSV event output are appended. "
|
|
f"Default: {DEFAULT_IRTT_RS_COMMAND!r}"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--client-command-prefix",
|
|
default="",
|
|
help=(
|
|
"Command prefix applied to both client invocations, for example "
|
|
"'sudo ip netns exec irtt-client'."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--skip-build",
|
|
action="store_true",
|
|
help="Do not run cargo build -p irtt-cli before comparisons.",
|
|
)
|
|
parser.add_argument(
|
|
"--localhost-port",
|
|
type=int,
|
|
default=2112,
|
|
help="Port used by the localhost upstream irtt server check (default: 2112).",
|
|
)
|
|
parser.add_argument(
|
|
"--upstream-extra-arg",
|
|
action="append",
|
|
default=[],
|
|
help="Extra argument passed to upstream `irtt client`; repeat as needed.",
|
|
)
|
|
parser.add_argument(
|
|
"--rust-extra-arg",
|
|
action="append",
|
|
default=[],
|
|
help="Extra argument passed to irtt-rs; repeat as needed.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def build_cases(args: argparse.Namespace) -> list[TestCase]:
|
|
cases: list[TestCase] = []
|
|
if args.target:
|
|
for target in args.target:
|
|
target_host, target_port = split_host_port(target)
|
|
is_local = target_host in {"127.0.0.1", "localhost", "::1"}
|
|
if args.remote_only and is_local:
|
|
continue
|
|
cases.append(
|
|
TestCase(
|
|
name=safe_name(target),
|
|
target=target,
|
|
duration=args.duration or "15s",
|
|
interval=args.interval or "250ms",
|
|
is_local=is_local,
|
|
)
|
|
)
|
|
if args.include_local and not args.remote_only:
|
|
host, duration, interval, name = DEFAULT_LOCAL_TARGET
|
|
cases.append(
|
|
TestCase(
|
|
name=name,
|
|
target=f"{host}:{args.localhost_port}",
|
|
duration=args.duration or duration,
|
|
interval=args.interval or interval,
|
|
is_local=True,
|
|
)
|
|
)
|
|
else:
|
|
for host, duration, interval, name in DEFAULT_REMOTE_TARGETS:
|
|
cases.append(
|
|
TestCase(
|
|
name=name,
|
|
target=host,
|
|
duration=args.duration or duration,
|
|
interval=args.interval or interval,
|
|
)
|
|
)
|
|
if not args.remote_only:
|
|
host, duration, interval, name = DEFAULT_LOCAL_TARGET
|
|
cases.append(
|
|
TestCase(
|
|
name=name,
|
|
target=f"{host}:{args.localhost_port}",
|
|
duration=args.duration or duration,
|
|
interval=args.interval or interval,
|
|
is_local=True,
|
|
)
|
|
)
|
|
return cases
|
|
|
|
|
|
def split_host_port(target: str) -> tuple[str, int | None]:
|
|
if target.startswith("["):
|
|
match = re.match(r"^\[([^\]]+)\](?::(\d+))?$", target)
|
|
if match:
|
|
port = int(match.group(2)) if match.group(2) else None
|
|
return match.group(1), port
|
|
return target, None
|
|
if target.count(":") == 1:
|
|
host, port = target.rsplit(":", 1)
|
|
if port.isdigit():
|
|
return host, int(port)
|
|
return target, None
|
|
|
|
|
|
def local_instruction(port: int, upstream: str) -> str:
|
|
return (
|
|
f"localhost:{port} is not reachable by upstream irtt no-test. "
|
|
"Start a local upstream server manually, for example:\n\n"
|
|
f" {shlex.quote(upstream)} server -b 127.0.0.1:{port} --tstamp=dual\n\n"
|
|
"Then rerun this harness, or pass --remote-only to intentionally skip "
|
|
"the local test."
|
|
)
|
|
|
|
|
|
def classify_local_preflight_failure(
|
|
exc: BaseException | None, stdout: str = "", stderr: str = ""
|
|
) -> str:
|
|
text = f"{exc or ''}\n{stdout}\n{stderr}".lower()
|
|
if isinstance(exc, subprocess.TimeoutExpired) or any(
|
|
marker in text for marker in ["timed out", "timeout", "i/o timeout", "deadline exceeded"]
|
|
):
|
|
return "timeout: no response/firewall"
|
|
if any(
|
|
marker in text
|
|
for marker in [
|
|
"connection refused",
|
|
"connection reset by peer",
|
|
"connect: refused",
|
|
"actively refused",
|
|
]
|
|
):
|
|
return "connection refused: likely no local server"
|
|
if isinstance(exc, PermissionError) or any(
|
|
marker in text
|
|
for marker in [
|
|
"permission denied",
|
|
"operation not permitted",
|
|
"address already in use",
|
|
"bind: permission",
|
|
"bind: address",
|
|
]
|
|
):
|
|
return "permission/bind error: local environment permission issue"
|
|
return "other: unknown local preflight failure"
|
|
|
|
|
|
def check_local_server(args: argparse.Namespace) -> tuple[bool, str]:
|
|
target = f"127.0.0.1:{args.localhost_port}"
|
|
command = [
|
|
args.upstream_irtt,
|
|
"client",
|
|
"-n",
|
|
"-Q",
|
|
"--timeouts=1s",
|
|
target,
|
|
]
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
timeout=3,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
classification = classify_local_preflight_failure(exc)
|
|
return (
|
|
False,
|
|
f"{local_instruction(args.localhost_port, args.upstream_irtt)}\n"
|
|
f"check classification: {classification}\n"
|
|
f"check error: {exc}",
|
|
)
|
|
if completed.returncode == 0:
|
|
return True, ""
|
|
details = (completed.stderr or completed.stdout or "").strip()
|
|
classification = classify_local_preflight_failure(
|
|
None, completed.stdout or "", completed.stderr or ""
|
|
)
|
|
suffix = f"\ncheck output:\n{details}" if details else ""
|
|
return (
|
|
False,
|
|
local_instruction(args.localhost_port, args.upstream_irtt)
|
|
+ f"\ncheck classification: {classification}"
|
|
+ suffix,
|
|
)
|
|
|
|
|
|
def preflight_build_irtt_cli(run_dir: Path) -> bool:
|
|
command = ["cargo", "build", "-p", "irtt-cli"]
|
|
started = utc_now()
|
|
print(f"[{started}] preflight build: {shlex.join(command)}", flush=True)
|
|
completed = subprocess.run(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
(run_dir / "preflight-build.stdout").write_text(completed.stdout, encoding="utf-8")
|
|
(run_dir / "preflight-build.stderr").write_text(completed.stderr, encoding="utf-8")
|
|
if completed.returncode != 0:
|
|
print(
|
|
f"preflight build failed with exit {completed.returncode}; "
|
|
f"see {run_dir / 'preflight-build.stderr'}",
|
|
file=sys.stderr,
|
|
)
|
|
return False
|
|
return True
|
|
|
|
|
|
def run_case(
|
|
case: TestCase, run_dir: Path, args: argparse.Namespace
|
|
) -> tuple[dict[str, Any], bool]:
|
|
case_dir = run_dir / case.name
|
|
case_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
upstream_json = case_dir / "upstream.json"
|
|
client_prefix = shlex.split(args.client_command_prefix)
|
|
upstream_cmd = [
|
|
*client_prefix,
|
|
args.upstream_irtt,
|
|
"client",
|
|
"-d",
|
|
case.duration,
|
|
"-i",
|
|
case.interval,
|
|
"-q",
|
|
"-o",
|
|
str(upstream_json),
|
|
*args.upstream_extra_arg,
|
|
case.target,
|
|
]
|
|
rust_cmd = [
|
|
*client_prefix,
|
|
*shlex.split(args.irtt_rs_command),
|
|
case.target,
|
|
"--duration",
|
|
case.duration,
|
|
"--interval",
|
|
case.interval,
|
|
"--format",
|
|
"csv",
|
|
"--columns",
|
|
"event,seq,effective_rtt_us,raw_rtt_us,adjusted_rtt_us,server_processing_us,client_send_wall_ns",
|
|
*args.rust_extra_arg,
|
|
]
|
|
|
|
print(
|
|
f"[{utc_now()}] running {case.name}: target={case.target} "
|
|
f"duration={case.duration} interval={case.interval}",
|
|
flush=True,
|
|
)
|
|
pair_result = run_pair(case_dir, upstream_cmd, rust_cmd)
|
|
upstream_result = pair_result.upstream
|
|
rust_result = pair_result.rust
|
|
|
|
rust_samples = case_dir / "irtt-rs-rtt-samples.csv"
|
|
upstream_samples = case_dir / "upstream-rtt-samples.csv"
|
|
rust_metrics = parse_rust_csv(case_dir / "irtt-rs.stdout", rust_samples)
|
|
upstream_metrics = parse_upstream(case_dir / "upstream.stdout", upstream_json, upstream_samples)
|
|
|
|
comparison = classify_difference(upstream_metrics, rust_metrics)
|
|
first_send_skew_ms = paired_first_send_skew_ms(upstream_metrics, rust_metrics)
|
|
result = {
|
|
"case": case.__dict__,
|
|
"upstream": upstream_result.__dict__,
|
|
"irtt_rs": rust_result.__dict__,
|
|
"paired_start_skew_ms": pair_result.start_skew_ms,
|
|
"paired_first_send_skew_ms": first_send_skew_ms,
|
|
"metrics": {
|
|
"upstream": upstream_metrics.as_dict(),
|
|
"irtt_rs": rust_metrics.as_dict(),
|
|
},
|
|
"comparison": comparison,
|
|
}
|
|
|
|
write_json(case_dir / "comparison.json", result)
|
|
write_case_summary(case_dir / "summary.md", result)
|
|
ok = upstream_result.returncode == 0 and rust_result.returncode == 0
|
|
return result, ok
|
|
|
|
|
|
def paired_first_send_skew_ms(upstream: Metrics, rust: Metrics) -> float | None:
|
|
if upstream.first_send_wall_ns is None or rust.first_send_wall_ns is None:
|
|
return None
|
|
return (rust.first_send_wall_ns - upstream.first_send_wall_ns) / 1_000_000.0
|
|
|
|
|
|
def run_pair(
|
|
case_dir: Path, upstream_cmd: list[str], rust_cmd: list[str]
|
|
) -> PairResult:
|
|
(case_dir / "commands.txt").write_text(
|
|
"upstream: " + shlex.join(upstream_cmd) + "\n"
|
|
"irtt-rs: " + shlex.join(rust_cmd) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
upstream_stdout = case_dir / "upstream.stdout"
|
|
upstream_stderr = case_dir / "upstream.stderr"
|
|
rust_stdout = case_dir / "irtt-rs.stdout"
|
|
rust_stderr = case_dir / "irtt-rs.stderr"
|
|
|
|
upstream_out = upstream_stdout.open("wb")
|
|
upstream_err = upstream_stderr.open("wb")
|
|
upstream_started = utc_now()
|
|
upstream_start_mono = time.monotonic()
|
|
try:
|
|
upstream_proc = subprocess.Popen(
|
|
upstream_cmd,
|
|
stdout=upstream_out,
|
|
stderr=upstream_err,
|
|
)
|
|
except OSError as exc:
|
|
upstream_out.close()
|
|
upstream_err.close()
|
|
upstream_stderr.write_text(f"failed to start upstream irtt: {exc}\n", encoding="utf-8")
|
|
upstream_proc = None
|
|
|
|
rust_out = rust_stdout.open("wb")
|
|
rust_err = rust_stderr.open("wb")
|
|
rust_started = utc_now()
|
|
rust_start_mono = time.monotonic()
|
|
try:
|
|
rust_proc = subprocess.Popen(
|
|
rust_cmd,
|
|
stdout=rust_out,
|
|
stderr=rust_err,
|
|
)
|
|
except OSError as exc:
|
|
rust_out.close()
|
|
rust_err.close()
|
|
rust_stderr.write_text(f"failed to start irtt-rs: {exc}\n", encoding="utf-8")
|
|
rust_proc = None
|
|
|
|
upstream_rc = wait_process(upstream_proc)
|
|
upstream_finished = utc_now()
|
|
upstream_elapsed = time.monotonic() - upstream_start_mono
|
|
rust_rc = wait_process(rust_proc)
|
|
rust_finished = utc_now()
|
|
rust_elapsed = time.monotonic() - rust_start_mono
|
|
for handle in [upstream_out, upstream_err, rust_out, rust_err]:
|
|
try:
|
|
handle.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return PairResult(
|
|
upstream=CommandResult(
|
|
name="upstream",
|
|
command=upstream_cmd,
|
|
started_at=upstream_started,
|
|
finished_at=upstream_finished,
|
|
elapsed_seconds=upstream_elapsed,
|
|
returncode=upstream_rc,
|
|
stdout_path=str(upstream_stdout),
|
|
stderr_path=str(upstream_stderr),
|
|
extra_paths={"json": str(case_dir / "upstream.json")},
|
|
),
|
|
rust=CommandResult(
|
|
name="irtt-rs",
|
|
command=rust_cmd,
|
|
started_at=rust_started,
|
|
finished_at=rust_finished,
|
|
elapsed_seconds=rust_elapsed,
|
|
returncode=rust_rc,
|
|
stdout_path=str(rust_stdout),
|
|
stderr_path=str(rust_stderr),
|
|
),
|
|
start_skew_ms=(rust_start_mono - upstream_start_mono) * 1000.0,
|
|
)
|
|
|
|
|
|
def wait_process(proc: subprocess.Popen[bytes] | None) -> int:
|
|
if proc is None:
|
|
return 127
|
|
try:
|
|
return proc.wait()
|
|
except KeyboardInterrupt:
|
|
proc.terminate()
|
|
try:
|
|
return proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
return proc.wait()
|
|
|
|
|
|
def parse_rust_csv(path: Path, samples_path: Path) -> Metrics:
|
|
metrics = Metrics(source="irtt-rs", samples_path=str(samples_path))
|
|
effective: list[float] = []
|
|
raw: list[float] = []
|
|
adjusted: list[float] = []
|
|
sent_seqs: set[int] = set()
|
|
received_seqs: set[int] = set()
|
|
lost = 0
|
|
rows = ["seq,effective_rtt_us,raw_rtt_us,adjusted_rtt_us,server_processing_us\n"]
|
|
|
|
if not path.exists():
|
|
metrics.parse_notes.append(f"missing {path}")
|
|
return metrics
|
|
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
for fields in csv.DictReader(line for line in text.splitlines() if line.strip()):
|
|
event = fields.get("event")
|
|
if event is None:
|
|
continue
|
|
seq = parse_int(fields.get("seq"))
|
|
if event == "echo_reply":
|
|
if seq is not None:
|
|
received_seqs.add(seq)
|
|
sent_seqs.add(seq)
|
|
effective_value = parse_float(fields.get("effective_rtt_us"))
|
|
raw_value = parse_float(fields.get("raw_rtt_us"))
|
|
adjusted_value = parse_float(fields.get("adjusted_rtt_us"))
|
|
server_processing = parse_float(fields.get("server_processing_us"))
|
|
append_if_number(effective, effective_value)
|
|
append_if_number(raw, raw_value)
|
|
append_if_number(adjusted, adjusted_value)
|
|
if metrics.first_send_wall_ns is None:
|
|
metrics.first_send_wall_ns = parse_int(fields.get("client_send_wall_ns"))
|
|
rows.append(
|
|
f"{blank_none(seq)},{blank_none(effective_value)},{blank_none(raw_value)},"
|
|
f"{blank_none(adjusted_value)},{blank_none(server_processing)}\n"
|
|
)
|
|
elif event == "loss":
|
|
lost += 1
|
|
if seq is not None:
|
|
sent_seqs.add(seq)
|
|
elif event in {"duplicate", "late"}:
|
|
metrics.parse_notes.append(f"excluded {event} event from primary RTT stats")
|
|
|
|
samples_path.write_text("".join(rows), encoding="utf-8")
|
|
metrics.sent = len(sent_seqs) if sent_seqs else None
|
|
metrics.received = len(received_seqs)
|
|
metrics.lost = lost if lost else (
|
|
metrics.sent - metrics.received if metrics.sent is not None else None
|
|
)
|
|
if metrics.sent:
|
|
metrics.loss_percent = 100.0 * (metrics.lost or 0) / metrics.sent
|
|
fill_stats(metrics, effective, "rtt")
|
|
fill_stats(metrics, raw, "raw_rtt")
|
|
fill_stats(metrics, adjusted, "adjusted_rtt")
|
|
if not effective:
|
|
metrics.parse_notes.append("no echo_reply effective_rtt_us samples parsed")
|
|
return metrics
|
|
|
|
|
|
def parse_upstream(stdout_path: Path, json_path: Path, samples_path: Path) -> Metrics:
|
|
metrics = Metrics(source="upstream", samples_path=str(samples_path))
|
|
samples: list[float] = []
|
|
if json_path.exists():
|
|
try:
|
|
data = json.loads(json_path.read_text(encoding="utf-8", errors="replace"))
|
|
extract_upstream_json_metrics(data, metrics, samples)
|
|
except json.JSONDecodeError as exc:
|
|
metrics.parse_notes.append(f"failed to parse upstream JSON: {exc}")
|
|
else:
|
|
metrics.parse_notes.append(f"missing upstream JSON output: {json_path}")
|
|
|
|
if stdout_path.exists():
|
|
parse_upstream_text(stdout_path.read_text(encoding="utf-8", errors="replace"), metrics)
|
|
|
|
samples_path.write_text(
|
|
"rtt_us\n" + "".join(f"{value}\n" for value in samples),
|
|
encoding="utf-8",
|
|
)
|
|
if samples:
|
|
fill_stats(metrics, samples, "rtt")
|
|
if not samples and all(
|
|
value is None
|
|
for value in [
|
|
metrics.rtt_min_us,
|
|
metrics.rtt_mean_us,
|
|
metrics.rtt_median_us,
|
|
metrics.rtt_max_us,
|
|
]
|
|
):
|
|
metrics.parse_notes.append(
|
|
"upstream RTT samples/stats were not recognized; inspect upstream.json manually"
|
|
)
|
|
return metrics
|
|
|
|
|
|
def extract_upstream_json_metrics(data: Any, metrics: Metrics, samples: list[float]) -> None:
|
|
metrics.first_send_wall_ns = extract_upstream_first_send_wall_ns(data)
|
|
for key, value in walk_items(data):
|
|
key_norm = leaf_key(key)
|
|
if metrics.sent is None and key_norm in {
|
|
"packetssent",
|
|
"packetssend",
|
|
"sent",
|
|
"send",
|
|
"npackets",
|
|
}:
|
|
metrics.sent = parse_int(value)
|
|
elif metrics.received is None and key_norm in {
|
|
"packetsreceived",
|
|
"packetsrecv",
|
|
"received",
|
|
"recv",
|
|
}:
|
|
metrics.received = parse_int(value)
|
|
elif metrics.lost is None and key_norm in {"packetslost", "lost"}:
|
|
metrics.lost = parse_int(value)
|
|
elif metrics.loss_percent is None and key_norm in {
|
|
"packetlosspercent",
|
|
"losspercent",
|
|
"losspct",
|
|
}:
|
|
metrics.loss_percent = parse_float(value)
|
|
|
|
for key, value in walk_items(data):
|
|
if is_upstream_rtt_stats_key(key) and isinstance(value, dict):
|
|
apply_stat_dict(metrics, value, "rtt")
|
|
|
|
collect_upstream_samples(data, samples)
|
|
|
|
|
|
def walk_items(value: Any, parent_key: str = ""):
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
joined = f"{parent_key}.{key}" if parent_key else str(key)
|
|
yield joined, child
|
|
yield from walk_items(child, joined)
|
|
elif isinstance(value, list):
|
|
for index, child in enumerate(value):
|
|
yield from walk_items(child, f"{parent_key}[{index}]")
|
|
|
|
|
|
def normalize_key(value: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "", value.lower())
|
|
|
|
|
|
def leaf_key(value: str) -> str:
|
|
leaf = re.split(r"[.\[]", value)[-1].rstrip("]")
|
|
return normalize_key(leaf)
|
|
|
|
|
|
def is_upstream_rtt_stats_key(key: str) -> bool:
|
|
parts = [normalize_key(part) for part in key.split(".")]
|
|
if not parts:
|
|
return False
|
|
leaf = parts[-1]
|
|
if leaf not in {"rtt", "roundtrip", "roundtriptime"}:
|
|
return False
|
|
if len(parts) == 1:
|
|
return True
|
|
return parts[-2] in {"stats", "statistics", "summary"}
|
|
|
|
|
|
def apply_stat_dict(metrics: Metrics, value: dict[str, Any], prefix: str) -> None:
|
|
by_key = {normalize_key(str(key)): child for key, child in value.items()}
|
|
for attr, candidates in [
|
|
(f"{prefix}_min_us", ["min", "minimum"]),
|
|
(f"{prefix}_mean_us", ["mean", "avg", "average"]),
|
|
(f"{prefix}_median_us", ["median", "p50"]),
|
|
(f"{prefix}_max_us", ["max", "maximum"]),
|
|
]:
|
|
if getattr(metrics, attr) is not None:
|
|
continue
|
|
for candidate in candidates:
|
|
if candidate in by_key:
|
|
parsed = parse_duration_to_us(by_key[candidate])
|
|
if parsed is not None:
|
|
setattr(metrics, attr, parsed)
|
|
break
|
|
|
|
|
|
def collect_upstream_samples(value: Any, samples: list[float]) -> None:
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
if normalize_key(str(key)) == "roundtrips" and isinstance(child, list):
|
|
collect_upstream_round_trip_samples(child, samples)
|
|
else:
|
|
collect_upstream_samples(child, samples)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
collect_upstream_samples(child, samples)
|
|
|
|
|
|
def collect_upstream_round_trip_samples(round_trips: list[Any], samples: list[float]) -> None:
|
|
for item in round_trips:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if str(item.get("lost", "")).lower() == "true":
|
|
continue
|
|
delay = item.get("delay")
|
|
if not isinstance(delay, dict):
|
|
continue
|
|
parsed = parse_duration_to_us(delay.get("rtt"))
|
|
if parsed is not None:
|
|
samples.append(parsed)
|
|
|
|
|
|
def extract_upstream_first_send_wall_ns(data: Any) -> int | None:
|
|
round_trips = data.get("round_trips") if isinstance(data, dict) else None
|
|
if not isinstance(round_trips, list):
|
|
return None
|
|
for item in round_trips:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if str(item.get("lost", "")).lower() == "true":
|
|
continue
|
|
timestamps = item.get("timestamps")
|
|
if not isinstance(timestamps, dict):
|
|
continue
|
|
client = timestamps.get("client")
|
|
if not isinstance(client, dict):
|
|
continue
|
|
send = client.get("send")
|
|
if not isinstance(send, dict):
|
|
continue
|
|
wall = parse_int(send.get("wall"))
|
|
if wall is not None:
|
|
return wall
|
|
return None
|
|
|
|
|
|
def parse_upstream_text(text: str, metrics: Metrics) -> None:
|
|
packet_match = re.search(
|
|
r"(?P<sent>\d+)\s+packets?\s+sent.*?(?P<received>\d+)\s+"
|
|
r"(?:packets?\s+)?received.*?(?P<loss>[0-9.]+)%\s+"
|
|
r"(?:packet\s+)?loss",
|
|
text,
|
|
flags=re.IGNORECASE | re.DOTALL,
|
|
)
|
|
if packet_match:
|
|
metrics.sent = metrics.sent if metrics.sent is not None else int(packet_match.group("sent"))
|
|
metrics.received = (
|
|
metrics.received
|
|
if metrics.received is not None
|
|
else int(packet_match.group("received"))
|
|
)
|
|
metrics.loss_percent = (
|
|
metrics.loss_percent
|
|
if metrics.loss_percent is not None
|
|
else float(packet_match.group("loss"))
|
|
)
|
|
|
|
rtt_match = re.search(
|
|
r"(?:rtt|round[- ]trip).*?min/mean/(?:median/)?max.*?=\s*"
|
|
r"(?P<min>[0-9.]+)/(?P<mean>[0-9.]+)/(?:(?P<median>[0-9.]+)/)?(?P<max>[0-9.]+)\s*(?P<unit>us|µs|ms|s)?",
|
|
text,
|
|
flags=re.IGNORECASE | re.DOTALL,
|
|
)
|
|
if rtt_match:
|
|
unit = rtt_match.group("unit") or "ms"
|
|
for name in ["min", "mean", "median", "max"]:
|
|
value = rtt_match.group(name)
|
|
if value is None:
|
|
continue
|
|
attr = f"rtt_{name}_us"
|
|
if getattr(metrics, attr) is None:
|
|
setattr(metrics, attr, unit_to_us(float(value), unit))
|
|
|
|
|
|
def run_parser_self_check() -> None:
|
|
metrics = Metrics(source="upstream-self-check")
|
|
samples: list[float] = []
|
|
fixture = {
|
|
"stats": {
|
|
"rtt": {
|
|
"min": "1.2ms",
|
|
"mean": "1.3ms",
|
|
"median": "1.3ms",
|
|
"max": "1.4ms",
|
|
},
|
|
"ipdv_round_trip": {
|
|
"min": "-900us",
|
|
"mean": "-100us",
|
|
"median": "-50us",
|
|
"max": "200us",
|
|
},
|
|
},
|
|
"round_trips": [
|
|
{
|
|
"timestamps": {"client": {"send": {"wall": 111}}},
|
|
"delay": {"rtt": "1.2ms"},
|
|
"ipdv": {"rtt": "-900us"},
|
|
},
|
|
{
|
|
"timestamps": {"client": {"send": {"wall": 222}}},
|
|
"delay": {"rtt": "1.4ms"},
|
|
"ipdv": {"rtt": "-100us"},
|
|
},
|
|
{
|
|
"lost": True,
|
|
"delay": {"rtt": "99ms"},
|
|
"ipdv": {"rtt": "-99ms"},
|
|
},
|
|
],
|
|
}
|
|
extract_upstream_json_metrics(fixture, metrics, samples)
|
|
expected_samples = [1200.0, 1400.0]
|
|
expected_stats = {
|
|
"rtt_min_us": 1200.0,
|
|
"rtt_mean_us": 1300.0,
|
|
"rtt_median_us": 1300.0,
|
|
"rtt_max_us": 1400.0,
|
|
}
|
|
if samples != expected_samples:
|
|
raise RuntimeError(
|
|
"upstream parser self-check failed: "
|
|
f"expected RTT samples {expected_samples}, got {samples}"
|
|
)
|
|
for attr, expected in expected_stats.items():
|
|
actual = getattr(metrics, attr)
|
|
if actual != expected:
|
|
raise RuntimeError(
|
|
"upstream parser self-check failed: "
|
|
f"expected {attr}={expected}, got {actual}"
|
|
)
|
|
if metrics.first_send_wall_ns != 111:
|
|
raise RuntimeError(
|
|
"upstream parser self-check failed: "
|
|
f"expected first_send_wall_ns=111, got {metrics.first_send_wall_ns}"
|
|
)
|
|
|
|
|
|
def classify_difference(upstream: Metrics, rust: Metrics) -> dict[str, Any]:
|
|
notes: list[str] = []
|
|
deltas: dict[str, Any] = {}
|
|
|
|
if upstream.sent is not None and rust.sent is not None:
|
|
deltas["sent"] = rust.sent - upstream.sent
|
|
if abs(deltas["sent"]) > 1:
|
|
notes.append("packet count/configuration difference")
|
|
if upstream.received is not None and rust.received is not None:
|
|
deltas["received"] = rust.received - upstream.received
|
|
if abs(deltas["received"]) > 1:
|
|
notes.append("received packet/loss difference")
|
|
if upstream.loss_percent is not None and rust.loss_percent is not None:
|
|
deltas["loss_percent"] = rust.loss_percent - upstream.loss_percent
|
|
if abs(deltas["loss_percent"]) >= 5.0:
|
|
notes.append("packet loss difference")
|
|
if upstream.rtt_mean_us is not None and rust.rtt_mean_us is not None:
|
|
deltas["mean_rtt_us"] = rust.rtt_mean_us - upstream.rtt_mean_us
|
|
baseline = max(abs(upstream.rtt_mean_us), 1.0)
|
|
if abs(deltas["mean_rtt_us"]) / baseline > 0.25 and abs(deltas["mean_rtt_us"]) > 1000:
|
|
notes.append(
|
|
"RTT difference after parser self-check; potential network/timing difference"
|
|
)
|
|
|
|
if upstream.parse_notes or rust.parse_notes:
|
|
notes.append("output parsing limitation")
|
|
if not notes:
|
|
notes.append("no obvious difference detected by conservative parser")
|
|
|
|
return {"deltas": deltas, "notes": sorted(set(notes))}
|
|
|
|
|
|
def fill_stats(metrics: Metrics, values: list[float], prefix: str) -> None:
|
|
clean = [value for value in values if math.isfinite(value)]
|
|
if not clean:
|
|
return
|
|
setattr(metrics, f"{prefix}_min_us", min(clean))
|
|
setattr(metrics, f"{prefix}_mean_us", mean(clean))
|
|
setattr(metrics, f"{prefix}_median_us", median(clean))
|
|
setattr(metrics, f"{prefix}_max_us", max(clean))
|
|
|
|
|
|
def parse_duration_to_us(value: Any) -> float | None:
|
|
if isinstance(value, (int, float)):
|
|
# Upstream Go JSON commonly serializes time.Duration as nanoseconds.
|
|
return float(value) / 1000.0
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
numeric = parse_float(stripped)
|
|
if numeric is not None:
|
|
return numeric / 1000.0
|
|
match = re.fullmatch(r"(-?[0-9]+(?:\.[0-9]+)?)(ns|us|µs|ms|s)", stripped)
|
|
if match:
|
|
return unit_to_us(float(match.group(1)), match.group(2))
|
|
if isinstance(value, dict):
|
|
for key in ["us", "usec", "microseconds"]:
|
|
if key in value:
|
|
return parse_float(value[key])
|
|
for key in ["ns", "nsec", "nanoseconds"]:
|
|
parsed = parse_float(value[key]) if key in value else None
|
|
if parsed is not None:
|
|
return parsed / 1000.0
|
|
return None
|
|
|
|
|
|
def unit_to_us(value: float, unit: str) -> float:
|
|
if unit in {"us", "µs"}:
|
|
return value
|
|
if unit == "ns":
|
|
return value / 1000.0
|
|
if unit == "ms":
|
|
return value * 1000.0
|
|
if unit == "s":
|
|
return value * 1_000_000.0
|
|
return value
|
|
|
|
|
|
def parse_int(value: Any) -> int | None:
|
|
try:
|
|
if value is None:
|
|
return None
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def parse_float(value: Any) -> float | None:
|
|
try:
|
|
if value is None:
|
|
return None
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def append_if_number(values: list[float], value: float | None) -> None:
|
|
if value is not None and math.isfinite(value):
|
|
values.append(value)
|
|
|
|
|
|
def blank_none(value: Any) -> str:
|
|
return "" if value is None else str(value)
|
|
|
|
|
|
def write_json(path: Path, value: Any) -> None:
|
|
path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
|
|
def write_case_summary(path: Path, result: dict[str, Any]) -> None:
|
|
upstream = result["metrics"]["upstream"]
|
|
rust = result["metrics"]["irtt_rs"]
|
|
lines = [
|
|
f"# {result['case']['name']}",
|
|
"",
|
|
f"target: `{result['case']['target']}`",
|
|
f"duration: `{result['case']['duration']}`",
|
|
f"interval: `{result['case']['interval']}`",
|
|
f"paired_start_skew_ms: `{result['paired_start_skew_ms']:.3f}`",
|
|
"paired_first_send_skew_ms: `"
|
|
+ format_optional_float(result.get("paired_first_send_skew_ms"))
|
|
+ "`",
|
|
"",
|
|
"| client | exit | sent | received | loss % | RTT min/mean/median/max us |",
|
|
"| --- | ---: | ---: | ---: | ---: | --- |",
|
|
summary_row("upstream", result["upstream"]["returncode"], upstream),
|
|
summary_row("irtt-rs", result["irtt_rs"]["returncode"], rust),
|
|
"",
|
|
"## Difference Notes",
|
|
]
|
|
for note in result["comparison"]["notes"]:
|
|
lines.append(f"- {note}")
|
|
lines.extend(["", "## irtt-rs RTT Semantics", ""])
|
|
lines.append("| field | min/mean/median/max us |")
|
|
lines.append("| --- | --- |")
|
|
lines.append(
|
|
"| effective | "
|
|
+ metric_quad(rust, ["rtt_min_us", "rtt_mean_us", "rtt_median_us", "rtt_max_us"])
|
|
+ " |"
|
|
)
|
|
lines.append(
|
|
"| raw | "
|
|
+ metric_quad(
|
|
rust,
|
|
[
|
|
"raw_rtt_min_us",
|
|
"raw_rtt_mean_us",
|
|
"raw_rtt_median_us",
|
|
"raw_rtt_max_us",
|
|
],
|
|
)
|
|
+ " |"
|
|
)
|
|
lines.append(
|
|
"| adjusted | "
|
|
+ metric_quad(
|
|
rust,
|
|
[
|
|
"adjusted_rtt_min_us",
|
|
"adjusted_rtt_mean_us",
|
|
"adjusted_rtt_median_us",
|
|
"adjusted_rtt_max_us",
|
|
],
|
|
)
|
|
+ " |"
|
|
)
|
|
lines.extend(["", "## Artifacts"])
|
|
for label, command in [("upstream", result["upstream"]), ("irtt-rs", result["irtt_rs"])]:
|
|
lines.append(f"- {label} stdout: `{command['stdout_path']}`")
|
|
lines.append(f"- {label} stderr: `{command['stderr_path']}`")
|
|
if upstream.get("samples_path"):
|
|
lines.append(f"- upstream samples: `{upstream['samples_path']}`")
|
|
if rust.get("samples_path"):
|
|
lines.append(f"- irtt-rs samples: `{rust['samples_path']}`")
|
|
for side, metrics in [("upstream", upstream), ("irtt-rs", rust)]:
|
|
if metrics.get("parse_notes"):
|
|
lines.extend(["", f"## {side} Parse Notes"])
|
|
for note in metrics["parse_notes"]:
|
|
lines.append(f"- {note}")
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def summary_row(name: str, exit_code: int, metrics: dict[str, Any]) -> str:
|
|
rtt = metric_quad(metrics, ["rtt_min_us", "rtt_mean_us", "rtt_median_us", "rtt_max_us"])
|
|
return (
|
|
f"| {name} | {exit_code} | {format_metric(metrics.get('sent'))} | "
|
|
f"{format_metric(metrics.get('received'))} | "
|
|
f"{format_metric(metrics.get('loss_percent'))} | {rtt} |"
|
|
)
|
|
|
|
|
|
def metric_quad(metrics: dict[str, Any], keys: list[str]) -> str:
|
|
return "/".join(format_metric(metrics.get(key)) for key in keys)
|
|
|
|
|
|
def format_metric(value: Any) -> str:
|
|
if value is None:
|
|
return "-"
|
|
if isinstance(value, float):
|
|
return f"{value:.3f}"
|
|
return str(value)
|
|
|
|
|
|
def format_optional_float(value: Any) -> str:
|
|
if value is None:
|
|
return "unavailable"
|
|
return f"{float(value):.3f}"
|
|
|
|
|
|
def write_index(run_dir: Path, results: list[dict[str, Any]], local_error: str | None) -> None:
|
|
lines = [
|
|
"# IRTT Client Comparison Run",
|
|
"",
|
|
f"created_at: `{utc_now()}`",
|
|
"",
|
|
"| case | target | upstream exit | irtt-rs exit | notes |",
|
|
"| --- | --- | ---: | ---: | --- |",
|
|
]
|
|
for result in results:
|
|
notes = "; ".join(result["comparison"]["notes"])
|
|
lines.append(
|
|
f"| [{result['case']['name']}]({result['case']['name']}/summary.md) | "
|
|
f"`{result['case']['target']}` | {result['upstream']['returncode']} | "
|
|
f"{result['irtt_rs']['returncode']} | {notes} |"
|
|
)
|
|
if local_error:
|
|
lines.extend(["", "## Local Server Required", "", local_error])
|
|
(run_dir / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
write_json(run_dir / "comparison.json", {"results": results, "local_error": local_error})
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
try:
|
|
run_parser_self_check()
|
|
except RuntimeError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 2
|
|
run_stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
run_dir = Path(args.output_dir) / run_stamp
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
cases = build_cases(args)
|
|
if not cases:
|
|
print("no test cases selected", file=sys.stderr)
|
|
return 2
|
|
|
|
local_error: str | None = None
|
|
if any(case.is_local for case in cases):
|
|
ok, message = check_local_server(args)
|
|
if not ok:
|
|
local_error = message
|
|
print(message, file=sys.stderr)
|
|
cases = [case for case in cases if not case.is_local]
|
|
|
|
if cases and not args.skip_build and not preflight_build_irtt_cli(run_dir):
|
|
write_index(run_dir, [], local_error)
|
|
return 1
|
|
|
|
results: list[dict[str, Any]] = []
|
|
all_ok = local_error is None
|
|
for case in cases:
|
|
result, ok = run_case(case, run_dir, args)
|
|
results.append(result)
|
|
all_ok = all_ok and ok
|
|
|
|
write_index(run_dir, results, local_error)
|
|
print(f"[{utc_now()}] wrote comparison artifacts to {run_dir}", flush=True)
|
|
return 0 if all_ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|