cs_survival_kit.bench ¶

def case[T](
    self,
    label: str,
    *,
    setup: Callable[[int], T],
    run: Callable[[T], object],
    sizes: Sequence[int] | None = None,
) -> None:
    """Add a case to the benchmark.

    Args:
        label: A name for the case, unique within the benchmark.
        setup: Builds the inputs for a size `n`. Not timed. It is called
            again before every timed call, so `run` is free to mutate
            what it is given.
        run: The code to time. It receives whatever `setup` returned.
        sizes: Sizes for this case only, overriding the benchmark's. Use
            it to cap a slow case at smaller inputs.

    Raises:
        ValueError: If `label` is already taken, or `sizes` is invalid.
    """
    if any(case.label == label for case in self._cases):
        raise ValueError(f"duplicate case label: {label!r}")
    case_sizes = self.sizes if sizes is None else _validate_sizes(sizes)
    self._cases.append(_Case(label, setup, run, case_sizes))

def run(self, repeat: int = 5, *, smoke: bool = False) -> Results:
    """Time every case at every size.

    Each measurement calls `run` in a loop until the timed calls add up
    to about 0.1 seconds, so that fast calls are not lost in timer noise.
    The reported time is per call, and is the minimum over `repeat`
    measurements: the minimum is the run least disturbed by the rest of
    the machine. Garbage collection is disabled while timing.

    A case that raises `NotImplementedError` is reported as
    `"not implemented"` and skipped, so benchmarks can be written before
    the code they measure.

    Args:
        repeat: How many measurements to take per size.
        smoke: Only check that the cases execute: time the smallest size
            once, with a single call. The numbers are meaningless.

    Returns:
        The measurements for every case.

    Raises:
        ValueError: If `repeat` is less than 1.
    """
    if repeat < 1:
        raise ValueError("repeat must be at least 1")
    run_at = datetime.now(UTC).isoformat(timespec="seconds")
    cases = [_run_case(case, repeat, smoke) for case in self._cases]
    return Results(
        self.name, run_at, __version__, _environment(), cases, self.per_item
    )

def fit(self) -> dict[str, Fit]:
    """Estimate each case's growth from its measurements.

    The slope of time against size on a log-log scale approximates the
    exponent `k` in `O(n^k)`: about 0 is constant, about 1 is linear,
    about 2 is quadratic. `O(n log n)` has no exponent of its own and
    reads as slightly above 1.

    This is an empirical sanity check, not a proof. Constant factors,
    caches and small sizes all bend the line.

    Returns:
        A mapping from case label to its fit.
    """
    fits: dict[str, Fit] = {}
    for case in self.cases:
        if case.status != OK:
            growth = case.status
        elif case.slope is None:
            growth = "-"
        else:
            growth = describe_slope(case.slope)
        fits[case.label] = Fit(case.slope, growth)
    return fits

def format(self) -> str:
    """Render the results as a plain-text table.

    Returns:
        One row per size and one column per case, followed by each
        case's slope and growth. For a `per_item` benchmark, a second
        table follows with every time divided by its size: the amortized
        cost of one operation. A flat column there means constant cost
        per operation; a growing one means each operation gets more
        expensive as the input grows.
    """
    fits = self.fit()
    sizes = sorted({n for case in self.cases for n in case.sizes})
    columns: list[list[str]] = [
        ["n", *(f"{n:,}" for n in sizes), "slope", "growth"]
    ]
    per_item_columns: list[list[str]] = [["n", *(f"{n:,}" for n in sizes)]]
    for case in self.cases:
        times = dict(zip(case.sizes, case.seconds, strict=True))
        fit = fits[case.label]
        columns.append(
            [
                case.label,
                *(_format_seconds(times[n]) if n in times else "-" for n in sizes),
                "-" if fit.slope is None else f"{fit.slope:.2f}",
                fit.growth,
            ]
        )
        per_item_columns.append(
            [
                case.label,
                *(
                    _format_seconds(times[n] / n) if n in times else "-"
                    for n in sizes
                ),
            ]
        )
    lines = [self.name, *_format_columns(columns)]
    if self.per_item:
        lines += ["", "per item (time / n)", *_format_columns(per_item_columns)]
    return "\n".join(lines)

def table(self) -> None:
    """Print the results as a plain-text table."""
    print(self.format())

def to_dict(self) -> dict[str, Any]:
    """Convert the results to a JSON-serializable dictionary.

    Returns:
        The benchmark entry stored in `benchmarks.json`.
    """
    return asdict(self)