oak/benchmarks

"""git-sizer-style repository statistics, re-implemented in stdlib Python.

Metric names, definitions and "level of concern" reference values follow

github/git-sizer (MIT, Copyright (c) 2018 GitHub; see suites/scale/third_party/git-sizer/NOTICE and

docs/prior-art.md). This is an independent re-implementation: no git-sizer

code is copied. Values are gathered with plain git plumbing

(``cat-file --batch-check --batch-all-objects``, ``log --all``, ``cat-file

--batch`` for trees and tags) and recorded in every scale row's provenance.

Differences from git-sizer, stated so a number is never misread:

- object totals count every object in the object database

  (``object_scope: all_objects_in_odb``); git-sizer counts objects reachable

  from references. Generated fixtures contain no unreachable objects, so the

  two agree on them;

- "biggest checkout" metrics are the maximum over every commit's root tree

  (git-sizer semantics, ``checkout_scope: all_commits``) while the number of

  distinct root trees is at most ``max_checkout_roots``; above that the

  maximum is taken over reference tips only (``checkout_scope: ref_tips``),

  which is a lower bound, and the row says so.

"""

from __future__ import annotations

import subprocess

import time

from pathlib import Path

from typing import Any, Dict, List, Optional, Tuple

from .generator import git_env

SIZER_VERSION = "scale-sizer-2026-09-28.1"

MAX_CHECKOUT_ROOTS = 20_000

# (symbol, unit, git-sizer reference value for one "star" of concern).

METRICS: Tuple[Tuple[str, str, float], ...] = (

    ("uniqueCommitCount", "", 500e3),

    ("uniqueCommitSize", "B", 250e6),

    ("uniqueTreeCount", "", 1.5e6),

    ("uniqueTreeSize", "B", 2e9),

    ("uniqueTreeEntries", "", 50e6),

    ("uniqueBlobCount", "", 1.5e6),

    ("uniqueBlobSize", "B", 10e9),

    ("uniqueTagCount", "", 25e3),

    ("referenceCount", "", 25e3),

    ("maxCommitSize", "B", 50e3),

    ("maxCommitParentCount", "", 10),

    ("maxTreeEntries", "", 1000),

    ("maxBlobSize", "B", 10e6),

    ("maxHistoryDepth", "", 500e3),

    ("maxTagDepth", "", 1.001),

    ("maxCheckoutTreeCount", "", 2000),

    ("maxCheckoutPathDepth", "", 10),

    ("maxCheckoutPathLength", "B", 100),

    ("maxCheckoutBlobCount", "", 50e3),

    ("maxCheckoutBlobSize", "B", 1e9),

    ("maxCheckoutLinkCount", "", 25e3),

    ("maxCheckoutSubmoduleCount", "", 100),

)

REFERENCE_VALUES = {symbol: scale for symbol, _, scale in METRICS}

def level_of_concern(symbol: str, value: Optional[float]) -> Optional[float]:

    """value / reference value (git-sizer prints one '*' per whole unit >= 1)."""

    if value is None:

        return None

    return round(float(value) / REFERENCE_VALUES[symbol], 4)

def parse_tree(raw: bytes) -> List[Tuple[bytes, bytes, bytes]]:

    """Binary tree object -> [(mode, name, 20-byte oid)]."""

    entries = []

    pos = 0

    size = len(raw)

    while pos < size:

        space = raw.index(b" ", pos)

        nul = raw.index(b"\0", space)

        entries.append((raw[pos:space], raw[space + 1:nul], raw[nul + 1:nul + 21]))

        pos = nul + 21

    return entries

class _Batch:

    """A persistent ``git cat-file --batch`` reader."""

    def __init__(self, git_bin: str, repo: Path) -> None:

        self.proc = subprocess.Popen([git_bin, "-C", str(repo), "cat-file", "--batch"],

                                     stdin=subprocess.PIPE, stdout=subprocess.PIPE, env=git_env())

    def read(self, oid_hex: str) -> Tuple[str, bytes]:

        assert self.proc.stdin and self.proc.stdout

        self.proc.stdin.write(oid_hex.encode() + b"\n")

        self.proc.stdin.flush()

        header = self.proc.stdout.readline().split()

        if len(header) < 3:

            raise RuntimeError(f"cat-file --batch: missing object {oid_hex}")

        size = int(header[2])

        data = self.proc.stdout.read(size)

        self.proc.stdout.read(1)

        return header[1].decode(), data

    def close(self) -> None:

        if self.proc.stdin:

            self.proc.stdin.close()

        if self.proc.stdout:

            self.proc.stdout.close()

        self.proc.wait()

