rustfs/scripts/run_conditional_read_abba.py

#!/usr/bin/env python3
"""Run the identical application-read test harness in serial ABBA processes.

This measures local EC reads, not network/authentication or deployment capacity.
Use matching toolchain, features and profiles for both executable artifacts.
"""

import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import shutil
import subprocess
import time


def digest(path):
    result = hashlib.sha256()
    with path.open("rb") as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            result.update(chunk)
    return result.hexdigest()


def parse_rows(output):
    rows = {}
    prefix = "CONDITIONAL_BENCH "
    for line in output.splitlines():
        if prefix in line:
            row = json.loads(line.split(prefix, 1)[1])
            key = (row["size"], row["kind"], row["cache"], row["slowtail_ms"])
            if key in rows:
                raise ValueError(f"duplicate benchmark row {key}")
            rows[key] = row
    if len(rows) != 6:
        raise ValueError(f"expected six benchmark rows, received {len(rows)}")
    if {(key[0], key[1]) for key in rows} != {
        (size, kind) for size in (4096, 1_300_000) for kind in ("unconditional", "hit", "miss")
    }:
        raise ValueError("benchmark workload matrix is incomplete")
    for row in rows.values():
        for metric in ("ops_per_sec", "p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms"):
            if not math.isfinite(row[metric]) or row[metric] <= 0:
                raise ValueError(f"invalid benchmark metric {metric}")
    return rows


def check_resource_isolation(check_load=True):
    commands = subprocess.check_output(["ps", "-A", "-o", "comm="], text=True)
    if any(Path(command.strip()).name in ("cargo", "rustc") for command in commands.splitlines()):
        raise RuntimeError("stop concurrent Cargo builds before measuring")
    load = os.getloadavg()
    if check_load and load[0] > (os.cpu_count() or 1) * 0.5:
        raise RuntimeError(f"host load too high for paired measurements: {load[0]}")
    return load


