chunkers: stale kernel comments, explicit kernel dispatch, kernel benchmark script

scripts/bench_chunker_kernels.py

@@ -0,0 +1,323 @@
+#!/usr/bin/env python3
+"""Benchmark every chunker scan kernel of a borg build and print a chunkers x kernels table.
+
+borg picks a default scan kernel per platform (the *_kernel_default() functions of
+the C kernels); BORG_FASTCDC_KERNEL / BORG_BUZHASH64_KERNEL / BORG_AES_CHUNKER_KERNEL
+name another one. Which kernel is fastest is not predictable from the instruction set
+- NEON beats the sequential loop on an Apple M3, while on a Zen 4 the sequential loop
+beats AVX-512 - so the only way to know is to measure on the machine in question.
+That is what this does.
+
+It drives `borg benchmark cpu --chunking --json` once per kernel combination, so it
+needs nothing but a `borg` binary in $PATH (or --borg PATH) and python3. It works on
+arm64 and x86-64, on Linux, the BSDs and macOS; which kernels exist is discovered
+from the binary, not assumed.
+
+Usage:
+
+    scripts/bench_chunker_kernels.py                 # 3 repetitions, ~1 GiB per row
+    scripts/bench_chunker_kernels.py --reps 5
+    scripts/bench_chunker_kernels.py --smoke         # tiny data, just checks it works
+    scripts/bench_chunker_kernels.py --json out.json # keep the raw samples too
+
+Note on the data: `borg benchmark cpu` makes its own random buffer per process, so
+every run chunks different (but equally random, equally large) bytes. Repetitions and
+the median column absorb that; the reported spread says how much it mattered.
+"""
+
+import argparse
+import json
+import os
+import platform
+import re
+import shutil
+import statistics
+import subprocess
+import sys
+import time
+
+# Which env var selects the scan kernel of which chunker. A chunker that is in no
+# group has no kernel to select and is reported in the "n/a" column.
+KERNEL_ENVS = [
+    ("BORG_FASTCDC_KERNEL", ["fastcdc"]),
+    ("BORG_BUZHASH64_KERNEL", ["buzhash64"]),
+    ("BORG_AES_CHUNKER_KERNEL", ["toeplitz-aes", "rabin-aes", "goldilocks-aes"]),
+]
+
+NO_KERNEL = "n/a"  # column for chunkers that have no selectable kernel
+
+# borg reports its kernels most-specialised first ("avx512,avx2,blockwise,scalar");
+# reversed, the table reads left to right from the simplest kernel to the fanciest.
+VALID_VALUES_RE = re.compile(r"Valid values:\s*(\S.*?)\s*$", re.MULTILINE)
+
+# Keeps a probe run to 128 KiB and one iteration per row: enough to construct every
+# chunker (which is what validates a kernel request) without benchmarking anything.
+PROBE_ENV = {"_BORG_BENCHMARK_CPU_TEST": "1"}
+
+
+def log(msg):
+    """Progress goes to stderr, so that stdout stays a clean table."""
+    print(msg, file=sys.stderr, flush=True)
+
+
+def cpu_model():
+    """A human-readable CPU name, or the bare architecture if nothing better is found."""
+    system = platform.system()
+    try:
+        if system == "Darwin":
+            out = subprocess.check_output(["sysctl", "-n", "machdep.cpu.brand_string"], text=True)
+            return out.strip()
+        if system == "Linux":
+            with open("/proc/cpuinfo") as f:
+                for line in f:
+                    # x86 says "model name", arm64 often only has "Model"/"CPU part"
+                    if line.split(":")[0].strip() in ("model name", "Model"):
+                        return line.split(":", 1)[1].strip()
+        else:  # the BSDs
+            out = subprocess.check_output(["sysctl", "-n", "hw.model"], text=True)
+            return out.strip()
+    except (OSError, subprocess.SubprocessError, IndexError):
+        pass
+    return platform.processor() or platform.machine()
+
+
+def run_borg(borg, extra_env, smoke):
+    """One `borg benchmark cpu --chunking --json` run.
+
+    Returns (rows, error): rows is the parsed chunkers list on success, error is a
+    one-line reason on failure (an unsupported kernel, a borg too old, ...).
+    """
+    env = dict(os.environ)
+    # start from a clean slate, so that a kernel var set in the caller's shell
+    # cannot silently apply to runs that mean to measure the default
+    for envvar, _ in KERNEL_ENVS:
+        env.pop(envvar, None)
+    env.pop("_BORG_BENCHMARK_CPU_TEST", None)
+    env.update(extra_env)
+    if smoke:
+        env.update(PROBE_ENV)
+    proc = subprocess.run(
+        [borg, "benchmark", "cpu", "--chunking", "--json"],
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    output = proc.stdout + proc.stderr
+    if proc.returncode != 0:
+        return None, error_line(output, proc.returncode)
+    try:
+        # tolerate anything borg may print around the JSON document
+        start = proc.stdout.index("{")
+        doc, _ = json.JSONDecoder().raw_decode(proc.stdout[start:])
+        rows = doc["chunkers"]
+    except (ValueError, KeyError) as e:
+        return None, "could not parse the --json output (%s)" % e
+    # a chunker that could not be constructed (e.g. an unusable kernel was
+    # requested) is reported as an error row, with the benchmark going on
+    for row in rows:
+        if "error" in row:
+            return None, row["error"]
+    return rows, None
+
+
+def error_line(output, returncode):
+    """The most informative single line of a failed run."""
+    for line in output.splitlines():
+        # "Error:" alone is just the header borg prints above the real message
+        if "ValueError:" in line or ("Error:" in line and line.strip() != "Error:"):
+            return line.strip()
+    return "borg exited with rc=%d" % returncode
+
+
+def discover_kernels(borg, envvar):
+    """The kernels of <envvar> that this build has AND this CPU can run.
+
+    The build's list comes from borg itself: an impossible value makes it name the
+    valid ones. Each is then probed, because a kernel can be compiled in and still
+    be unusable here (an AVX-512 binary on a CPU without it), and a kernel that
+    cannot run must not enter the plan - borg refuses rather than falling back, so
+    it would take an entire run down with it.
+    """
+    _, err = run_borg(borg, {envvar: "!invalid!"}, smoke=True)
+    if err is None:
+        log("  %s: this borg does not reject an invalid kernel - skipping" % envvar)
+        return []
+    match = VALID_VALUES_RE.search(err)
+    if not match:
+        log("  %s: not supported by this borg (%s)" % (envvar, err))
+        return []
+    names = [n.strip() for n in match.group(1).split(",") if n.strip()]
+    usable = []
+    for name in reversed(names):  # simplest kernel first, see VALID_VALUES_RE
+        _, err = run_borg(borg, {envvar: name}, smoke=True)
+        if err is None:
+            usable.append(name)
+        else:
+            log("  %s=%s: unusable here (%s)" % (envvar, name, err))
+    log("  %s: %s" % (envvar, ", ".join(usable) if usable else "none usable"))
+    return usable
+
+
+def build_plan(kernel_sets):
+    """The runs of one repetition, as a list of {envvar: kernel} dicts.
+
+    The env vars are independent, so a single run measures one kernel of each of
+    them at once: as many runs as the longest kernel list, with the shorter lists
+    cycling. Every kernel is measured at least once per repetition.
+    """
+    n_runs = max((len(ks) for ks in kernel_sets.values()), default=0)
+    plan = []
+    for i in range(n_runs):
+        plan.append({envvar: ks[i % len(ks)] for envvar, ks in kernel_sets.items() if ks})
+    return plan
+
+
+def kernel_of(algo, env_of_algo, run_env):
+    """Which kernel produced this chunker's row in this run."""
+    envvar = env_of_algo.get(algo)
+    return run_env.get(envvar, NO_KERNEL) if envvar else NO_KERNEL
+
+
+def render(rows, columns, samples):
+    """The chunkers x kernels table, throughput only."""
+    label_w = max(len(r) for r in rows)
+    cells = {}
+    for row in rows:
+        for col in columns:
+            values = samples.get((row, col))
+            cells[(row, col)] = "%.1f" % statistics.median(values) if values else "-"
+    widths = [max([len(col)] + [len(cells[(row, col)]) for row in rows]) for col in columns]
+
+    out = []
+    header = " " * label_w + " |" + "|".join(" %*s " % (w, c) for w, c in zip(widths, columns))
+    out.append(header)
+    out.append("-" * (label_w + 1) + "+" + "+".join("-" * (w + 2) for w in widths))
+    for row in rows:
+        line = "%-*s |" % (label_w, row)
+        line += "|".join(" %*s " % (w, cells[(row, col)]) for w, col in zip(widths, columns))
+        out.append(line)
+    return "\n".join(out)
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Benchmark all chunker scan kernels of a borg build (throughput in MB/s).",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--borg", default=None, help="borg binary to drive (default: borg from $PATH)")
+    parser.add_argument("--reps", type=int, default=3, help="repetitions of the whole matrix (default: 3)")
+    parser.add_argument(
+        "--smoke",
+        action="store_true",
+        help="run on minimal data - checks that the matrix works, the numbers are meaningless",
+    )
+    parser.add_argument("--json", metavar="FILE", help="also write the raw samples (MB/s per run) there")
+    parser.add_argument(
+        "--warmup",
+        type=int,
+        default=1,
+        help="discarded runs before measuring, to let the CPU clock ramp up (default: 1)",
+    )
+    args = parser.parse_args()
+
+    borg = args.borg or shutil.which("borg")
+    if not borg:
+        log("no borg binary in $PATH - pass --borg PATH")
+        return 2
+    try:
+        version = subprocess.check_output([borg, "--version"], text=True).strip()
+    except (OSError, subprocess.SubprocessError) as e:
+        log("cannot run %s: %s" % (borg, e))
+        return 2
+
+    log("borg: %s (%s)" % (version, borg))
+    log("discovering kernels ...")
+    kernel_sets = {envvar: discover_kernels(borg, envvar) for envvar, _ in KERNEL_ENVS}
+    plan = build_plan(kernel_sets)
+    if not plan:
+        log("no selectable kernels at all - nothing to compare")
+        return 1
+
+    env_of_algo = {algo: envvar for envvar, algos in KERNEL_ENVS for algo in algos}
+    samples = {}  # (row label, kernel) -> [MB/s, ...]
+    row_order = []  # chunker labels, in the order the benchmark reports them
+    started = time.time()
+    total = len(plan) * args.reps
+    done = 0
+
+    # A cold CPU is a slow CPU: the first runs of a session come out low while the
+    # clock ramps up, which skews whichever kernel happened to go first.
+    for i in range(args.warmup):
+        log("[warmup %d/%d]" % (i + 1, args.warmup))
+        run_borg(borg, plan[0], args.smoke)
+
+    for rep in range(1, args.reps + 1):
+        for run_env in plan:
+            done += 1
+            what = " ".join("%s=%s" % (k.split("_", 1)[1], v) for k, v in sorted(run_env.items()))
+            log("[%d/%d] %s" % (done, total, what))
+            rows, err = run_borg(borg, run_env, args.smoke)
+            if err:
+                log("        failed: %s" % err)
+                continue
+            for row in rows:
+                algo = row["algo"]
+                label = "%s,%s" % (algo, row["algo_params"]) if row.get("algo_params") else algo
+                if label not in row_order:
+                    row_order.append(label)
+                mbs = row["size"] / row["time"] / 1e6
+                samples.setdefault((label, kernel_of(algo, env_of_algo, run_env)), []).append(mbs)
+
+    if not samples:
+        log("every run failed - no results")
+        return 1
+
+    # Columns in env var order, simplest kernel first, kernel-less chunkers last.
+    # Different env vars share kernel names (fastcdc and buzhash64 both have
+    # 'scalar'), and one column per name is enough: the rows keep them apart.
+    columns = []
+    for envvar, _ in KERNEL_ENVS:
+        columns.extend(k for k in kernel_sets[envvar] if k not in columns)
+    if any(col == NO_KERNEL for _, col in samples):
+        columns.append(NO_KERNEL)
+
+    title = "%s (%s, %s)" % (cpu_model(), platform.machine(), platform.system())
+    print(title)
+    print(
+        "%s, median of %d repetitions, MB/s%s"
+        % (version, args.reps, " [SMOKE - not real numbers]" if args.smoke else "")
+    )
+    print()
+    print(render(row_order, columns, samples))
+
+    spreads = [((max(v) - min(v)) / max(v), key) for key, v in samples.items() if len(v) > 1 and max(v) > 0]
+    if spreads:
+        worst, (row, col) = max(spreads)
+        print()
+        print(
+            "run-to-run spread: %.1f%% median, %.1f%% worst (%s / %s)"
+            % (100 * statistics.median(s for s, _ in spreads), 100 * worst, row, col)
+        )
+    log("total time: %.0fs" % (time.time() - started))
+
+    if args.json:
+        with open(args.json, "w") as f:
+            json.dump(
+                {
+                    "cpu": title,
+                    "borg": version,
+                    "reps": args.reps,
+                    "smoke": args.smoke,
+                    "samples_mbs": {"%s|%s" % key: values for key, values in samples.items()},
+                },
+                f,
+                indent=1,
+                sort_keys=True,
+            )
+        log("raw samples: %s" % args.json)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/borg/chunkers/__init__.py

