rustfs-erasure-codec 7.0.1

Rust implementation of Reed-Solomon erasure coding
Documentation
#!/usr/bin/env python3
import argparse
import csv
import json
import pathlib
import platform
import subprocess
from statistics import mean
from typing import Dict, List, Tuple

KNOWN_BACKENDS = {
    "auto",
    "scalar",
    "rust-ssse3",
    "simd-c",
    "rust-avx2",
    "rust-avx512",
    "rust-gfni-avx2",
    "rust-gfni-avx512",
}

DEFAULT_POLICY_ELIGIBLE_BACKENDS_X86 = {
    "rust-gfni-avx512",
    "rust-gfni-avx2",
    "rust-avx2",
    "rust-avx512",
    "rust-ssse3",
    "simd-c",
    "scalar",
}

SUPPORTED_RELEASE_SMOKE_FILES = {
    "smoke-results-release-auto.csv",
    "smoke-results-release-scalar.csv",
    "smoke-results-release-rust-ssse3.csv",
    "smoke-results-release-simd-c.csv",
    "smoke-results-release-rust-avx2.csv",
    "smoke-results-release-rust-avx512.csv",
    "smoke-results-release-rust-gfni-avx2.csv",
    "smoke-results-release-rust-gfni-avx512.csv",
}

CURRENT_RUNTIME_PRIORITY_X86 = [
    "rust-gfni-avx512",
    "rust-gfni-avx2",
    "rust-avx2",
    "rust-avx512",
    "rust-ssse3",
    "simd-c",
    "scalar",
]


def load_json(path: pathlib.Path):
    with path.open() as f:
        return json.load(f)


def load_csv(path: pathlib.Path):
    with path.open() as f:
        return list(csv.DictReader(f))


def capture_machine_info() -> Dict[str, str]:
    info = {
        "hostname": platform.node(),
        "arch": platform.machine(),
        "platform": platform.platform(),
    }

    try:
        info["lscpu"] = subprocess.check_output(["lscpu"], text=True)
    except (FileNotFoundError, subprocess.CalledProcessError):
        info["lscpu"] = ""

    try:
        info["uname_a"] = subprocess.check_output(["uname", "-a"], text=True).strip()
    except (FileNotFoundError, subprocess.CalledProcessError):
        info["uname_a"] = ""

    return info


def collect_criterion(root: pathlib.Path) -> List[Dict]:
    criterion_dir = root / "target" / "criterion"
    rows = []
    for path in sorted(criterion_dir.glob("**/new/estimates.json")):
        rel = path.relative_to(criterion_dir)
        parts = rel.parts
        if len(parts) < 3:
            continue
        benchmark_meta = load_json(path.parent / "benchmark.json")
        bench_name = benchmark_meta["group_id"]
        length = benchmark_meta["function_id"]
        data = load_json(path)
        rows.append(
            {
                "benchmark": bench_name,
                "length": length,
                "mean_ns": data["mean"]["point_estimate"],
                "lower_ns": data["mean"]["confidence_interval"]["lower_bound"],
                "upper_ns": data["mean"]["confidence_interval"]["upper_bound"],
            }
        )
    return rows


def collect_release_smoke(root: pathlib.Path) -> Dict[str, List[Dict]]:
    smoke_dir = root / "target" / "benchmark-smoke"
    out = {}
    for path in sorted(smoke_dir.glob("smoke-results-release-*.csv")):
        if path.name not in SUPPORTED_RELEASE_SMOKE_FILES:
            continue
        out[path.name] = load_csv(path)
    return out


def parse_backend_override_from_benchmark(benchmark_name: str) -> str:
    marker = "_override_"
    if marker not in benchmark_name:
        return benchmark_name
    override = benchmark_name.split(marker, 1)[1]
    return override.rstrip("_")


def backend_rankings(machine_json: Dict) -> Dict[str, List[Dict]]:
    rankings = {}
    smoke = machine_json["release_smoke"]
    focus_case = {
        "data_shards": "10",
        "parity_shards": "4",
        "shard_size": "1048576",
    }
    for op in ["encode", "verify", "reconstruct", "reconstruct_data"]:
        rows = []
        for file_name, records in smoke.items():
            for record in records:
                if record["backend_override"] not in KNOWN_BACKENDS:
                    continue
                if record.get("override_honored", "true").lower() != "true":
                    continue
                if (
                    record["operation"] == op
                    and record["data_shards"] == focus_case["data_shards"]
                    and record["parity_shards"] == focus_case["parity_shards"]
                    and record["shard_size"] == focus_case["shard_size"]
                ):
                    rows.append(
                        {
                            "backend": record["backend"],
                            "backend_override": record["backend_override"],
                            "throughput_mb_s": float(record["throughput_mb_s"]),
                            "source": file_name,
                        }
                    )
        rankings[op] = sorted(rows, key=lambda item: item["throughput_mb_s"], reverse=True)
    return rankings


