diff --git a/docs/free-threading.md b/docs/free-threading.md index f0139325..29270ae8 100644 --- a/docs/free-threading.md +++ b/docs/free-threading.md @@ -93,9 +93,32 @@ hiredis is excluded because it does not declare free-threaded support (no `Py_mod_gil` slot); redis-py transparently falls back to its pure-Python parser. On a GIL build nothing changes — hiredis remains the default parser. +The exclusion also means the lane never sees what a default install gets. +`pip install cachekit` pulls in hiredis, and on 3.14t importing cachekit with +hiredis present still re-enables the GIL. `cachekit.hiredis_compat` has to +import `redis.connection` to clear its `HIREDIS_AVAILABLE` flag, and that +import loads hiredis. The flag is inert by then anyway: redis-py binds its +default parser at import time, so the hiredis parser stays the default. +`tests/unit/test_free_threading.py::test_default_install_keeps_gil_disabled_after_redis_backend` +checks this in a fresh interpreter. It is marked `xfail(strict=True)`, so it +turns red the day the GIL stays off and the mark must go. +`test_gil_stays_disabled_with_hiredis_blocked` is its control: the same check +with hiredis kept out must pass, so hiredis is the only thing the xfail waits +on. Both skip where hiredis is not installed, as in this lane, so run them by +hand in a 3.14t environment that has hiredis: + +```bash +uv sync --python 3.14t --no-default-groups --group test +uv run --no-sync pytest tests/unit/test_free_threading.py +``` + +In that environment `test_gil_stays_disabled_after_importing_cachekit` and the +session-teardown GIL check also report the re-enabled GIL, because the pytest +process itself has imported hiredis. + ## Measured performance -A post-merge benchmark run (commit `bda770bce822d9a6eff98e555c5f6fd92e509a9c`, CPython 3.14.3 free-threaded build, eight logical CPUs, pinned with `taskset -c 0-7` on a Ryzen 9 5950X) compared no-GIL and GIL cache throughput: +A post-merge benchmark run (commit `bda770bce822d9a6eff98e555c5f6fd92e509a9c`, CPython 3.14.3 free-threaded build, eight logical CPUs, pinned with `taskset -c 0-7` on a Ryzen 9 5950X) compared no-GIL and GIL serializer throughput: `tests/performance/gil_benchmark.py`, which times `StandardSerializer.serialize` alone, not decorated cache calls: | threads | no-GIL median s (min–max) | GIL median s (min–max) | GIL / no-GIL | | --: | --: | --: | --: | @@ -104,7 +127,7 @@ A post-merge benchmark run (commit `bda770bce822d9a6eff98e555c5f6fd92e509a9c`, C | 4 | 1.3402 (1.1175–1.9018) | 3.5293 (3.4629–4.8231) | 2.63x | | 8 | 1.1550 (0.8962–1.4837) | 3.6749 (3.5853–4.7236) | 3.18x | -**Measurement conditions:** Five isolated repetitions each; 16,000-operation workload; harness built-in warmup; GIL state asserted via `sys._is_gil_enabled()` at runtime. See [verification comment](https://github.com/cachekit-io/cachekit-py/pull/188#issuecomment-5557418229) for full details. +**Measurement conditions:** Five isolated repetitions each; 16,000 `serialize` calls (2,000 per thread at eight threads); harness built-in warmup; GIL state asserted via `sys._is_gil_enabled()` at runtime. See [verification comment](https://github.com/cachekit-io/cachekit-py/pull/188#issuecomment-5557418229) for full details. **Key findings:** - **Threaded throughput confirmed.** no-GIL reaches 2.57x one→four-thread scaling (64.2% efficiency) and is 2.63x faster than the GIL arm at four threads. @@ -112,6 +135,61 @@ A post-merge benchmark run (commit `bda770bce822d9a6eff98e555c5f6fd92e509a9c`, C **Cross-library comparison:** The benchmark measures cachekit operations only. Cross-library throughput (orjson, numpy, pandas, pyarrow) was not run. It waits on orjson, which does not publish free-threaded (`cp314t`) wheels as of 2026-09-29; cross-stack performance will be measured once it does. +### Decorated-call scaling + +`tests/performance/ft_scaling_bench.py` measures whole decorated calls, so it +sees the shared state a real call touches: the L1 lock, per-function stats, +the metrics path, the Rust `ByteStorage` and the backend client. It is a +manual bench and no CI job runs it. It runs three cells at 1, 4 and 16 +threads: an L1 hit, an L2 hit over an in-process backend, and an L2 hit +through the shipped CachekitIO backend against a loopback TLS fake +(`tests/performance/loopback_saas.py`). The CachekitIO cell measures +client-side contention only, never service latency. The fake serves each +connection on one worker thread, so the summary flags a cell `SERVER-BOUND` +when a worker was more than 80% busy: there the fake, not the client, sets +the ceiling. + +Each process runs one arm. Every repetition runs every arm, in a seeded +shuffled order, so no arm always follows the same neighbour: + +| arm | interpreter | what it shows | +| --- | --- | --- | +| `ft-nogil` | free-threaded build, `PYTHON_GIL=0` | the no-GIL result | +| `ft-nogil-aa` | the same again | the A/A noise floor of this session | +| `ft-gil` | the same binary, `PYTHON_GIL=1` | the clean GIL vs no-GIL comparison | +| `ft-default` | the same binary, `PYTHON_GIL` unset | the GIL state the imports leave behind | +| `gil-build` | a default GIL build | context only: a different binary | + +```bash +uv sync --python 3.14t --no-default-groups --group test # FT env (UV_PROJECT_ENVIRONMENT=... to keep it apart) +uv sync --python 3.14 --no-default-groups --group test # GIL env +python tests/performance/ft_scaling_bench.py --ft-python /bin/python \ + --gil-python /bin/python --reps 7 --cpus 0-15 --server-cpus 16-23 --out ft-scaling.jsonl +``` + +The primary metric is scaling: calls/s at N threads over calls/s at one +thread, within one process. A cell is `TAINTED` when a timed call was not a +hit or a backend call raised, which usually means the backend failed and the +call fell back to computing; a tainted cell measures that fallback, so it is +left out of every statistic. The summary gives the median and min–max over +clean repetitions. Each repetition is a session block that holds every arm, +so each arm's difference from `ft-nogil` is paired by repetition, with a 95% +bootstrap CI that resamples whole repetitions; with fewer than five clean +pairs it prints "insufficient clean reps" instead of a CI. A difference counts +only if its CI excludes zero and it is larger than the `ft-nogil-aa` floor. +The driver stops if an arm ran under the wrong GIL state, if `--ft-python` is +not a free-threaded build, or if `--gil-python` is one. It refuses an existing +`--out` file so two sessions are never pooled, and the summary refuses a file +that repeats an arm's repetition. The summary also drops a process whose GIL +state changed while the cells ran. + +Today the CachekitIO cell is tainted under no-GIL. Threads that share one +backend share one HTTP/2 connection, and httpcore's sync HTTP/2 send path +allocates stream ids and HPACK-encodes headers outside its locks +([encode/httpcore#1118](https://github.com/encode/httpcore/pull/1118)). The +peer then rejects the connection and every request in flight on it fails. The +race is rare under the GIL and frequent without it. + ## Deferred: declared support + free-threaded wheels Publishing `cp314t` wheels and declaring official free-threaded support is diff --git a/tests/performance/README.md b/tests/performance/README.md index 4d70267a..1d99f510 100644 --- a/tests/performance/README.md +++ b/tests/performance/README.md @@ -15,6 +15,8 @@ elsewhere via the `--benchmark-skip` default in `pyproject.toml`. fingerprint, an environment pre-flight check (throttling / load), and a timer self-calibration (`test_measurement_calibration.py`) — run before trusting any number. `gil_benchmark.py` reports serializer thread-scaling under the current interpreter. +`ft_scaling_bench.py` reports decorated-call thread-scaling, GIL vs no-GIL, in interleaved +arms (manual; see [docs/free-threading.md](../../docs/free-threading.md#decorated-call-scaling)). Run the whole battery with `make perf`; gate regressions with `make perf-compare`. --- diff --git a/tests/performance/ft_scaling_bench.py b/tests/performance/ft_scaling_bench.py new file mode 100644 index 00000000..ccebc2a9 --- /dev/null +++ b/tests/performance/ft_scaling_bench.py @@ -0,0 +1,457 @@ +"""Free-threaded thread scaling of decorated cachekit calls, GIL vs no-GIL, interleaved arms. + +``gil_benchmark.py`` times ``StandardSerializer.serialize`` alone. This bench times whole decorated +calls, so it sees the shared state a real call touches: the L1 lock, the per-function stats, the +metrics path, the Rust ByteStorage and the backend client. + +Arms. Each process runs one arm once; every rep runs every arm, in a seeded shuffled order. + ft-nogil the free-threaded binary with PYTHON_GIL=0 + ft-nogil-aa the same arm again: an A/A pair, whose difference is this session's noise floor + ft-gil the same binary with PYTHON_GIL=1: the clean GIL vs no-GIL comparison + ft-default the same binary with PYTHON_GIL unset: the GIL state the imports leave behind + (with hiredis installed it is back on; without hiredis it matches ft-nogil) + gil-build a default GIL build (--gil-python): context only, since it is a different binary + +Cells, each at 1, 4 and 16 threads: + l1hit @cache over an in-process dict backend, 64 hot keys, every call an L1 hit + l2stub the same with L1 off: every call is an L2 hit through cachekit's L2 path, no transport + cachekitio L1 off, the shipped CachekitIO backend and client against a loopback TLS fake + (loopback_saas.py). It measures client-side contention only, never SaaS latency. + The fake reports its busiest worker's CPU share per cell; above 80% the cell is + flagged SERVER-BOUND, because then the fake, not the client, sets the ceiling. + +Every thread starts at a barrier, warms up, then counts calls for a fixed time. Counts are summed +after the threads join, so no lock sits in the timed loop. The primary metric is scaling: calls/s at +N threads over calls/s at 1 thread, within one process. It drops any process whose GIL state changed +while the cells ran, and flags a cell TAINTED, and leaves it out of every statistic, when a timed call +was not a hit or a backend call raised: such a cell measures fallback throughput, not cache hits. The +summary prints the median and min-max over clean reps. Each rep is a session block that holds every +arm, so each arm's difference from ft-nogil is paired by rep, and its 95% CI resamples whole reps; +with fewer than MIN_CLEAN_PAIRS clean pairs it prints "insufficient clean reps" instead. A difference +counts only if its CI excludes 0 and it is larger than the ft-nogil-aa difference. + +The free-threaded interpreter should come from an environment without the [data] and [json] extras, +for example ``uv sync --python 3.14t --no-default-groups --group test``: some of their builds +re-enable the GIL on import. Needs ``openssl`` on PATH for the loopback certificate. + +Run: + python tests/performance/ft_scaling_bench.py --ft-python /bin/python \\ + --gil-python /bin/python --reps 7 --cpus 0-15 --server-cpus 16-23 --out ft-scaling.jsonl + python tests/performance/ft_scaling_bench.py --summarise ft-scaling.jsonl +""" + +from __future__ import annotations + +import argparse +import json +import os +import random +import statistics +import subprocess +import sys +import tempfile +import threading +import time +from collections import Counter, defaultdict +from collections.abc import Callable +from pathlib import Path + +HERE = Path(__file__).resolve().parent +THREADS = (1, 4, 16) +SCENARIOS = ("l1hit", "l2stub", "cachekitio") +ARMS = { + "ft-nogil": ("ft", {"PYTHON_GIL": "0"}), + "ft-nogil-aa": ("ft", {"PYTHON_GIL": "0"}), + "ft-gil": ("ft", {"PYTHON_GIL": "1"}), + "ft-default": ("ft", {}), + "gil-build": ("gil", {}), +} +EXPECTED_GIL = {"ft-nogil": False, "ft-nogil-aa": False, "ft-gil": True, "gil-build": True} +CELL_TIMEOUT_S = 600 +BOOTSTRAP_RESAMPLES = 2000 +MIN_CLEAN_PAIRS = 5 # below this a bootstrap CI over reps is not worth printing + + +def _gil_enabled() -> bool: + return bool(getattr(sys, "_is_gil_enabled", lambda: True)()) + + +# --- one arm, one process --------------------------------------------------------------------------- + + +def run_cell(n: int, make_op: Callable[[int], Callable[[], object]], dur: float, warm: float) -> dict[str, float | None]: + """Run make_op(tid)() on n threads for about dur seconds after warm seconds; return aggregate calls/s. + + Each thread's rate is its calls over its own counted interval, so a call that overruns the window + (a backend timeout) lowers the rate instead of being credited to dur. + """ + ops = [make_op(tid) for tid in range(n)] + rates = [0.0] * n + errors: list[BaseException] = [] + barrier = threading.Barrier(n + 1) + window: dict[str, float] = {} + + def loop(tid: int) -> None: + op, clock = ops[tid], time.perf_counter + barrier.wait() + try: + while clock() < window["measure_from"]: + op() + calls, started = 0, clock() + while clock() < window["end"]: + op() + op() + op() + op() + calls += 4 + rates[tid] = calls / (clock() - started) + except BaseException as exc: # noqa: BLE001 -- re-raised after join: a dead thread must not read as a slow one + errors.append(exc) + + threads = [threading.Thread(target=loop, args=(tid,)) for tid in range(n)] + for thread in threads: + thread.start() + now = time.perf_counter() + window["measure_from"] = now + warm + window["end"] = now + warm + dur + barrier.wait() + for thread in threads: + thread.join() + if errors: + raise RuntimeError(f"{len(errors)} of {n} threads raised") from errors[0] + total = sum(rates) + return {"calls_per_s": round(total), "us_per_call_per_thread": round(n / total * 1e6, 3) if total else None} + + +class _DictBackend: + """In-process backend: cachekit's L2 code path with no transport.""" + + def __init__(self) -> None: + self.store: dict[str, bytes] = {} + + def get(self, key: str) -> bytes | None: + return self.store.get(key) + + def set(self, key: str, value: bytes, ttl: int | None = None) -> None: + self.store[key] = value + + def delete(self, key: str) -> bool: + return self.store.pop(key, None) is not None + + def exists(self, key: str) -> bool: + return key in self.store + + def health_check(self) -> tuple[bool, dict[str, str]]: + return True, {"backend_type": "dict"} + + +def _payload(i: int) -> dict[str, object]: + return {"id": i, "pad": "y" * 200, "tags": ["a", "b"], "n": 1.5} + + +def _decorated(scenario: str, port: int, computed: list[int], errors: list[str]): + """The scenario's decorated function, primed so every later call is a hit. + + Each miss appends to computed; each CachekitIO get/set that raises appends to errors, which also + catches a failure that a retry then hides. + """ + from cachekit import cache + from cachekit.config.nested import L1CacheConfig + + def payload(i: int) -> dict[str, object]: + computed.append(i) + return _payload(i) + + if scenario == "l1hit": + fn = cache(backend=_DictBackend(), ttl=3600, namespace="ftb_l1")(payload) + elif scenario == "l2stub": + fn = cache(backend=_DictBackend(), ttl=3600, namespace="ftb_l2", l1=L1CacheConfig(enabled=False))(payload) + elif scenario == "cachekitio": + from cachekit.backends.cachekitio import CachekitIOBackend + from cachekit.backends.cachekitio import config as cachekitio_config + + # The fake listens on 127.0.0.1, which the SSRF guard rejects; lift the guard in this process only. + cachekitio_config.is_private_ip = lambda hostname: False + + class CountingBackend(CachekitIOBackend): + pass + + def counting(method): + def wrapper(self, *args, **kwargs): + try: + return method(self, *args, **kwargs) + except Exception as exc: + errors.append(type(exc).__name__) + raise + + return wrapper + + for name in ("get", "get_with_freshness", "set"): # what the sync decorated path calls + setattr(CountingBackend, name, counting(getattr(CachekitIOBackend, name))) + + backend = CountingBackend(api_url=f"https://127.0.0.1:{port}", api_key="ck_test_bench") + fn = cache(backend=backend, ttl=3600, namespace="ftb_io", l1=L1CacheConfig(enabled=False))(payload) + else: + raise ValueError(f"unknown scenario {scenario!r}") + for i in range(64): + fn(i) # miss -> compute -> store; every timed call after this is a hit + return fn + + +def _fake_cpu(port: int) -> list[float]: + """CPU seconds used so far by each worker of the loopback fake (on its own HTTP/1.1 connection).""" + import httpx + + return httpx.get(f"https://127.0.0.1:{port}/__stats").raise_for_status().json()["worker_cpu_s"] + + +def cell_main(port: int, dur: float, warm: float) -> None: + import sysconfig + + sys.path.insert(0, str(HERE.parents[1])) + from tests.performance.measurement_env import fingerprint_hash, system_fingerprint + + gil_at_launch = _gil_enabled() + load_start = os.getloadavg() + computed: dict[str, list[int]] = {scenario: [] for scenario in SCENARIOS} + errors: dict[str, list[str]] = {scenario: [] for scenario in SCENARIOS} + fns = {scenario: _decorated(scenario, port, computed[scenario], errors[scenario]) for scenario in SCENARIOS} + gil_before = _gil_enabled() + + def make_op(fn): + def make(tid: int): + box = [tid] + + def op(): + box[0] = (box[0] + 1) & 63 + return fn(box[0]) + + return op + + return make + + results = [] + for scenario in SCENARIOS: + for n in THREADS: + primed, errored = len(computed[scenario]), len(errors[scenario]) + fake_before, wall = (_fake_cpu(port), time.perf_counter()) if scenario == "cachekitio" else (None, 0.0) + cell = run_cell(n, make_op(fns[scenario]), dur, warm) + # A timed call that was not a hit, or a backend call that raised, taints the cell. + cell["misses"] = len(computed[scenario]) - primed + cell["backend_errors"] = len(errors[scenario]) - errored + if fake_before is not None: + wall = time.perf_counter() - wall + cell["fake_busiest_worker"] = round( + max(b - a for a, b in zip(fake_before, _fake_cpu(port), strict=True)) / wall, 3 + ) + results.append({"scenario": scenario, "threads": n, **cell}) + print( + json.dumps( + { + "python": sys.version.split()[0], + "free_threaded_build": bool(sysconfig.get_config_var("Py_GIL_DISABLED")), + "gil_at_launch": gil_at_launch, + "gil_before_cells": gil_before, + "gil_after_cells": _gil_enabled(), + "hiredis_loaded": sys.modules.get("hiredis") is not None, + "cpu_count": os.cpu_count(), + "cpus_allowed": len(os.sched_getaffinity(0)), + "load_start": load_start, + "load_end": os.getloadavg(), + "fingerprint": fingerprint_hash(system_fingerprint()), + "results": results, + } + ) + ) + + +# --- driver ----------------------------------------------------------------------------------------- + + +def _pinned(cpus: str | None, cmd: list[str]) -> list[str]: + return ["taskset", "-c", cpus, *cmd] if cpus else cmd + + +def _start_fake(python: str, tmp: Path, cpus: str | None, workers: int) -> tuple[subprocess.Popen[str], int]: + cert, key = tmp / "cert.pem", tmp / "key.pem" + subprocess.run( # noqa: S603 (trusted: literal openssl argv) + ["openssl", "req", "-x509", "-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:prime256v1", "-nodes", "-days", "1"] + + ["-subj", "/CN=127.0.0.1", "-addext", "subjectAltName=IP:127.0.0.1", "-keyout", str(key), "-out", str(cert)], + check=True, + capture_output=True, + ) + env = {**os.environ, "PYTHON_GIL": "0"} # the fake runs its workers in parallel on a free-threaded build + cmd = _pinned(cpus, [python, str(HERE / "loopback_saas.py"), "0", str(cert), str(key), str(workers)]) + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, env=env) # noqa: S603 (trusted: operator-passed interpreter) + assert proc.stdout is not None + line = proc.stdout.readline().split() + if line[:1] != ["ready"]: + proc.kill() + raise RuntimeError(f"loopback fake failed to start (exit {proc.wait()})") + return proc, int(line[1]) + + +def _check_row(arm: str, row: dict) -> None: + """Fail loudly when an arm did not run under the interpreter state it claims.""" + if ARMS[arm][0] == "ft" and not row["free_threaded_build"]: + raise RuntimeError(f"--ft-python is not a free-threaded build (arm {arm})") + # A GIL that an import switched back on does not make a free-threaded binary a GIL build. + if ARMS[arm][0] == "gil" and row["free_threaded_build"]: + raise RuntimeError(f"--gil-python is a free-threaded build; arm {arm} needs a default GIL build") + if arm in EXPECTED_GIL and row["gil_before_cells"] != EXPECTED_GIL[arm]: + raise RuntimeError(f"arm {arm} ran with gil_enabled={row['gil_before_cells']}, expected {EXPECTED_GIL[arm]}") + + +def drive(args: argparse.Namespace) -> None: + pythons = {"ft": args.ft_python, "gil": args.gil_python} + wanted = args.arms.split(",") if args.arms else list(ARMS) + if unknown := set(wanted) - set(ARMS): + raise SystemExit(f"unknown arms: {sorted(unknown)}") + arms = [arm for arm in wanted if pythons[ARMS[arm][0]]] + if not arms: + raise SystemExit("no arm has an interpreter: pass --ft-python (and --gil-python for gil-build)") + out = Path(args.out) + if out.exists(): + raise SystemExit(f"{out} exists; give each session its own --out so sessions are never pooled") + rng = random.Random(0) + with tempfile.TemporaryDirectory() as tmp: + fake, port = _start_fake(args.ft_python, Path(tmp), args.server_cpus, args.server_workers) + base_env = {k: v for k, v in os.environ.items() if k != "PYTHON_GIL"} + base_env.update(CACHEKIT_ALLOW_CUSTOM_HOST="true", SSL_CERT_FILE=str(Path(tmp) / "cert.pem"), PYTHONWARNINGS="ignore") + try: + for rep in range(args.reps): + order = rng.sample(arms, len(arms)) # shuffled, so no arm always follows the same neighbour + for arm in order: + build, arm_env = ARMS[arm] + cmd = [pythons[build], __file__, "--cell", str(port), "--dur", str(args.dur), "--warm", str(args.warm)] + proc = subprocess.run( # noqa: S603 (trusted: interpreter paths the operator passed) + _pinned(args.cpus, cmd), + env={**base_env, **arm_env}, + capture_output=True, + text=True, + check=False, + timeout=CELL_TIMEOUT_S, + ) + if proc.returncode != 0: + raise RuntimeError(f"arm {arm} rep {rep} failed:\n{proc.stderr}") + row = {"arm": arm, "rep": rep, **json.loads(proc.stdout.strip().splitlines()[-1])} + _check_row(arm, row) + with out.open("a") as fh: + fh.write(json.dumps(row) + "\n") + print(f"rep {rep} {arm} done", file=sys.stderr) + finally: + fake.terminate() + fake.wait() + summarise(out) + + +def _spread(values: list[float], fmt: str) -> str: + return f"{statistics.median(values):{fmt}} ({min(values):{fmt}}-{max(values):{fmt}})" + + +def summarise(path: Path) -> None: + rows = [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + # Reps restart at 0 every session, so a repeated (arm, rep) means pooled sessions, which would + # silently overwrite rows and pair one session's arm with another's ft-nogil. + if pooled := sorted(k for k, count in Counter((r["arm"], r["rep"]) for r in rows).items() if count > 1): + raise SystemExit(f"{path}: arm/rep {pooled[0]} appears more than once; the file pools more than one session") + kept = [r for r in rows if r["gil_before_cells"] == r["gil_after_cells"]] + print(f"{len(rows)} processes, {len(rows) - len(kept)} dropped (GIL state changed while the cells ran)") + loads = [r["load_start"][0] for r in kept] + [r["load_end"][0] for r in kept] + if kept: + print(f"1-min load average {min(loads):.1f}-{max(loads):.1f}; cpus allowed {sorted({r['cpus_allowed'] for r in kept})}") + # Only clean cells (every timed call a hit, no backend raise) enter the statistics; a tainted cell + # measures fallback throughput, not cache-hit scaling. A ratio needs its 1-thread cell clean too. + calls: dict[tuple[str, str, int], dict[int, float]] = defaultdict(dict) + scaling: dict[tuple[str, str, int], dict[int, float]] = defaultdict(dict) + gil: dict[str, set[bool]] = defaultdict(set) + tainted: dict[tuple[str, str, int], list[int]] = defaultdict(lambda: [0, 0, 0]) + for r in kept: + gil[r["arm"]].add(r["gil_before_cells"]) + clean = {} + for c in r["results"]: + key = (r["arm"], c["scenario"], c["threads"]) + if c["misses"] or c["backend_errors"]: + counts = tainted[key] + counts[0] += 1 + counts[1] += c["misses"] + counts[2] += c["backend_errors"] + else: + clean[(c["scenario"], c["threads"])] = c["calls_per_s"] + for (scenario, n), value in clean.items(): + calls[(r["arm"], scenario, n)][r["rep"]] = value + if clean.get((scenario, 1)): # a 1-thread cell with no completed call has no ratio + scaling[(r["arm"], scenario, n)][r["rep"]] = value / clean[(scenario, 1)] + arms = [arm for arm in ARMS if arm in gil] + print("arm GIL state during the cells: " + ", ".join(f"{arm}={sorted(gil[arm])}" for arm in arms)) + for (arm, scenario, n), (reps, misses, errors) in sorted(tainted.items()): + print( + f"TAINTED {arm} {scenario} {n}t: excluded from {reps} reps " + f"({misses} timed calls were not hits, {errors} backend calls raised)" + ) + for r in kept: + for c in r["results"]: + if (c.get("fake_busiest_worker") or 0) > 0.8: + print( + f"SERVER-BOUND {r['arm']} rep {r['rep']} {c['scenario']} {c['threads']}t: a fake worker was " + f"{c['fake_busiest_worker']:.0%} busy, so the fake may set this cell's ceiling" + ) + for title, table, fmt in (("calls/s", calls, ",.0f"), ("scaling vs 1 thread", scaling, ".2f")): + print(f"\n{title}: median (min-max) over clean reps, [n clean]") + print(f"{'scenario':<11}{'thr':>4} " + "".join(f"{arm:>32}" for arm in arms)) + for scenario in SCENARIOS: + for n in THREADS: + values = [list(table[(arm, scenario, n)].values()) for arm in arms] + cells = [f"{_spread(v, fmt)} [{len(v)}]" if v else "- [0]" for v in values] + print(f"{scenario:<11}{n:>4} " + "".join(f"{cell:>32}" for cell in cells)) + if "ft-nogil" not in arms: + return + print( + f"\nscaling, arm minus ft-nogil: median of per-rep paired differences [95% CI, {BOOTSTRAP_RESAMPLES} " + "resamples of whole reps]" + ) + print("each rep is a session block holding every arm; only reps where both cells are clean pair up") + print("the ft-nogil-aa row is the A/A floor: a difference counts only if its CI excludes 0 and it exceeds the floor") + rng = random.Random(0) + lo, hi = int(BOOTSTRAP_RESAMPLES * 0.025), int(BOOTSTRAP_RESAMPLES * 0.975) - 1 + for arm in [a for a in arms if a != "ft-nogil"]: + for scenario in SCENARIOS: + parts = [] + for n in THREADS[1:]: + ref, other = scaling[("ft-nogil", scenario, n)], scaling[(arm, scenario, n)] + paired = [other[rep] - ref[rep] for rep in sorted(ref.keys() & other.keys())] + if len(paired) < MIN_CLEAN_PAIRS: + parts.append(f"{n}t insufficient clean reps ({len(paired)} < {MIN_CLEAN_PAIRS})") + continue + boot = sorted(statistics.median(rng.choices(paired, k=len(paired))) for _ in range(BOOTSTRAP_RESAMPLES)) + parts.append(f"{n}t {statistics.median(paired):+.2f} [{boot[lo]:+.2f}, {boot[hi]:+.2f}] n={len(paired)}") + print(f" {arm:<12}{scenario:<11}" + " ".join(parts)) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--ft-python", help="free-threaded interpreter with cachekit installed") + parser.add_argument("--gil-python", help="default (GIL) build with cachekit installed; adds the gil-build arm") + parser.add_argument("--arms", help=f"comma-separated subset of {','.join(ARMS)} (default: all available)") + parser.add_argument("--reps", type=int, default=5) + parser.add_argument("--out", default="ft-scaling.jsonl", help="JSONL results file; must not exist yet") + parser.add_argument("--dur", type=float, default=1.2, help="seconds counted per cell") + parser.add_argument("--warm", type=float, default=0.25, help="seconds of warm-up per cell") + parser.add_argument("--cpus", help="taskset CPU list for the arm processes, e.g. 0-15") + parser.add_argument("--server-cpus", help="taskset CPU list for the loopback fake, e.g. 16-19") + parser.add_argument("--server-workers", type=int, default=4, help="event-loop threads in the loopback fake") + parser.add_argument("--summarise", metavar="JSONL", help="print the summary of an existing results file") + parser.add_argument("--cell", type=int, metavar="PORT", help=argparse.SUPPRESS) + args = parser.parse_args() + if args.cell is not None: + cell_main(args.cell, args.dur, args.warm) + elif args.summarise: + summarise(Path(args.summarise)) + elif args.ft_python: + drive(args) + else: + parser.error("pass --ft-python (to run) or --summarise") + + +if __name__ == "__main__": + main() diff --git a/tests/performance/loopback_saas.py b/tests/performance/loopback_saas.py new file mode 100644 index 00000000..b2b983b6 --- /dev/null +++ b/tests/performance/loopback_saas.py @@ -0,0 +1,164 @@ +"""Loopback fake of the CachekitIO data plane, for ``ft_scaling_bench.py``'s ``cachekitio`` cell. + +Speaks HTTP/2 and HTTP/1.1 over TLS, chosen by ALPN as at the real edge, so the shipped client +negotiates what it would in production. One in-memory store shared by every worker thread: +GET 200 body | 404, HEAD 200 | 404, PUT 200 ``{"success":true}``, DELETE 200 | 404. +``GET /__stats`` returns each worker's CPU seconds so far, so a caller can tell when the fake is the +bottleneck. Each worker thread runs its own event loop on the shared listening socket; on a +free-threaded interpreter with the GIL off they serve in parallel. One connection is served by one +worker, so a client multiplexing everything over one HTTP/2 connection is bounded by one worker. +Bodies must fit one HTTP/2 frame and the initial flow-control window (well under 16 KiB). + +It measures client-side contention only, never SaaS latency. + +Usage: python loopback_saas.py [workers] +Prints ``ready `` on stdout once it is listening. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import socket +import ssl +import sys +import threading +import time + +import h2.config +import h2.connection +import h2.events +import h2.exceptions +import h11 + +STORE: dict[bytes, bytes] = {} +WORKERS: list[threading.Thread] = [] +_PUT_OK = b'{"success":true}' + + +def handle(method: bytes, path: bytes, body: bytes) -> tuple[int, bytes]: + if path == b"/__stats": + cpu = [time.clock_gettime(time.pthread_getcpuclockid(w.ident)) for w in WORKERS if w.ident is not None] + return 200, json.dumps({"worker_cpu_s": cpu}).encode() + if method == b"GET": + value = STORE.get(path) + return (404, b"") if value is None else (200, value) + if method == b"HEAD": + return (200 if path in STORE else 404), b"" + if method == b"PUT": + STORE[path] = body + return 200, _PUT_OK + if method == b"DELETE": + return (200, _PUT_OK) if STORE.pop(path, None) is not None else (404, b"") + return 405, b"" + + +class _Protocol(asyncio.Protocol): + def connection_made(self, transport: asyncio.BaseTransport) -> None: + assert isinstance(transport, asyncio.Transport) + self.transport = transport + self.h2 = transport.get_extra_info("ssl_object").selected_alpn_protocol() == "h2" + if self.h2: + self.conn = h2.connection.H2Connection(h2.config.H2Configuration(client_side=False, header_encoding=None)) + self.conn.initiate_connection() + self.streams: dict[int, tuple[dict[bytes, bytes], bytearray]] = {} + transport.write(self.conn.data_to_send()) + else: + self.h1 = h11.Connection(h11.SERVER) + self.request: h11.Request | None = None + self.body = bytearray() + + def data_received(self, data: bytes) -> None: + if self.h2: + self._h2_received(data) + else: + self._h1_received(data) + + def _h2_received(self, data: bytes) -> None: + try: + self._h2_events(data) + except h2.exceptions.ProtocolError as exc: + # A client that breaks the protocol (a corrupt HPACK block, a stream id out of order) lands here. + print(f"loopback_saas: HTTP/2 protocol error from the client, connection closed: {exc!r}", file=sys.stderr) + self.transport.write(self.conn.data_to_send()) # the GOAWAY h2 queued, so the client sees why + self.transport.close() + + def _h2_events(self, data: bytes) -> None: + for event in self.conn.receive_data(data): + if isinstance(event, h2.events.RequestReceived): + self.streams[event.stream_id] = (dict(event.headers), bytearray()) # type: ignore[arg-type] + elif isinstance(event, h2.events.DataReceived): + self.streams[event.stream_id][1].extend(event.data) + self.conn.acknowledge_received_data(event.flow_controlled_length, event.stream_id) + elif isinstance(event, h2.events.StreamEnded): + headers, body = self.streams.pop(event.stream_id) + method = headers[b":method"] + status, payload = handle(method, headers[b":path"], bytes(body)) + send_body = bool(payload) and method != b"HEAD" + response = [(b":status", str(status).encode()), (b"content-length", str(len(payload)).encode())] + self.conn.send_headers(event.stream_id, response, end_stream=not send_body) + if send_body: + self.conn.send_data(event.stream_id, payload, end_stream=True) + elif isinstance(event, h2.events.StreamReset): + self.streams.pop(event.stream_id, None) + elif isinstance(event, h2.events.ConnectionTerminated): + self.transport.close() + self.transport.write(self.conn.data_to_send()) + + def _h1_received(self, data: bytes) -> None: + self.h1.receive_data(data) + while True: + event = self.h1.next_event() + if event is h11.NEED_DATA or event is h11.PAUSED: + return + if isinstance(event, h11.Request): + self.request, self.body = event, bytearray() + elif isinstance(event, h11.Data): + self.body.extend(event.data) + elif isinstance(event, h11.EndOfMessage): + assert self.request is not None + status, payload = handle(self.request.method, self.request.target, bytes(self.body)) + out = self.h1.send(h11.Response(status_code=status, headers=[(b"content-length", str(len(payload)))])) + if payload and self.request.method != b"HEAD": + out += self.h1.send(h11.Data(data=payload)) + out += self.h1.send(h11.EndOfMessage()) + self.transport.write(out) + self.h1.start_next_cycle() + elif isinstance(event, h11.ConnectionClosed): + self.transport.close() + return + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("port", type=int, help="port to listen on, 0 for any free port") + parser.add_argument("certfile") + parser.add_argument("keyfile") + parser.add_argument("workers", type=int, nargs="?", default=4, help="event-loop threads (default 4)") + args = parser.parse_args() + if not 0 <= args.port <= 65535: + parser.error(f"port must be 0-65535, got {args.port}") + if args.workers < 1: + parser.error(f"workers must be at least 1, got {args.workers}") + ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + ctx.load_cert_chain(args.certfile, args.keyfile) + ctx.set_alpn_protocols(["h2", "http/1.1"]) + sock = socket.create_server(("127.0.0.1", args.port), backlog=1024) + sock.setblocking(False) + + def serve() -> None: + loop = asyncio.new_event_loop() + loop.run_until_complete(loop.create_server(_Protocol, sock=sock, ssl=ctx)) + loop.run_forever() + + WORKERS.extend(threading.Thread(target=serve, daemon=True) for _ in range(args.workers)) + for thread in WORKERS: + thread.start() + print(f"ready {sock.getsockname()[1]}", flush=True) + for thread in WORKERS: + thread.join() + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_free_threading.py b/tests/unit/test_free_threading.py index e2fe8ba3..30fcc345 100644 --- a/tests/unit/test_free_threading.py +++ b/tests/unit/test_free_threading.py @@ -33,6 +33,94 @@ def test_gil_stays_disabled_after_importing_cachekit(): assert sys._is_gil_enabled() is False +# Run in a fresh interpreter: the importing process's GIL state is what is under test, and the +# pytest process has already imported cachekit. argv[1:] names modules to block. The [data] and +# [json] extras are always blocked: a default install does not have them, and some of their builds +# re-enable the GIL on their own. +_DEFAULT_INSTALL_PROBE = """ +import json, sys +for blocked in ("numpy", "pandas", "pyarrow", "orjson", *sys.argv[1:]): + sys.modules[blocked] = None +import cachekit +from cachekit.backends.redis import RedisBackend +RedisBackend(redis_url="redis://127.0.0.1:6379") +import redis.connection +print(json.dumps({ + "gil_enabled": sys._is_gil_enabled(), + "hiredis_loaded": sys.modules.get("hiredis") is not None, + "redis_parser": redis.connection.DefaultParser.__name__, +})) +""" + + +def _default_install_skip_reason() -> str | None: + if not _FREE_THREADED_BUILD: + return "requires a free-threaded CPython build" + # Package metadata, not find_spec: a fix that blocks hiredis via sys.modules must not skip this. + import importlib.metadata + + try: + importlib.metadata.distribution("hiredis") + except importlib.metadata.PackageNotFoundError: + return "the default install includes hiredis (redis[hiredis]); this environment excludes it" + return None + + +_SKIP_REASON = _default_install_skip_reason() +_needs_default_install = pytest.mark.skipif(_SKIP_REASON is not None, reason=_SKIP_REASON or "") + + +def _probe_default_install(*blocked: str) -> dict[str, object]: + """GIL state in a fresh interpreter after importing cachekit and building a RedisBackend.""" + import json + import os + import subprocess + + env = {k: v for k, v in os.environ.items() if k != "PYTHON_GIL"} # PYTHON_GIL=0 would force a vacuous pass + try: + proc = subprocess.run( # noqa: S603 (trusted: sys.executable + literal code) + [sys.executable, "-W", "ignore", "-c", _DEFAULT_INSTALL_PROBE, *blocked], + capture_output=True, + text=True, + env=env, + timeout=60, + check=False, + ) + if proc.returncode != 0: + pytest.fail(f"probe exited {proc.returncode}:\n{proc.stderr}") + return json.loads(proc.stdout.strip().splitlines()[-1]) + except (subprocess.TimeoutExpired, IndexError, json.JSONDecodeError) as exc: + # pytest.fail, not an assertion: a broken probe must not count as the expected xfail. + pytest.fail(f"probe produced no result: {exc!r}") + + +@_needs_default_install +def test_gil_stays_disabled_with_hiredis_blocked(): + """Control for the xfail below: with hiredis kept out, nothing else in the default install re-enables the GIL.""" + state = _probe_default_install("hiredis") + assert state["gil_enabled"] is False, state + + +@_needs_default_install +@pytest.mark.xfail( + strict=True, + raises=AssertionError, + reason=( + "hiredis_compat's own import of redis.connection loads hiredis, and the HIREDIS_AVAILABLE flag " + "it then clears is inert; passes once hiredis is kept from loading before any redis import" + ), +) +def test_default_install_keeps_gil_disabled_after_redis_backend(): + """With the default dependency set (redis[hiredis] included), cachekit must leave the GIL off. + + The free-threaded CI lane installs without hiredis, so the GIL assertions above never see what a + plain `pip install cachekit` on 3.14t gets. Run this in a 3.14t environment that has hiredis, for + example `uv sync --python 3.14t --no-default-groups --group test`. + """ + state = _probe_default_install() + assert state["gil_enabled"] is False, state + + def test_session_init_hammer_no_partial_publish_observed(): """Many threads racing first-touch session init never observe a partial identity.