def run_case(executable, expected_digest, env, log_path=None):
    if digest(executable) != expected_digest:
        raise RuntimeError("benchmark executable changed before execution")
    # Poll for concurrent builds without attributing this benchmark's own CPU
    # use to external host load. Load gates run before and after the process.
    with subprocess.Popen([str(executable.resolve()), "conditional_read_paired_benchmark",
                           "--ignored", "--nocapture", "--test-threads=1"],
                          env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as run:
        try:
            while True:
                try:
                    output, _ = run.communicate(timeout=1)
                    break
                except subprocess.TimeoutExpired:
                    check_resource_isolation(check_load=False)
        except BaseException:
            run.kill()
            output, _ = run.communicate()
            if log_path is not None:
                log_path.write_text(output)
            raise
        returncode = run.returncode
    if log_path is not None:
        log_path.write_text(output)
    check_resource_isolation()
    if digest(executable) != expected_digest:
        raise RuntimeError("benchmark executable changed during execution")
    return returncode, output


def drift_passes(first, last):
    if abs(last["ops_per_sec"] / first["ops_per_sec"] - 1) > 0.10:
        return False
    return all(
        abs(last[metric] - first[metric]) <= max(first[metric] * 0.15, 0.25)
        for metric in ("p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms")
    )


def compare(rows):
    if not all(rows[0].keys() == leg.keys() for leg in rows):
        raise ValueError("ABBA legs have different workload matrices")
    unstable = [key for key in rows[0] if not drift_passes(rows[0][key], rows[3][key])]
    if unstable:
        return {"accepted": False, "reason": "baseline_drift", "unstable_cases": unstable}
    results = []
    for key, first in rows[0].items():
        baseline = {metric: math.sqrt(first[metric] * rows[3][key][metric]) for metric in
                    ("ops_per_sec", "p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms")}
        candidate = {metric: math.sqrt(rows[1][key][metric] * rows[2][key][metric]) for metric in baseline}
        # Protect ordinary reads and condition misses; 304 savings alone do not pass.
        passed = candidate["ops_per_sec"] >= baseline["ops_per_sec"] * 0.95 and all(
            candidate[metric] <= baseline[metric] + max(baseline[metric] * 0.10, 0.25)
            for metric in ("p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms")
        )
        results.append({"case": key, "baseline": baseline, "candidate": candidate,
                        "within_regression_budget": passed})
    return {"accepted": all(row["within_regression_budget"] for row in results), "cases": results}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--baseline", type=Path, required=True)
    parser.add_argument("--candidate", type=Path, required=True)
    parser.add_argument("--baseline-revision", required=True)
    parser.add_argument("--candidate-revision", required=True)
    parser.add_argument("--harness", type=Path, required=True)
    parser.add_argument("--profile", required=True)
    parser.add_argument("--out-dir", type=Path, required=True)
    parser.add_argument("--iterations", type=int, default=1000)
    parser.add_argument("--concurrency", type=int, default=8)
    parser.add_argument("--cooldown-seconds", type=float, default=5)
    args = parser.parse_args()
    if args.iterations < 100 or args.concurrency < 1 or args.cooldown_seconds < 0:
        parser.error("iterations must be >=100, concurrency >0, cooldown >=0")
    args.out_dir.mkdir(parents=True, exist_ok=False)
    executables = {}
    for name, source in (("baseline", args.baseline), ("candidate", args.candidate)):
        original_digest = digest(source)
        snapshot = args.out_dir / f"{name}.binary"
        shutil.copyfile(source, snapshot)
        snapshot.chmod(0o555)
        if digest(snapshot) != original_digest:
            raise RuntimeError("source executable changed while taking its snapshot")
        executables[name] = snapshot
    manifest = {
        "baseline_revision": args.baseline_revision, "candidate_revision": args.candidate_revision,
        "baseline_sha256": digest(executables["baseline"]), "candidate_sha256": digest(executables["candidate"]),
        "harness_sha256": digest(args.harness), "profile": args.profile,
        "iterations": args.iterations, "concurrency": args.concurrency,
        "order": ["A1", "B1", "B2", "A2"], "slowtail_ms": [0, 20], "cache": [False, True],
        "drift_budget": "throughput 10%; latency and lock percentiles 15% or 0.25ms",
        "regression_budget": "throughput -5%; latency and lock percentiles +10% or 0.25ms",
        "cpu_count": os.cpu_count(), "load_at_start": check_resource_isolation(),
    }
    (args.out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
    legs = []
    for name, artifact in (("A1", "baseline"), ("B1", "candidate"),
                           ("B2", "candidate"), ("A2", "baseline")):
        executable = executables[artifact]
        cases = {}
        for cache in (False, True):
            for delay in (0, 20):
                check_resource_isolation()
                env = os.environ.copy()
                # Keep ambient experiments and proxies out of localhost-sensitive runs.
                for variable in tuple(env):
                    if variable.startswith("RUSTFS_") or variable.lower().endswith("_proxy"):
                        del env[variable]
                env.update({
                    "NO_PROXY": "*", "no_proxy": "*", "RUSTFS_OBJECT_LOCK_DIAG_ENABLE": "true",
                    "RUSTFS_OBJECT_DATA_CACHE_ENABLE": str(cache).lower(),
                    "RUSTFS_OBJECT_DATA_CACHE_MODE": "fill_materialize_enabled",
                    "RUSTFS_OBJECT_DATA_CACHE_MAX_BYTES": "8388608",
                    "RUSTFS_OBJECT_DATA_CACHE_MAX_ENTRY_BYTES": "2097152",
                    "RUSTFS_OBJECT_DATA_CACHE_MIN_FREE_MEMORY_PERCENT": "0",
                    "RUSTFS_CONDITIONAL_BENCH_ITERATIONS": str(args.iterations),
                    "RUSTFS_CONDITIONAL_BENCH_CONCURRENCY": str(args.concurrency),
                    "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS": str(delay),
                    "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS": "3",
                    "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET": "conditional-bench",
                    "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX": "bench-",
                })
                returncode, output = run_case(executable, manifest[f"{artifact}_sha256"], env,
                                             args.out_dir / f"{name}-cache{int(cache)}-delay{delay}.log")
                if returncode:
                    raise RuntimeError(f"{name} cache={cache} delay={delay} failed; see saved log")
                for key, row in parse_rows(output).items():
                    if (row["cache"] != cache or row["slowtail_ms"] != str(delay)
                            or row["iterations"] != args.iterations or row["concurrency"] != args.concurrency
                            or row["metadata_cache_hits"] != 0 or not row["metadata_cache_bypassed"]
                            or row["lock_samples"] <= 0):
                        raise ValueError("benchmark did not observe requested environment")
                    cases[key] = row
                print(f"{name}: cache={cache} slowtail={delay}ms complete", flush=True)
        legs.append(cases)
        time.sleep(args.cooldown_seconds)
    result = compare(legs)
    result["load_at_end"] = os.getloadavg()
    (args.out_dir / "summary.json").write_text(json.dumps(result, indent=2) + "\n")
    print(json.dumps(result, indent=2))
    return 0 if result["accepted"] else 2


if __name__ == "__main__":
    raise SystemExit(main())