def criterion_rankings(machine_json: Dict) -> Dict[str, List[Dict]]:
    grouped: Dict[Tuple[str, str], List[float]] = {}
    for row in machine_json["criterion_galois_backend"]:
        benchmark = row["benchmark"]
        length = row["length"]
        grouped.setdefault((benchmark, length), []).append(float(row["mean_ns"]))

    rankings: Dict[str, List[Dict]] = {}
    for op in ["galois_mul_slice", "galois_mul_slice_xor"]:
        rows = []
        for (benchmark, length), values in grouped.items():
            if not benchmark.startswith(f"{op}_"):
                continue
            rows.append(
                {
                    "backend_override": parse_backend_override_from_benchmark(benchmark),
                    "benchmark": benchmark,
                    "length": length,
                    "mean_ns": mean(values),
                }
            )
        rankings[op] = sorted(rows, key=lambda item: item["mean_ns"])
    return rankings


def choose_recommended_priority(machine_json: Dict) -> Dict:
    smoke = machine_json.get("rankings_10x4_1m", {})
    criterion = criterion_rankings(machine_json)

    score: Dict[str, float] = {}

    smoke_weights = {
        "encode": 1.0,
        "verify": 1.2,
        "reconstruct": 1.5,
        "reconstruct_data": 1.5,
    }
    for op, weight in smoke_weights.items():
        rows = smoke.get(op, [])
        if not rows:
            continue
        best = rows[0]["throughput_mb_s"]
        for idx, row in enumerate(rows):
            override = row["backend_override"]
            if override not in KNOWN_BACKENDS:
                continue
            relative = row["throughput_mb_s"] / best if best else 0.0
            score[override] = score.get(override, 0.0) + relative * weight
            score[override] += max(0.0, (len(rows) - idx - 1) * 0.01)

    criterion_focus = {
        "galois_mul_slice": {"len_1048576": 0.5, "len_4194304": 0.75},
        "galois_mul_slice_xor": {"len_1048576": 0.5, "len_4194304": 0.75},
    }
    for op, lengths in criterion_focus.items():
        rows = criterion.get(op, [])
        per_length = {}
        for row in rows:
            per_length.setdefault(row["length"], []).append(row)
        for length, weight in lengths.items():
            length_rows = per_length.get(length, [])
            if not length_rows:
                continue
            best = length_rows[0]["mean_ns"]
            for idx, row in enumerate(length_rows):
                override = row["backend_override"]
                if override not in KNOWN_BACKENDS:
                    continue
                relative = best / row["mean_ns"] if row["mean_ns"] else 0.0
                score[override] = score.get(override, 0.0) + relative * weight
                score[override] += max(0.0, (len(length_rows) - idx - 1) * 0.005)

    if "auto" in score:
        del score["auto"]

    ordered = [
        {"backend_override": backend, "score": round(value, 4)}
        for backend, value in sorted(score.items(), key=lambda item: item[1], reverse=True)
    ]

    recommendation = {
        "priority_order": [row["backend_override"] for row in ordered],
        "scored_backends": ordered,
        "rationale": [
            "Release smoke results for 10x4_1m are weighted most heavily.",
            "Reconstruct and reconstruct_data are weighted above encode.",
            "Criterion mul_slice and mul_slice_xor at 1 MiB and 4 MiB break close ties.",
            "The recommendation is benchmark-driven and may differ from the current runtime dispatch order.",
        ],
        "current_runtime_priority_x86": CURRENT_RUNTIME_PRIORITY_X86,
        "diverges_from_current_runtime_priority_x86": [row["backend_override"] for row in ordered] != CURRENT_RUNTIME_PRIORITY_X86[: len(ordered)],
    }
    return recommendation


def choose_policy_eligible_priority(machine_json: Dict) -> Dict:
    recommendation = choose_recommended_priority(machine_json)
    eligible = [
        row
        for row in recommendation["scored_backends"]
        if row["backend_override"] in DEFAULT_POLICY_ELIGIBLE_BACKENDS_X86
    ]
    return {
        "priority_order": [row["backend_override"] for row in eligible],
        "scored_backends": eligible,
        "rationale": [
            "Starts from the benchmark-driven ranking.",
            "Filters out backends that are not currently eligible for default runtime dispatch.",
            "The eligible set is aligned to the runtime order implemented in src/galois_8/backend.rs.",
        ],
        "eligible_backends_x86": sorted(DEFAULT_POLICY_ELIGIBLE_BACKENDS_X86),
        "current_runtime_priority_x86": CURRENT_RUNTIME_PRIORITY_X86,
        "diverges_from_current_runtime_priority_x86": [row["backend_override"] for row in eligible] != CURRENT_RUNTIME_PRIORITY_X86[: len(eligible)],
    }


