ruststream 0.7.5

#!/usr/bin/env python3
"""Turn a benchmark run into the document the benchmarks page reads.

Input is the machine-readable summary `cargo bench -- --output-format=json` writes, one JSON
object per benchmark, in the summary layout gungraun 0.20 writes (its version 7). Output is
`docs/benchmarks/results.json` (schema 2): the `code` section, one entry per scenario, with
instructions and allocations per message, plus what starting the service cost once.

Every scenario is measured three times: over one delivery, over MESSAGES of them, and over twice
MESSAGES. The slope between the last two is the steady-state cost of a message - everything that
happens once is in both totals and cancels in the subtraction. The one-delivery run is the cold
start itself: starting the service and handling the first delivery, measured rather than derived,
because the intercept of a line through two totals of several million carries the noise of both
and comes out negative as easily as not.

Two things are checked while converting, because both failures are silent in the benchmark
output itself:

- every scenario named below is present, so a renamed benchmark is an error rather than a row
  that quietly disappears from the published table;
- no scenario reports an implausibly small instruction count. Collection is switched on for one
  frame, and a frame that stops matching (inlining, a rename) leaves the run reporting the cost
  of the process exit and nothing else. That reads as a spectacular improvement rather than as
  the broken measurement it is.

A benchmark that breaches one of its limits fails the run, and in this output format the runner
says nothing more about it: what went over is recorded in the summary alone. So every breach is
printed under the table, the value the run was compared against next to the new one, and a
summary that cannot be converted still prints its breaches before it stops.
"""

import argparse
import json
import platform
import re
import os
import subprocess
import sys
import tomllib
from datetime import datetime, timezone
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent

# The summary layout read below. Every summary states its layout in `version`, and a gungraun
# release that changes the layout changes the number, so a summary of another version stops the
# conversion with a message naming both rather than with a missing field.
SUMMARY_VERSION = "7"

# Deliveries per measured run, the default of the suite's `MESSAGES` constant. Every published
# number is per message, so the totals are divided by it. `just bench N` builds the
# benches with another count and passes the same one here through `--messages`.
DEFAULT_MESSAGES = 1000
MESSAGES = DEFAULT_MESSAGES

# An instruction count below this on a default-count run means the measurement broke, not that
# the code got faster: the cheapest scenario here costs a thousand times more. The floor scales
# with the count. The cold run handles one delivery, so it is held to a much lower floor - but
# not to none, because a collapsed measurement reports nearly nothing at all.
FLOOR_PER_DEFAULT_RUN = 100_000
FLOOR = FLOOR_PER_DEFAULT_RUN
COLD_FLOOR = 1_000


def configure(messages):
    """Measure against another count of deliveries per run: the per-message division and the
    floor a run is held to follow it."""
    global MESSAGES, FLOOR
    if messages <= 0:
        sys.exit("--messages must be a positive number of deliveries")
    MESSAGES = messages
    FLOOR = FLOOR_PER_DEFAULT_RUN * messages // DEFAULT_MESSAGES


class Scenario:
    """One published row: what it is called, which benchmark measured it, and whether a
    regression in it fails a run against a baseline.

    The flag is what the document publishes. The limits themselves are held by the recipe's
    benchmark lists and by the configuration the benchmark declares.
    """

    def __init__(self, name, framework, gated=True, note=None):
        self.name = name
        self.framework = framework
        self.gated = gated
        self.note = note


# The three runs of every scenario, by the benchmark id that carries each: the cold start on its
# own, and the two counts whose difference is the steady state.
COLD = "first"
COUNTS = ("base", "twice")