@@ -16,9 +16,10 @@
 def _made(algo, chunker):
     """Log which chunker was constructed and which scan kernel it ended up with.
 
-    The kernel is worth seeing because it is not implied by the platform: it
-    defaults to the simplest implementation and is only changed by the
-    BORG_*_KERNEL env vars (see kernel_env.py).
+    The kernel is worth seeing because it is not implied by the platform: the
+    default is chosen per platform by the *_kernel_default() functions of the C
+    kernels, from benchmarks rather than from the instruction set, and the
+    BORG_*_KERNEL env vars can override it (see kernel_env.py).
 
     Chunkers that have nothing to select - buzhash and fixed have exactly one
     implementation each - say so rather than omitting the field, so that a

src/borg/chunkers/buzhash64_impl.c

@@ -223,11 +223,14 @@ __attribute__((target("avx2"))) static inline __m256i bz64_rotl4(__m256i v, __m2
  * half (two vpermq+vpxor steps) plus a broadcast of the low half's total into
  * the high half.
  *
- * Only the 2x8 table lookups stay scalar, and they are done one block ahead
- * into a double buffer: a 32-byte vector load placed right after the eight
- * 8-byte scalar stores that filled it cannot use store-to-load forwarding and
- * stalls, twice per block. Loading block j+1 while block j is tested puts a
- * full loop body between the stores and the load, so the stall disappears. */
+ * Only the 2x8 table lookups stay scalar. They are done one block ahead into
+ * a double buffer, so that a whole loop body separates the eight 8-byte
+ * stores from the two 32-byte vector loads reading them back: wider than the
+ * stores, those loads cannot be store-forwarded and wait for the stores to
+ * commit. The distance hides most of that wait, not all of it - performance
+ * counters still show the loads as failed forwards (Zen 4: 4 per block) -
+ * which is part of why this kernel does not beat the sequential loop on
+ * x86-64, see bz64_kernel_default(). */
 __attribute__((target("avx2"))) static size_t
 bz64_scan_simd(const uint64_t *T, const uint64_t *Trot,
                const uint8_t *pr, const uint8_t *pa,
@@ -434,12 +437,14 @@ size_t bz64_scan(const uint64_t *table, const uint64_t *table_rot,
         return bz64_scan_seq(table, table_rot, p_rem, p_add, n, sum, mask);
     case BZ_K_BLOCKWISE:
         return bz64_scan_blockwise(table, table_rot, p_rem, p_add, n, sum, mask);
+    case BZ_K_VECTOR:
+        return bz64_scan_simd(table, table_rot, p_rem, p_add, n, sum, mask);
 #ifdef BZ_KIND_512
     case BZ_K_VECTOR512:
         return bz64_scan_simd512(table, table_rot, p_rem, p_add, n, sum, mask);
 #endif
-    default:
-        return bz64_scan_simd(table, table_rot, p_rem, p_add, n, sum, mask);
+    default: /* not an id of this build: the portable kernel, never one that could SIGILL */
+        return bz64_scan_blockwise(table, table_rot, p_rem, p_add, n, sum, mask);
     }
 }
 
@@ -450,11 +455,13 @@ const char *bz64_kernel_name(int kernel)
         return "scalar";
     case BZ_K_BLOCKWISE:
         return "blockwise";
+    case BZ_K_VECTOR:
+        return BZ_KIND;
 #ifdef BZ_KIND_512
     case BZ_K_VECTOR512:
         return BZ_KIND_512;
 #endif
     default:
-        return BZ_KIND;
+        return "blockwise"; /* what bz64_scan() runs for an unknown id */
     }
 }