def adoption_decision_stub(machine_json: Dict) -> Dict:
    recommendation = machine_json.get("recommended_default_priority", {})
    mismatches = machine_json.get("release_smoke_override_mismatches", {})
    mismatch_count = sum(len(rows) for rows in mismatches.values())
    diverges = recommendation.get("diverges_from_current_runtime_priority_x86", False)

    return {
        "status": "manual-review-required",
        "reason": (
            "same-machine evidence must be reviewed manually before changing the "
            "default runtime priority"
        ),
        "override_mismatch_count": mismatch_count,
        "diverges_from_current_runtime_priority_x86": diverges,
        "recommended_priority_order": recommendation.get("priority_order", []),
        "suggested_labels": [
            "candidate-only",
            "candidate-default",
            "fallback-only",
        ],
    }


def write_machine_json(root: pathlib.Path, out_json: pathlib.Path, machine_slug: str, date_utc: str):
    release_smoke = collect_release_smoke(root)
    machine_info = capture_machine_info()
    report = {
        "date_utc": date_utc,
        "machine_slug": machine_slug,
        "hostname": machine_info["hostname"],
        "arch": machine_info["arch"],
        "platform": machine_info["platform"],
        "lscpu": machine_info["lscpu"],
        "uname_a": machine_info["uname_a"],
        "criterion_galois_backend": collect_criterion(root),
        "release_smoke": release_smoke,
        "release_smoke_override_mismatches": {
            file_name: [
                row for row in rows if row.get("override_honored", "true").lower() != "true"
            ]
            for file_name, rows in release_smoke.items()
        },
    }
    report["rankings_10x4_1m"] = backend_rankings(report)
    report["criterion_rankings"] = criterion_rankings(report)
    report["recommended_default_priority"] = choose_recommended_priority(report)
    report["policy_eligible_default_priority"] = choose_policy_eligible_priority(report)
    report["adoption_decision_stub"] = adoption_decision_stub(report)
    report["default_switch_checklist"] = {
        "same_machine_required": True,
        "repeat_runs_required": True,
        "kernel_and_workload_evidence_required": True,
        "override_mismatches_must_be_empty": True,
        "manual_review_required_when_priority_diverges": True,
    }
    out_json.parent.mkdir(parents=True, exist_ok=True)
    out_json.write_text(json.dumps(report, indent=2))


def print_summary(root: pathlib.Path):
    bench_dir = root / "benchmarks" / "x86_64-simd"
    rows = []
    for path in sorted(
        path for path in bench_dir.glob("*.json") if not path.name.endswith(".run-meta.json")
    ):
        data = load_json(path)
        rankings = data.get("rankings_10x4_1m", {})
        recommendation = data.get("recommended_default_priority", {})
        top = {op: (rankings.get(op) or [{}])[0] for op in ["encode", "verify", "reconstruct", "reconstruct_data"]}
        rows.append(
            {
                "file": path.name,
                "encode": top["encode"].get("backend_override", "n/a"),
                "verify": top["verify"].get("backend_override", "n/a"),
                "reconstruct": top["reconstruct"].get("backend_override", "n/a"),
                "reconstruct_data": top["reconstruct_data"].get("backend_override", "n/a"),
                "recommended_default_priority": recommendation.get("priority_order", []),
                "recommendation_rationale": recommendation.get("rationale", []),
                "adoption_decision_status": data.get("adoption_decision_stub", {}).get(
                    "status", "legacy-archive-no-stub"
                ),
                "override_mismatch_count": data.get("adoption_decision_stub", {}).get("override_mismatch_count", 0),
            }
        )
    print(json.dumps(rows, indent=2))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", default=".")
    parser.add_argument("--machine-json")
    parser.add_argument("--machine-slug")
    parser.add_argument("--date")
    parser.add_argument("--summary", action="store_true")
    args = parser.parse_args()

    root = pathlib.Path(args.root).resolve()

    if args.summary:
        print_summary(root)
        return

    if not args.machine_json or not args.machine_slug or not args.date:
        raise SystemExit("--machine-json, --machine-slug and --date are required unless --summary is used")

    write_machine_json(root, pathlib.Path(args.machine_json), args.machine_slug, args.date)


if __name__ == "__main__":
    main()