# The table, in reading order. A scenario names its benchmark as `file/function`: one scenario
# per benchmark file, measured at both counts.
SCENARIOS = [
    Scenario(
        "consume, JSON decode into a small struct",
        "consume_json/service",
    ),
    Scenario(
        "consume, JSON decode of a 1 KB body",
        "consume_json_kilobyte/service",
    ),
    Scenario(
        "consume on the byte lane, no codec",
        "consume_lane/service",
    ),
    Scenario(
        "consume through a pool of four workers",
        "consume_pool/service",
    ),
    Scenario(
        "consume on four dedicated threads",
        "consume_threads/service",
    ),
    Scenario(
        "consume through a middleware stack of one",
        "middleware/service_one",
    ),
    Scenario(
        "consume through a middleware stack of four",
        "middleware/service_four",
    ),
    Scenario(
        "consume in batches of 64",
        "batch/service",
    ),
    Scenario(
        "reply, encoded to a declared destination",
        "reply/service",
    ),
    Scenario(
        "reply to a transport that reads the payload",
        "reply_lending/service",
    ),
    Scenario(
        "publish through an Out slot with one transform",
        "out_slot/service",
    ),
    Scenario(
        "publish with a typed header contract",
        "typed_headers_write/service",
    ),
    Scenario(
        "read a typed header contract, then publish",
        "typed_headers_read/service",
    ),
    Scenario(
        "request and reply, one round trip",
        "request_reply/service",
    ),
    Scenario(
        "a delivery that asks to be redelivered, natively by the broker",
        "retry_copy/service",
        gated=False,
        note="cold path: measured and reported, never gated",
    ),
]


def metric(summary, tool, name):
    """The new value of one metric: the total of one tool's run, as the runner reports it."""
    for profile in summary["profiles"]:
        if profile["tool"] != tool:
            continue
        values = profile["data"]["total"]["metrics"].get(name, {}).get("values", {})
        # A run compared against a baseline carries the old value next to the new one.
        new = values.get("new")
        return None if new is None else int(new)
    return None


def summaries(path):
    """Every benchmark summary the run wrote, one per line, in the layout this script reads."""
    found = []
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line:
            continue
        summary = json.loads(line)
        version = summary.get("version")
        if version != SUMMARY_VERSION:
            sys.exit(
                f"the benchmark summary has layout version {version}, and this script reads "
                f"version {SUMMARY_VERSION}: read the new layout in `metric` and `breaches` "
                "and raise SUMMARY_VERSION"
            )
        found.append(summary)
    return found


def benchmark(summary):
    """The `file/function` a summary belongs to, which is how a scenario names its benchmark."""
    return f"{Path(summary['benchmark_file']).stem}/{summary['function_name']}"


def measurements(runs):
    """Every benchmark in the run, keyed by `file/function/id`.

    The file is part of the key because one scenario per file means the same two function names
    in every one of them.
    """
    found = {}
    for summary in runs:
        found[f"{benchmark(summary)}/{summary['id']}"] = {
            "instructions": metric(summary, "Callgrind", "Ir"),
            "allocations": metric(summary, "DHAT", "TotalBlocks"),
        }
    return found


# The two metrics the table reads, by the names it gives them. A limit on any other metric is
# reported under the runner's own name for it.
METRIC_NAMES = {("Callgrind", "Ir"): "instructions", ("Dhat", "TotalBlocks"): "allocations"}


def run_name(run):
    """A benchmark id as the number of deliveries its run handled."""
    counts = {COLD: 1, COUNTS[0]: MESSAGES, COUNTS[1]: 2 * MESSAGES}
    if run not in counts:
        return run
    return "one delivery" if counts[run] == 1 else f"{counts[run]} deliveries"


def as_text(value):
    """A metric value as a breach line writes it: a count as it is, a fraction in short form."""
    return str(value) if isinstance(value, int) else f"{value:g}"


def breach(regression, metrics):
    """One limit a run went over: the metric, the value it was compared against, the new one.

    A limit in percent holds the run to the one it is compared against, and the regression
    carries both values. A plain number is a ceiling the run is held to on its own, and the value
    it was compared against is the one the metric records next to the new one, where there is one.
    """
    [(kind, detail)] = regression.items()
    [(tool, name)] = detail["metric"].items()
    label = METRIC_NAMES.get((tool, name), f"{tool} {name}")
    if kind == "Soft":
        return (
            f"{label} {as_text(detail['old'])} -> {as_text(detail['new'])}, "
            f"{float(detail['diff_pct']):+.2f}% against a limit of +{float(detail['limit']):g}%"
        )
    old = metrics.get(name, {}).get("values", {}).get("old")
    change = "" if old is None else f"{as_text(old)} -> "
    return f"{label} {change}{as_text(detail['new'])} against a limit of {as_text(detail['limit'])}"


