crates/storage-git-gix/parked-gix/.nornir/bench/affinity-arm.py
#!/usr/bin/env python3
"""Where a CPU budget is allowed to RUN, measured on gunnar alone.
`bench/examples/forge/concurrent.rs` answers the same question with three
engines, three containers and a counting relay, and it takes half an hour. This
answers only the affinity half and takes two minutes, so a hypothesis about
`cpu.max`-versus-cpuset can be tried without booting gitea.
One `gunnar serve`, one cgroup CPU quota, one production-shaped corpus (many
packs, no repack, no bitmap), N concurrent `git clone`s over smart HTTP. What
changes between arms is the server's CPU affinity — `0-31`, the quota spread
over the whole box, which is what `podman run --cpus 8` gives, versus `0-7`, the
same quota as a place — and, optionally, the binary and the admission ceiling,
so a fix can be A/B'd against its own parent commit in one interleaved run.
Arms are interleaved and their order alternates every round, because on oden
load1 moves 20 -> 300 inside one measurement and a fixed arm order turns that
trend into a result. Every row carries load1 before and after itself.
TP_ARMS=new:0-31:256,new:0-7:256 <build>:<affinity>:<max-transfers>
TP_BINS via BINS below which binary each build name is
TP_QUOTA=8 TP_CLIENTS=32 TP_ROUNDS=3
TP_CONTAINER=1 boot in podman as the forge arm does
TP_RED=<guard> drive one guard red
GUARDS, and all five have been seen red (oden, 2026-08-06). Four of the five are
driven by a REAL condition, not by faking the observation:
clients every client exited 0 RED: one client asks for a
repository that is not there
objects every client's clone holds the RED: one client clones --depth 1
corpus's object count and tip
quota the cgroup really carries the RED: boot the scope with no
quota the arm claims CPUQuota at all
affinity every THREAD of the server RED: pin to 0-3 while the arm
carries the arm's claims its own set
Cpus_allowed_list, read after
the batch as well as before
cpu the cgroup burned CPU RED: synthetic — this one cannot
be driven by a real condition
without stopping the server
`objects` matters more than it looks: a client that transfers nothing finishes
first and RAISES the batch's ops/s, so a failed batch is never a slow result,
it is a fictional one.
Setup is `setup.sh` + `seed.sh` beside this file's scratch copy; the corpus is
built by pushing a real repository in twelve increments, because `git clone
--bare` repacks and writes no bitmap and would give a single-pack corpus that
cannot reproduce multi-pack behaviour.
"""
import json
import os
import re
import subprocess
import sys
import time
import shutil
from concurrent.futures import ThreadPoolExecutor
# Which binary each build name in TP_ARMS refers to. `TP_BINS=new=/path,old=/path`.
# NO DEFAULT that resolves to "whatever was lying around": a measurement whose
# subject was chosen by a fallback is a true number about the wrong build, which
# is the trap `forge/mod.rs::gunnar_binary` refuses for the same reason.
BINS = dict(
kv.split("=", 1)
for kv in os.environ.get("TP_BINS", "").split(",")
if "=" in kv
)
if not BINS:
print("TP_BINS is unset. Name the build(s) under test:\n"
" CARGO_TARGET_DIR=<dir> cargo build --release -p gunnar-cli --features http --bin gunnar\n"
" TP_BINS=new=<dir>/release/gunnar[,old=<other>/release/gunnar]", file=sys.stderr)
raise SystemExit(2)
BIN = next(iter(BINS.values()))
ROOT = os.environ.get("TP_ROOT", "/home/rickard/scratch/gunnar-tp")
DATA = f"{ROOT}/data"
PORT = int(os.environ.get("TP_PORT", "18899"))
SCOPE = "gunnar-tp"
CLIENTS = int(os.environ.get("TP_CLIENTS", "32"))
ROUNDS = int(os.environ.get("TP_ROUNDS", "3"))
QUOTA_CPUS = int(os.environ.get("TP_QUOTA", "8"))
# (label, affinity, --max-concurrent-transfers)
def _arms():
raw = os.environ.get("TP_ARMS", f"new:0-{os.cpu_count()-1}:256,new:0-{QUOTA_CPUS-1}:256")
out = []
for spec in raw.split(","):
which, cpus, maxt = spec.split(":")
out.append((f"{which}/cpus{cpus}/t{maxt}", which, cpus, int(maxt)))
return out
ARMS = _arms()
RED = set(os.environ.get("TP_RED", "").split(",")) - {""}
CONTAINER = os.environ.get("TP_CONTAINER") == "1"
IMAGE = os.environ.get("TP_IMAGE", "localhost/gunnar-bench-git:2.53.0")
CNAME = "gunnar-tp-bench"
def podman(*a, check=True):
r = subprocess.run(["podman", *a], capture_output=True, text=True)
if check and r.returncode != 0:
die(f"podman {' '.join(a)} exited {r.returncode}: {r.stderr.strip()[:400]}")
return r.stdout.strip()
def start_container(maxt, binary):
podman("rm", "-f", CNAME, check=False)
bindir = os.path.dirname(binary)
podman("run", "-d", "--name", CNAME,
"--cpus", str(QUOTA_CPUS), "--memory", "8192m",
"-p", f"127.0.0.1:{PORT}:{PORT}",
"-v", f"{DATA}:{DATA}", "-v", f"{bindir}:{bindir}:ro",
"-e", f"TOKIO_WORKER_THREADS={QUOTA_CPUS}",
IMAGE,
binary, "--data-root", DATA, "serve",
"--http-bind", f"0.0.0.0:{PORT}",
"--http-allow-cleartext-credentials", "--no-grpc",
"--ssh-bind", "127.0.0.1:2222",
"--max-concurrent-transfers", str(maxt),
"--max-transfers-per-principal", str(maxt))
for _ in range(120):
r = subprocess.run(
["curl", "-fsS", "-o", "/dev/null", "--user", f"bencher:{token()}",
f"http://127.0.0.1:{PORT}/bench/conc.git/info/refs?service=git-upload-pack"],
capture_output=True)
if r.returncode == 0:
return
time.sleep(0.5)
die("the container never answered info/refs: " + podman("logs", "--tail", "20", CNAME, check=False))
def container_pids():
cg = container_cgroup()
pids = []
for root, _dirs, files in os.walk(cg):
if "cgroup.procs" in files:
pids += [int(x) for x in open(os.path.join(root, "cgroup.procs")).read().split()]
if not pids:
die(f"no process in the container's cgroup at {cg}")
return sorted(set(pids))
def container_cgroup():
rel = podman("inspect", "--format", "{{.State.CgroupPath}}", CNAME)
if not rel:
die("podman reported an empty cgroup path")
return "/sys/fs/cgroup/" + rel.lstrip("/")
def pin_container(cpus):
for pid in container_pids():
r = subprocess.run(["podman", "unshare", "taskset", "-acp", cpus, str(pid)],
capture_output=True, text=True)
if r.returncode != 0:
die(f"pinning {pid} to {cpus}: {r.stderr.strip()[:200]}")
def die(msg):
print(f"FATAL: {msg}", file=sys.stderr)
raise SystemExit(1)
def load1():
return float(open("/proc/loadavg").read().split()[0])
def token():
return open(f"{ROOT}/token").read().strip()
def url():
return f"http://bencher:{token()}@127.0.0.1:{PORT}/bench/conc.git"
def cgroup_dir(pid):
rel = None
for line in open(f"/proc/{pid}/cgroup"):
if line.startswith("0::"):
rel = line[3:].strip().lstrip("/")
if rel is None:
die(f"pid {pid} has no cgroup v2 line")
return f"/sys/fs/cgroup/{rel}"
def cpu_usage_usec(cg):
for line in open(f"{cg}/cpu.stat"):
if line.startswith("usage_usec "):
return int(line.split()[1])
die(f"{cg}/cpu.stat has no usage_usec")
def cpu_stat(cg):
out = {}
for line in open(f"{cg}/cpu.stat"):
k, _, v = line.partition(" ")
out[k] = int(v)
return out
def threads_affinity(pid):
if CONTAINER:
seen = {}
for p in container_pids():
for k, v in _tids(p).items():
seen[k] = seen.get(k, 0) + v
return seen
return _tids(pid)
def _tids(pid):
seen = {}
for tid in os.listdir(f"/proc/{pid}/task"):
try:
txt = open(f"/proc/{pid}/task/{tid}/status").read()
except OSError:
continue
m = re.search(r"^Cpus_allowed_list:\s*(\S+)", txt, re.M)
if m:
seen.setdefault(m.group(1), 0)
seen[m.group(1)] += 1
return seen
def start_server(maxt, binary):
log = open(f"{ROOT}/server.log", "w")
cmd = [
"systemd-run", "--user", "--scope", f"--unit={SCOPE}", "--quiet",
] + ([] if "quota" in RED else ["-p", f"CPUQuota={QUOTA_CPUS*100}%"]) + [
binary, "--data-root", DATA, "serve",
"--http-bind", f"127.0.0.1:{PORT}",
"--http-allow-cleartext-credentials", "--no-grpc",
"--ssh-bind", "127.0.0.1:12299",
"--max-concurrent-transfers", str(maxt),
"--max-transfers-per-principal", str(maxt),
]
p = subprocess.Popen(cmd, stdout=log, stderr=log)
for _ in range(120):
r = subprocess.run(
["curl", "-fsS", "-o", "/dev/null", "--user", f"bencher:{token()}",
f"http://127.0.0.1:{PORT}/bench/conc.git/info/refs?service=git-upload-pack"],
capture_output=True)
if r.returncode == 0:
break
time.sleep(0.5)
else:
die("the server never answered info/refs; see " + f"{ROOT}/server.log")
return p
def pin(pid, cpus):
r = subprocess.run(["taskset", "-acp", cpus, str(pid)], capture_output=True, text=True)
if r.returncode != 0:
die(f"taskset -acp {cpus} {pid}: {r.stderr.strip()}")
def clone_one(i, base):
dest = f"{base}/client-{i}"
t0 = time.perf_counter()
env = dict(os.environ, GIT_TERMINAL_PROMPT="0")
# ── REAL RED SWITCHES ─────────────────────────────────────────────────
# `clients`: one client of every batch asks for a repository that is not
# there. It fails in tens of milliseconds and RAISES the batch's ops/s.
# `objects`: one client clones --depth 1 — a valid repository, a different
# object set, and a wonderfully small pack.
u = url()
args = ["git", "clone", "-q", "--bare"]
if i == 0 and "clients" in RED:
u = u.replace("/conc.git", "/this-repository-does-not-exist.git")
if i == 0 and "objects" in RED:
args += ["--depth", "1"]
r = subprocess.run(args + [u, dest],
capture_output=True, text=True, env=env)
ms = (time.perf_counter() - t0) * 1000.0
return i, r.returncode, r.stderr.strip()[-300:], dest, ms
def corpus_truth():
d = f"{DATA}/repos/bench/conc.git"
n = subprocess.run(["git", "--git-dir", d, "rev-list", "--objects", "--all"],
capture_output=True, text=True, check=True).stdout
tip = subprocess.run(["git", "--git-dir", d, "rev-parse", "refs/heads/main"],
capture_output=True, text=True, check=True).stdout.strip()
return len(set(l.split()[0] for l in n.splitlines() if l.strip())), tip
def verify(dest, want_objects, want_tip):
got = subprocess.run(["git", "--git-dir", dest, "rev-list", "--objects", "--all"],
capture_output=True, text=True)
if got.returncode != 0:
return f"{dest} could not be read: {got.stderr.strip()[:200]}"
n = len(set(l.split()[0] for l in got.stdout.splitlines() if l.strip()))
if n != want_objects:
return f"{dest} landed {n} objects, the corpus holds {want_objects}"
tip = subprocess.run(["git", "--git-dir", dest, "rev-parse", "refs/heads/main"],
capture_output=True, text=True).stdout.strip()
if tip != want_tip:
return f"{dest} landed main at {tip!r}, the corpus is at {want_tip!r}"
return None
def batch(pid, cg, arm, cpus, n, want_objects, want_tip):
base = f"{ROOT}/work/{arm}"
shutil.rmtree(base, ignore_errors=True)
os.makedirs(base)
# GUARD quota
cpumax = open(f"{cg}/cpu.max").read().split()
if cpumax[0] == "max" or int(cpumax[0]) / int(cpumax[1]) != QUOTA_CPUS:
die(f"{arm}: cpu.max is {' '.join(cpumax)} — the quota this arm claims was never applied")
# GUARD affinity, before
before = threads_affinity(pid)
if list(before) != [cpus]:
die(f"{arm}: the server's threads are on {before}, not {cpus!r}")
l_before = load1()
c0 = cpu_usage_usec(cg)
s0 = cpu_stat(cg)
t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=n) as pool:
results = list(pool.map(lambda i: clone_one(i, base), range(n)))
wall = time.perf_counter() - t0
c1 = cpu_usage_usec(cg)
s1 = cpu_stat(cg)
l_after = load1()
# GUARD clients
bad = [f"client {i} exited {rc}: {err}" for i, rc, err, _, _ in results if rc != 0]
if bad:
die(f"{arm}: {len(bad)} of {n} clients did not complete: {bad[:2]}")
# GUARD objects
for _, _, _, dest, _ in results:
why = verify(dest, want_objects, want_tip)
if why:
die(f"{arm}: {why}")
# GUARD cpu
cpu_s = (c1 - c0) / 1e6
if "cpu" in RED:
cpu_s = 0.0
if cpu_s <= 0.0:
die(f"{arm}: the cgroup burned {cpu_s} CPU-seconds over {n} clones")
# GUARD affinity, after (a thread spawned mid-batch could be anywhere)
after = threads_affinity(pid)
if list(after) != [cpus]:
die(f"{arm}: threads spawned during the batch sit on {after}, not {cpus!r}")
lat = sorted(ms for *_, ms in results)
shutil.rmtree(base, ignore_errors=True)
return {
"arm": arm, "cpus": cpus, "clients": n,
"wall_s": round(wall, 3),
"ops_s": round(n / wall, 4),
"cpu_s": round(cpu_s, 3),
"cpu_s_per_op": round(cpu_s / n, 4),
"cpu_used_frac_of_quota": round(cpu_s / (wall * QUOTA_CPUS), 4),
"nr_throttled": s1.get("nr_throttled", 0) - s0.get("nr_throttled", 0),
"throttled_usec": s1.get("throttled_usec", 0) - s0.get("throttled_usec", 0),
"nr_periods": s1.get("nr_periods", 0) - s0.get("nr_periods", 0),
"p50_ms": round(lat[len(lat)//2], 1),
"p99_ms": round(lat[-1], 1),
"threads": sum(after.values()),
"load1_before": l_before, "load1_after": l_after,
}
def _tail(rows):
out = os.environ.get("TP_OUT", f"{ROOT}/results.jsonl")
with open(out, "a") as f:
for row in rows:
f.write(json.dumps(row) + "\n")
print("\n arm n ops/s med ops/s best cpu-s/op quota used thr% p99 ms")
for arm, *_ in ARMS:
a = [x for x in rows if x["arm"] == arm]
if not a:
continue
g = lambda k: sorted(x[k] for x in a)[len(a)//2]
best = max(x["ops_s"] for x in a)
thr = 100.0 * sum(x["nr_throttled"] for x in a) / max(1, sum(x["nr_periods"] for x in a))
print(f" {arm:<18} {len(a):>2} {g('ops_s'):>9.3f} {best:>10.3f} {g('cpu_s_per_op'):>8.4f} "
f"{g('cpu_used_frac_of_quota'):>10.3f} {thr:>4.0f} {g('p99_ms'):>6.0f}")
def main():
want_objects, want_tip = corpus_truth()
for w, b in BINS.items():
if not os.access(b, os.X_OK):
die(f"the {w!r} binary is not there: {b}")
v = subprocess.run([b, "--version"], capture_output=True, text=True).stdout.strip()
print(f"{w}: {v}", flush=True)
print(f"corpus: {want_objects} objects, tip {want_tip[:12]}, "
f"{len([f for f in os.listdir(DATA+'/repos/bench/conc.git/objects/pack') if f.endswith('.pack')])} packs",
flush=True)
subprocess.run(["systemctl", "--user", "stop", f"{SCOPE}.scope"], capture_output=True)
def boot(maxt, which):
if CONTAINER:
start_container(maxt, BINS[which])
cg = container_cgroup()
pid = container_pids()[0]
q = open(f"{cg}/cpu.max").read().split()
if q[0] == "max":
die("the container has no cpu quota — --cpus was accepted and nothing was capped")
return pid, cg
subprocess.run(["systemctl", "--user", "stop", f"{SCOPE}.scope"], capture_output=True)
time.sleep(0.5)
start_server(maxt, BINS[which])
out = subprocess.run(["pgrep", "-f", f"{BINS[which]} --data-root {DATA} serve"],
capture_output=True, text=True).stdout.split()
if len(out) != 1:
die(f"expected exactly one serving gunnar, found {out}")
pid = int(out[0])
cg = cgroup_dir(pid)
if SCOPE not in cg:
die(f"the server sits in {cg}, which is not the {SCOPE} scope — no quota was applied")
return pid, cg
pid, cg = boot(ARMS[0][3], ARMS[0][1])
print(f"server pid {pid}, cgroup {cg}", flush=True)
rows = []
try:
for r in range(ROUNDS):
arms = ARMS if r % 2 == 0 else list(reversed(ARMS))
cur = None
for arm, which, cpus, maxt in arms:
if (which, maxt) != cur:
pid, cg = boot(maxt, which)
cur = (which, maxt)
# `affinity`: pin somewhere other than what the arm claims.
want = "0-3" if "affinity" in RED else cpus
(pin_container if CONTAINER else (lambda c: pin(pid, c)))(want)
base = f"{ROOT}/work/warm"
shutil.rmtree(base, ignore_errors=True)
os.makedirs(base)
clone_one(0, base)
shutil.rmtree(base, ignore_errors=True)
row = batch(pid, cg, arm, cpus, CLIENTS, want_objects, want_tip)
row["round"] = r
row["max_transfers"] = maxt
row["build"] = which
rows.append(row)
print(json.dumps(row), flush=True)
finally:
subprocess.run(["systemctl", "--user", "stop", f"{SCOPE}.scope"], capture_output=True)
podman("rm", "-f", CNAME, check=False)
_tail(rows)
main()