class _TreeStats:

    __slots__ = ("trees", "blobs", "blob_size", "links", "submodules", "depth", "path_len")

    def __init__(self) -> None:

        self.trees = 1

        self.blobs = 0

        self.blob_size = 0

        self.links = 0

        self.submodules = 0

        self.depth = 0

        self.path_len = 0

def measure(repo: Path, git_bin: str = "git", max_checkout_roots: int = MAX_CHECKOUT_ROOTS) -> Dict[str, Any]:

    """Compute the git-sizer metric set for ``repo`` (bare or not)."""

    started = time.perf_counter()

    env = git_env()

    sizes: Dict[bytes, int] = {}

    counts = {"commit": 0, "tree": 0, "blob": 0, "tag": 0}

    totals = {"commit": 0, "tree": 0, "blob": 0, "tag": 0}

    max_commit = max_blob = 0

    tag_ids: List[str] = []

    proc = subprocess.Popen([git_bin, "-C", str(repo), "cat-file", "--batch-check", "--batch-all-objects",

                             "--unordered"], stdout=subprocess.PIPE, env=env)

    assert proc.stdout is not None

    for line in proc.stdout:

        oid, kind, size_text = line.split()

        kind_s = kind.decode()

        size = int(size_text)

        counts[kind_s] = counts.get(kind_s, 0) + 1

        totals[kind_s] = totals.get(kind_s, 0) + size

        if kind_s == "blob":

            sizes[bytes.fromhex(oid.decode())] = size

            max_blob = max(max_blob, size)

        elif kind_s == "commit":

            max_commit = max(max_commit, size)

        elif kind_s == "tag":

            tag_ids.append(oid.decode())

    proc.stdout.close()

    if proc.wait() != 0:

        raise RuntimeError("git cat-file --batch-all-objects failed")

    # History: generation depth over parents, walked parents-first.

    depth: Dict[str, int] = {}

    roots: Dict[str, None] = {}

    max_parents = 0

    log = subprocess.Popen([git_bin, "-C", str(repo), "log", "--all", "--topo-order", "--reverse",

                            "--format=%H %T %P"], stdout=subprocess.PIPE, env=env)

    assert log.stdout is not None

    for line in log.stdout:

        parts = line.decode().split()

        if len(parts) < 2:

            continue

        commit, tree, parents = parts[0], parts[1], parts[2:]

        max_parents = max(max_parents, len(parents))

        depth[commit] = 1 + max((depth.get(parent, 0) for parent in parents), default=0)

        roots[tree] = None

    log.stdout.close()

    log.wait()

    max_history = max(depth.values(), default=0)

    refs = subprocess.run([git_bin, "-C", str(repo), "for-each-ref", "--format=%(objectname) %(objecttype)"],

                          check=True, capture_output=True, text=True, env=env).stdout.split("\n")

    ref_lines = [line for line in refs if line]

    batch = _Batch(git_bin, repo)

    try:

        # Tag chains.

        tag_depth_memo: Dict[str, int] = {}

        def tag_depth(oid: str) -> int:

            if oid in tag_depth_memo:

                return tag_depth_memo[oid]

            _, data = batch.read(oid)

            header = dict(line.split(" ", 1) for line in data.split(b"\n\n", 1)[0].decode().split("\n") if " " in line)

            value = 1 + (tag_depth(header["object"]) if header.get("type") == "tag" else 0)

            tag_depth_memo[oid] = value

            return value

        max_tag_depth = max((tag_depth(oid) for oid in tag_ids), default=0)

        if len(roots) <= max_checkout_roots:

            scope = "all_commits"

            root_trees = list(roots)

        else:

            scope = "ref_tips"

            tips = subprocess.run([git_bin, "-C", str(repo), "for-each-ref", "--format=%(objectname)^{tree}"],

                                  check=True, capture_output=True, text=True, env=env).stdout.split()

            root_trees = sorted(set(subprocess.run(

                [git_bin, "-C", str(repo), "rev-parse"] + tips, check=True, capture_output=True, text=True,

                env=env).stdout.split())) if tips else []

        memo: Dict[bytes, _TreeStats] = {}

        max_tree_entries = 0

        tree_entries_total = 0

        def stats_for(root_hex: str) -> _TreeStats:

            nonlocal max_tree_entries, tree_entries_total

            root = bytes.fromhex(root_hex)

            stack: List[Tuple[bytes, bool]] = [(root, False)]

            pending: Dict[bytes, List[Tuple[bytes, bytes, bytes]]] = {}

            while stack:

                oid, expanded = stack.pop()

                if oid in memo:

                    continue

                if not expanded:

                    _, raw = batch.read(oid.hex())

                    entries = parse_tree(raw)

                    pending[oid] = entries

                    max_tree_entries = max(max_tree_entries, len(entries))

                    tree_entries_total += len(entries)

                    stack.append((oid, True))

                    for mode, _, child in entries:

                        if mode == b"40000" and child not in memo:

                            stack.append((child, False))

                    continue

                entries = pending.pop(oid)

                stats = _TreeStats()

                for mode, name, child in entries:

                    if mode == b"40000":

                        sub = memo[child]

                        stats.trees += sub.trees

                        stats.blobs += sub.blobs

                        stats.blob_size += sub.blob_size

                        stats.links += sub.links

                        stats.submodules += sub.submodules

                        stats.depth = max(stats.depth, 1 + sub.depth)

                        stats.path_len = max(stats.path_len, len(name) + 1 + sub.path_len)

                    else:

                        if mode == b"120000":

                            stats.links += 1

                        elif mode == b"160000":

                            stats.submodules += 1

                        else:

                            stats.blobs += 1

                            stats.blob_size += sizes.get(child, 0)

                        stats.depth = max(stats.depth, 1)

                        stats.path_len = max(stats.path_len, len(name))

                memo[oid] = stats

            return memo[root]

        best = {"trees": 0, "blobs": 0, "blob_size": 0, "links": 0, "submodules": 0, "depth": 0, "path_len": 0}

        for root_hex in root_trees:

            stats = stats_for(root_hex)

            for key in best:

                best[key] = max(best[key], getattr(stats, key))

    finally:

        batch.close()

    values: Dict[str, Any] = {

        "uniqueCommitCount": counts.get("commit", 0),

        "uniqueCommitSize": totals.get("commit", 0),

        "uniqueTreeCount": counts.get("tree", 0),

        "uniqueTreeSize": totals.get("tree", 0),

        # Tree entries are only counted for trees reachable from the scanned

        # roots; with checkout_scope all_commits that is every commit tree.

        "uniqueTreeEntries": tree_entries_total if scope == "all_commits" else None,

        "uniqueBlobCount": counts.get("blob", 0),

        "uniqueBlobSize": totals.get("blob", 0),

        "uniqueTagCount": counts.get("tag", 0),

        "referenceCount": len(ref_lines),

        "maxCommitSize": max_commit,

        "maxCommitParentCount": max_parents,

        "maxTreeEntries": max_tree_entries,

        "maxBlobSize": max_blob,

        "maxHistoryDepth": max_history,

        "maxTagDepth": max_tag_depth,

        "maxCheckoutTreeCount": best["trees"],

        "maxCheckoutPathDepth": best["depth"],

        "maxCheckoutPathLength": best["path_len"],

        "maxCheckoutBlobCount": best["blobs"],

        "maxCheckoutBlobSize": best["blob_size"],

        "maxCheckoutLinkCount": best["links"],

        "maxCheckoutSubmoduleCount": best["submodules"],

    }

    concern = {symbol: level_of_concern(symbol, values[symbol]) for symbol, _, _ in METRICS}

    return {

        "sizer_version": SIZER_VERSION,

        "metric_definitions": "github/git-sizer (MIT) re-implemented; see suites/scale/third_party/git-sizer/NOTICE",

        "object_scope": "all_objects_in_odb",

        "checkout_scope": scope,

        "checkout_roots_scanned": len(root_trees),

        "values": values,

        "level_of_concern": concern,

        "max_level_of_concern": max((v for v in concern.values() if v is not None), default=0.0),

        "elapsed_s": round(time.perf_counter() - started, 3),

    }

def compact(stats: Dict[str, Any]) -> Dict[str, Any]:

    """The subset copied into every row (full stats live in the fixture dir)."""

    values = stats["values"]

    keep = ("uniqueCommitCount", "uniqueBlobCount", "uniqueBlobSize", "referenceCount", "maxHistoryDepth",

            "maxCheckoutBlobCount", "maxCheckoutBlobSize", "maxCheckoutPathDepth", "maxCheckoutPathLength",

            "maxCheckoutTreeCount", "maxTreeEntries", "maxBlobSize")

    return {"sizer_version": stats["sizer_version"], "checkout_scope": stats["checkout_scope"],

            **{key: values.get(key) for key in keep},

            "max_level_of_concern": stats["max_level_of_concern"]}