def breaches(runs):
    """Every limit the run breached, one line each, named by its scenario and its run."""
    names = {scenario.framework: scenario.name for scenario in SCENARIOS}
    found = []
    for summary in runs:
        where = f"{names.get(benchmark(summary), benchmark(summary))}, {run_name(summary['id'])}"
        for profile in summary["profiles"]:
            total = profile["data"]["total"]
            for regression in total["regressions"]:
                found.append(f"{where}: {breach(regression, total['metrics'])}")
    return found


def rounded(figure):
    """A per-message figure at the precision the published tables read it in.

    A cost that falls below one per message is shown to three places, so that "nothing per
    message, one allocation for the whole run" does not read as a flat zero.
    """
    return round(figure, 3) if abs(figure) < 1 else round(figure, 1)


def command(*args):
    return subprocess.run(args, capture_output=True, text=True, check=True).stdout.strip()


def read(path, default=None):
    """One line of a sysfs or proc file, or `default` where the file is not there."""
    try:
        return Path(path).read_text().strip()
    except OSError:
        return default


def proc_cpuinfo():
    return read("/proc/cpuinfo", "") or ""


def cpu_model():
    for line in proc_cpuinfo().splitlines():
        if line.startswith("model name"):
            return line.split(":", 1)[1].strip()
    return platform.processor() or "unknown"


# The feature sets the x86-64 levels are defined by, coarsest first. A level is claimed only when
# every flag of it and of the levels below is present, so this names what the binary could have
# been built for rather than guessing a marketing name for the chip.
X86_LEVELS = [
    ("x86-64-v2", ["sse4_2", "popcnt", "ssse3"]),
    ("x86-64-v3", ["avx2", "bmi1", "bmi2", "fma", "movbe"]),
    ("x86-64-v4", ["avx512f", "avx512bw", "avx512cd", "avx512dq", "avx512vl"]),
]


def architecture():
    machine = platform.machine()
    flags = set()
    for line in proc_cpuinfo().splitlines():
        if line.startswith("flags"):
            flags = set(line.split(":", 1)[1].split())
            break
    level = None
    for name, needed in X86_LEVELS:
        if all(flag in flags for flag in needed):
            level = name
        else:
            break
    return f"{machine} ({level})" if level else machine


def megahertz(path):
    value = read(path)
    try:
        return round(int(value) / 1000)
    except (TypeError, ValueError):
        return None


def cpu_frequency():
    """Base and maximum clock, as the kernel reports them.

    A CPU whose firmware does not publish a base clock (every AMD one here) says so rather than
    having a number invented for it.
    """
    cpufreq = "/sys/devices/system/cpu/cpu0/cpufreq"
    base = megahertz(f"{cpufreq}/base_frequency")
    top = megahertz(f"{cpufreq}/cpuinfo_max_freq")
    bottom = megahertz(f"{cpufreq}/cpuinfo_min_freq")
    parts = [f"base {base} MHz" if base else "base unknown"]
    if bottom:
        parts.append(f"min {bottom} MHz")
    parts.append(f"max {top} MHz" if top else "max unknown")
    return ", ".join(parts)


def cores():
    """Physical and logical count, counted off the topology the kernel publishes."""
    physical = set()
    package = core = None
    for line in proc_cpuinfo().splitlines():
        if line.startswith("physical id"):
            package = line.split(":", 1)[1].strip()
        elif line.startswith("core id"):
            core = line.split(":", 1)[1].strip()
            physical.add((package, core))
    logical = os.cpu_count() or 0
    count = len(physical) or logical
    return f"{count} physical, {logical} logical"


def memory():
    for line in (read("/proc/meminfo", "") or "").splitlines():
        if line.startswith("MemTotal"):
            kib = int(line.split()[1])
            return f"{kib / 1024 / 1024:.1f} GiB"
    return "unknown"


def memory_speed(previous=None):
    """Module speed and type, out of the DMI tables where they can be read at all.

    They are root-only on most Linux systems, and a benchmark run is not worth a root shell. A
    value a run cannot read is not lost, though: where the document already published one and the
    machine still looks like the one it was published from, it is carried over, so an ordinary run
    does not overwrite what a privileged one found. Otherwise the answer is `unknown` rather than
    a guess.
    """
    try:
        dmi = subprocess.run(
            ["dmidecode", "--type", "memory"], capture_output=True, text=True, timeout=10
        )
    except (OSError, subprocess.SubprocessError):
        return previous or "unknown"
    if dmi.returncode != 0:
        return previous or "unknown"
    speeds = set()
    kinds = set()
    for line in dmi.stdout.splitlines():
        line = line.strip()
        if line.startswith("Configured Memory Speed:") or line.startswith("Speed:"):
            value = line.split(":", 1)[1].strip()
            if value and "Unknown" not in value:
                speeds.add(value)
        elif line.startswith("Type:") and "Unknown" not in line:
            kinds.add(line.split(":", 1)[1].strip())
    if not speeds:
        return previous or "unknown"
    return ", ".join(sorted(kinds | speeds))


# What cargo compiles a release with when the manifest says nothing. The benchmark profile
# inherits release, so this is where its description starts.
RELEASE_DEFAULTS = {"opt-level": 3, "lto": False, "codegen-units": 16, "debug": False}


def profile():
    """The profile the benchmarks are built with, as it actually resolves."""
    manifest = tomllib.loads((REPO / "Cargo.toml").read_text())
    profiles = manifest.get("profile", {})
    resolved = dict(RELEASE_DEFAULTS)
    resolved.update(profiles.get("release", {}))
    resolved.update(profiles.get("bench", {}))
    resolved.pop("inherits", None)
    settings = ", ".join(
        f"{key} = {str(value).lower() if isinstance(value, bool) else value}"
        for key, value in resolved.items()
    )
    return f"bench, inheriting release ({settings})"


def features():
    """The feature list the benchmarks are built with, read where it is defined."""
    justfile = (REPO / "justfile").read_text()
    match = re.search(r'^bench_features := "([^"]+)"', justfile, re.MULTILINE)
    if not match:
        sys.exit("cannot read bench_features from the justfile")
    return f"--no-default-features --features {match.group(1)}"


def published_memory_speed(output):
    """The memory speed the document already carries, when it describes this machine.

    Matched on the processor and the memory size: a document written on another machine says
    nothing about this one's modules.
    """
    try:
        previous = json.loads(output.read_text()).get("environment", {})
    except (OSError, ValueError):
        return None
    if previous.get("cpu") != cpu_model() or previous.get("memory") != memory():
        return None
    speed = previous.get("memory_speed")
    return None if speed in (None, "unknown") else speed


def environment(output):
    """The machine and the build, so a number can be read against what produced it."""
    valgrind = command("valgrind", "--version")
    return {
        "cpu": cpu_model(),
        "architecture": architecture(),
        "cpu_frequency": cpu_frequency(),
        "cores": cores(),
        "memory": memory(),
        "memory_speed": memory_speed(published_memory_speed(output)),
        "os": f"{platform.system()} {platform.release()}",
        "rustc": command("rustc", "--version").split()[1],
        "valgrind": valgrind.removeprefix("valgrind-"),
        "profile": profile(),
        "features": features(),
        "rustflags": "",
    }


def crate_version():
    manifest = (REPO / "Cargo.toml").read_text()
    match = re.search(r'^version = "([^"]+)"', manifest, re.MULTILINE)
    if not match:
        sys.exit("cannot read the crate version from Cargo.toml")
    return match.group(1)


def totals(found, key, count, floor=None):
    """One benchmark's totals, checked for the two ways this measurement fails silently."""
    if floor is None:
        floor = FLOOR
    full = f"{key}/{count}"
    if full not in found:
        sys.exit(
            f"benchmark {full} is not in the run: it failed before it wrote a summary, or it was "
            "renamed (then rename it here or in benches/)"
        )
    measured = found[full]
    if measured["instructions"] is None or measured["instructions"] < floor:
        sys.exit(
            f"benchmark {full} reports {measured['instructions']} instructions, which is below "
            f"the floor of {floor}: collection did not cover the measured region"
        )
    return measured


def figures(found, key):
    """The steady state and the cold start of one scenario.

    The slope between the two counts is what a message costs once the service is running; the
    one-delivery run is what starting it and taking that delivery cost.
    """
    base = totals(found, key, COUNTS[0])
    twice = totals(found, key, COUNTS[1])
    if twice["instructions"] <= base["instructions"]:
        sys.exit(
            f"benchmark {key} does not grow with the message count "
            f"({base['instructions']} -> {twice['instructions']}): the two runs measured the "
            f"same work, so the slope means nothing"
        )
    first = totals(found, key, COLD, COLD_FLOOR)
    steady = {}
    for metric_name in ("instructions", "allocations"):
        steady[metric_name] = rounded((twice[metric_name] - base[metric_name]) / MESSAGES)
    cold = {name: first[name] for name in ("instructions", "allocations")}
    return steady, cold


def build(found):
    rows = []
    for scenario in SCENARIOS:
        framework, cold = figures(found, scenario.framework)
        row = {
            "name": scenario.name,
            "messages": MESSAGES,
            "framework": framework,
            "cold": cold,
            "gated": scenario.gated,
        }
        if scenario.note:
            row["note"] = scenario.note
        rows.append(row)
    return rows


def report(rows):
    """The same table the page publishes, for the terminal."""
    header = (
        f"{'scenario':<52}{'instructions':>14}{'allocations':>13}"
        f"{'cold instr':>12}{'cold alloc':>12}"
    )
    print(header)
    print("-" * len(header))
    for row in rows:
        cold = row["cold"]
        print(
            f"{row['name']:<52}"
            f"{row['framework']['instructions']:>14}"
            f"{row['framework']['allocations']:>13}"
            f"{cold['instructions']:>12}"
            f"{cold['allocations']:>12}"
        )
    print()
    print(
        f"instructions and allocations per message in the steady state over {MESSAGES} "
        "deliveries; cold is what starting"
    )
    print("the service and handling the first delivery cost once")


def report_breaches(lines):
    """The limits the run breached, which is why it fails, each with both values it compared."""
    if not lines:
        return
    print()
    print("limits breached (totals of one run, old -> new):")
    for line in lines:
        print(f"  {line}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("summary", type=Path, help="the JSON the benchmark run wrote")
    parser.add_argument("output", type=Path, help="where to write the results document")
    parser.add_argument(
        "--messages",
        type=int,
        default=DEFAULT_MESSAGES,
        help="deliveries per measured run, the count the benches were built with",
    )
    args = parser.parse_args()
    configure(args.messages)

    runs = summaries(args.summary)
    breached = breaches(runs)
    try:
        code = build(measurements(runs))
    except SystemExit:
        # One failure does not hide another: a summary that cannot be converted still shows what
        # the run breached.
        report_breaches(breached)
        sys.stdout.flush()
        raise

    version = crate_version()
    document = {
        "schema": 2,
        "crate": "ruststream",
        "crate_version": version,
        "core_version": version,
        "measured_at": datetime.now(timezone.utc).date().isoformat(),
        "environment": environment(args.output),
        # No `scenarios`: that section is a comparison against a broker's own client, and this
        # crate has no broker of its own to compare on. An empty list would claim a measurement
        # that is not coming.
        "code": code,
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(document, indent=2) + "\n")
    report(document["code"])
    report_breaches(breached)


if __name__ == "__main__":
    main()