diff --git a/delphi/polismath/poller/service.py b/delphi/polismath/poller/service.py index 6285e4272..8f12de2fe 100644 --- a/delphi/polismath/poller/service.py +++ b/delphi/polismath/poller/service.py @@ -64,14 +64,32 @@ def initial_watermark(poll_from_days_ago: float, now_millis: Optional[int] = Non def should_process_zid( - zid: int, allowlist: List[int], blocklist: List[int] + zid: int, + allowlist: List[int], + blocklist: List[int], + shard_index: int = 0, + shard_count: int = 1, ) -> bool: - """Allow/block filter (poller.clj:30-32). + """Allow/block filter (poller.clj:30-32), plus zid-sharding. Clojure ``cond``: if an allowlist is set, only listed zids pass; else if a blocklist is set, listed zids are excluded; else everything passes. The allowlist branch is evaluated first, so it wins over the blocklist. + + Sharding (``shard_count > 1``) selects a slice of zids for this process, so + that N single-worker PROCESSES can share the fleet's work -- threads cannot + (measured: threaded serial fraction 0.9884, i.e. 1.0x from 1 to 16 workers; + independent processes 0.0013, i.e. 15.7x). The default ``shard_count=1`` + is a no-op, so sharding is strictly opt-in. + + The shard test runs FIRST, and that ordering is a correctness property + rather than a style choice: a shard must never process a zid outside its + slice, even one an allowlist names. ``ConversationWorkerPool`` serialises + per zid only WITHIN a process, so two shards both accepting one zid would + run concurrent updates on the same conversation with no mutual exclusion. """ + if shard_count > 1 and zid % shard_count != shard_index: + return False if allowlist: return zid in allowlist if blocklist: @@ -122,6 +140,8 @@ class PollerConfig: allowlist POLL_ALLOWLIST | MATH_ZID_ALLOWLIST (default []) blocklist POLL_BLOCKLIST | MATH_ZID_BLOCKLIST (default []) engine_mode POLISMATH_ENGINE_MODE (default None -> compute's own default) + shard_index POLL_SHARD_INDEX | MATH_SHARD_INDEX (default 0) + shard_count POLL_SHARD_COUNT | MATH_SHARD_COUNT (default 1 = unsharded) worker_pool_size MATH_WORKER_POOL_SIZE (default 4) dump_dir MATH_POLLER_DUMP_DIR (default 'scratch/errorconv') retry_cap MATH_POLLER_RETRY_CAP (default 1) @@ -136,6 +156,19 @@ class PollerConfig: allowlist: List[int] = field(default_factory=list) blocklist: List[int] = field(default_factory=list) engine_mode: Optional[str] = None + # zid-sharding: this process handles zids where zid % shard_count == + # shard_index. shard_count=1 (the default) is unsharded -- every zid. + # One shard = one PROCESS: threads do not parallelise this workload + # (serial fraction 0.9884, 1.0x at 16 workers), independent processes do + # (0.0013, 15.7x at 16). See _validate_shard() for why a bad index must + # be fatal rather than silently empty. + shard_index: int = 0 + shard_count: int = 1 + # NOT lowered to 1 for sharding, deliberately: the pool's threads cannot + # overlap math with math, but the Clojure implementation this replaces does + # parallelise per conversation, and a >1 pool may still overlap DB write I/O + # with math. Treat as a tuning parameter to MEASURE once sharding is + # deployed -- the cost study measured a CPU-bound tick and cannot settle it. worker_pool_size: int = 4 dump_dir: str = "scratch/errorconv" retry_cap: int = 1 @@ -146,6 +179,29 @@ class PollerConfig: # reloaded from math_main + fully rebuilt on next touch (= Clojure restart). conv_cache_cap: int = 0 + def __post_init__(self) -> None: + self._validate_shard() + + def _validate_shard(self) -> None: + """Reject an unusable shard slice loudly, at construction time. + + This is the worst failure mode in the whole design if left silent: an + out-of-range index matches NO zid, so the process starts, polls, logs + happily and computes nothing. The fleet looks up while a slice of + conversations silently goes stale. Crash instead. + """ + if self.shard_count < 1: + raise ValueError( + f"shard_count must be >= 1, got {self.shard_count} " + "(1 = unsharded; set POLL_SHARD_COUNT to the fleet size)" + ) + if not 0 <= self.shard_index < self.shard_count: + raise ValueError( + f"shard_index must be in [0, {self.shard_count}), got " + f"{self.shard_index} -- such a shard would process NO zids " + "while appearing healthy (set POLL_SHARD_INDEX per instance)" + ) + @classmethod def from_env(cls) -> "PollerConfig": return cls( @@ -175,6 +231,12 @@ def from_env(cls) -> "PollerConfig": _env_first("POLL_BLOCKLIST", "MATH_ZID_BLOCKLIST") ), engine_mode=os.environ.get(ENGINE_MODE_ENV_VAR), + shard_index=int( + _env_first("POLL_SHARD_INDEX", "MATH_SHARD_INDEX", default="0") + ), + shard_count=int( + _env_first("POLL_SHARD_COUNT", "MATH_SHARD_COUNT", default="1") + ), worker_pool_size=int(os.environ.get("MATH_WORKER_POOL_SIZE", "4")), dump_dir=os.environ.get("MATH_POLLER_DUMP_DIR", "scratch/errorconv"), retry_cap=int(os.environ.get("MATH_POLLER_RETRY_CAP", "1")), @@ -241,10 +303,15 @@ def start(self) -> None: for t in self._threads: t.start() logger.info( - "MathPollerService started (math_env=%s engine_mode=%s pool=%d)", + "MathPollerService started (math_env=%s engine_mode=%s pool=%d shard=%s)", self.config.math_env, resolve_engine_mode(), self.config.worker_pool_size, + # Spelled out so a misconfigured fleet is visible in the logs rather + # than silently leaving a slice of conversations unprocessed. + f"{self.config.shard_index}/{self.config.shard_count}" + if self.config.shard_count > 1 + else "unsharded", ) def stop(self) -> None: @@ -312,7 +379,13 @@ def _poll_votes_once(self) -> None: rows = self._pg.poll_votes_since(self._vote_wm) logger.info("Polled %d votes since watermark %s", len(rows), self._vote_wm) for zid, batch in _group_by_zid(rows).items(): - if should_process_zid(zid, self.config.allowlist, self.config.blocklist): + if should_process_zid( + zid, + self.config.allowlist, + self.config.blocklist, + self.config.shard_index, + self.config.shard_count, + ): self._unpark(zid) # new batch self-heals a parked zid self._pool.submit(zid, VOTES, batch) self._vote_wm = advance_watermark( @@ -324,7 +397,13 @@ def _poll_moderation_once(self) -> None: rows = self._pg.poll_moderation_since(self._mod_wm) logger.info("Polled %d mod changes since watermark %s", len(rows), self._mod_wm) for zid, batch in _group_by_zid(rows).items(): - if should_process_zid(zid, self.config.allowlist, self.config.blocklist): + if should_process_zid( + zid, + self.config.allowlist, + self.config.blocklist, + self.config.shard_index, + self.config.shard_count, + ): self._unpark(zid) # new batch self-heals a parked zid self._pool.submit(zid, MODERATION, batch) self._mod_wm = advance_watermark( diff --git a/delphi/polismath/replay/shard_bench.py b/delphi/polismath/replay/shard_bench.py new file mode 100644 index 000000000..3f86e3eb9 --- /dev/null +++ b/delphi/polismath/replay/shard_bench.py @@ -0,0 +1,498 @@ +"""Shard-scaling benchmark — does aggregate throughput scale with shard count? + +Closes the gap named in ``HANDOFF_PYTHON_SHARDING.md`` §8: the cost study's +``py-zid-shard`` arm launched N independent processes on N separate cells and +"exercises no ``zid % N`` filter at all", so it measured the CEILING sharding +can reach rather than a sharding implementation. This harness measures the +shipped filter: one fixed workload of ``zid``s is partitioned by the REAL +:func:`polismath.poller.service.should_process_zid`, N processes each take +their slice, and aggregate throughput is compared against the single-shard arm. + +Design notes, each of which is load-bearing for the number this produces: + +* **Cost-balanced workload.** Every zid replays the SAME dataset, so an even + count split is an even work split. ``zid % N`` balances count, not cost + (handoff §4: median 2 in-conv participants, max 23,354), but that skew is a + capacity-planning property — mixing it in here would confound the question + "does the mechanism scale?" with "is this particular zid set balanced?". +* **BLAS pinning.** Unpinned numpy fans a single recompute across every core. + N such shards on one box thrash. Every shard therefore pins its BLAS/OpenMP + threads to 1 (handoff §0); the ``pin=False`` arm exists to MEASURE that + correction rather than assume it. +* **Startup is excluded.** Interpreter start + numpy import + dataset load is + ~2 s and does not shard; each child times only its compute phase, and the + parent releases every child from a barrier so the phases actually overlap. +* **Wall is the slowest shard**, never the sum — shards run concurrently, and + the arm is done when the last one finishes. + +The live benchmark is opt-in (it needs N processes and ~a minute of CPU); this +module's pure decision points are unit-tested with canned numbers. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Sequence + +from polismath.poller.service import should_process_zid + +# Every knob a BLAS/OpenMP backend might read. Pinning only OMP_NUM_THREADS +# leaves OpenBLAS free to fan out on its own, which is exactly the pathology +# being controlled for. +BLAS_ENV_VARS = ( + "OMP_NUM_THREADS", + "OPENBLAS_NUM_THREADS", + "MKL_NUM_THREADS", + "VECLIB_MAXIMUM_THREADS", + "NUMEXPR_NUM_THREADS", +) + +# 24 is divisible by every default shard count, so each arm gets an exactly +# even split and no arm is measuring a remainder. +DEFAULT_ZID_COUNT = 24 +DEFAULT_SHARD_COUNTS = (1, 2, 4, 8) +DEFAULT_DATASET = "biodiversity" +# Quasi-linear bar. Amdahl leaves headroom for real per-process overhead +# (interpreter start is excluded, but page cache, memory bandwidth and OS +# scheduling are not), so "linear" cannot mean 1.00. +DEFAULT_MIN_EFFICIENCY = 0.8 + +_DELPHI_ROOT = Path(__file__).resolve().parents[2] +CHILD_TIMEOUT_SEC = 1800.0 + + +# --------------------------------------------------------------------------- # +# Pure helpers +# --------------------------------------------------------------------------- # +def shard_workload( + zids: Sequence[int], shard_index: int, shard_count: int +) -> list[int]: + """The zids this shard owns, decided by the PRODUCTION filter. + + Deliberately delegates to :func:`should_process_zid` rather than + recomputing ``zid % shard_count``: a benchmark that reimplements the thing + under test can scale beautifully while the shipped code does not. + """ + return [ + z for z in zids if should_process_zid(z, [], [], shard_index, shard_count) + ] + + +def blas_env(base: dict[str, str], *, pin: bool) -> dict[str, str]: + """Child environment with BLAS threads pinned to 1, or explicitly unpinned. + + ``pin=False`` REMOVES the variables rather than leaving them alone: the + parent shell may already export them (certify and CI both do), and an + inherited "1" would silently pin the control arm and erase the very + difference this arm exists to show. + """ + env = dict(base) + for var in BLAS_ENV_VARS: + if pin: + env[var] = "1" + else: + env.pop(var, None) + return env + + +@dataclass(frozen=True) +class ShardResult: + """One shard process's own report of its compute phase.""" + + shard_index: int + ticks: int + compute_seconds: float + # user+sys CPU consumed by this shard during the compute phase. + cpu_seconds: float = 0.0 + + +@dataclass(frozen=True) +class ArmResult: + """One (shard_count) arm: all its shards, aggregated.""" + + shard_count: int + ticks: int + wall_seconds: float + throughput: float + cpu_seconds: float = 0.0 + cpu_per_tick: float = 0.0 + + +def summarize_arm(shard_count: int, results: Sequence[ShardResult]) -> ArmResult: + """Aggregate an arm. Wall = the SLOWEST shard, since they run concurrently. + + ``cpu_per_tick`` is the diagnostic that makes a disappointing speedup + interpretable: if it stays flat as N grows, each shard is doing the same + work and the wall-clock ceiling is core availability (a property of the + BOX). If it climbs, the shards are genuinely interfering (a property of + the MECHANISM). Without it, a sub-linear number cannot be attributed. + """ + if not results: + raise ValueError(f"no shard results for shard_count={shard_count}") + wall = max(r.compute_seconds for r in results) + if wall <= 0: + raise ValueError( + f"non-positive wall {wall!r} for shard_count={shard_count}: " + "the compute phase was not measured" + ) + ticks = sum(r.ticks for r in results) + cpu = sum(r.cpu_seconds for r in results) + return ArmResult( + shard_count=shard_count, + ticks=ticks, + wall_seconds=wall, + throughput=ticks / wall, + cpu_seconds=cpu, + cpu_per_tick=cpu / ticks if ticks else 0.0, + ) + + +def load_warning( + *, load1: float, cpu_count: int | None, shard_count: int +) -> str | None: + """Warn when the machine cannot actually give ``shard_count`` shards a core. + + A scaling sweep on a loaded box measures the BOX. This is not hypothetical: + a sweep taken at load 9.0 on a 10-core laptop reported 3.60x at N=8 while + CPU per tick stayed flat — the shards were starved, not contending, and + nothing in the table said so. + """ + if not cpu_count or shard_count <= 1: + return None + free = cpu_count - load1 + if free >= shard_count: + return None + return ( + f"WARNING: load average {load1:.1f} on {cpu_count} cores leaves ~{free:.1f} " + f"free, but the largest arm wants {shard_count}. Wall-clock speedup is a " + "FLOOR, not the mechanism's ceiling — compare cpu/tick instead, and " + "re-run on a quiet machine for a real scaling number." + ) + + +def best_arm(arms: Sequence[ArmResult]) -> ArmResult: + """The fastest repeat of one arm. + + Timing noise on a shared machine is one-sided: background load can only ADD + wall time. The minimum is therefore the best estimate of the true cost, + where a mean would encode whatever else the box happened to be running. + """ + if not arms: + raise ValueError("no arms to choose from") + counts = {a.shard_count for a in arms} + if len(counts) != 1: + raise ValueError(f"all repeats must share the same shard_count, got {counts}") + return max(arms, key=lambda a: a.throughput) + + +def karp_flatt(speedup: float, shard_count: int) -> float | None: + """Karp-Flatt experimentally-determined serial fraction. + + ``e = (1/S - 1/N) / (1 - 1/N)``. Reported because the handoff quotes + serial fractions (py-threads 0.9884, py-zid-shard 0.0013), so this is the + directly comparable statistic — and unlike raw speedup it exposes overhead + that grows with N. Undefined for a single worker. + """ + if shard_count <= 1: + return None + inv_n = 1.0 / shard_count + return (1.0 / speedup - inv_n) / (1.0 - inv_n) + + +def scaling_table(arms: Iterable[ArmResult]) -> list[dict[str, Any]]: + """Rows of (shard_count, ticks, wall, throughput, speedup, efficiency, + serial_fraction), speedup measured against the single-shard arm.""" + ordered = sorted(arms, key=lambda a: a.shard_count) + baseline = next((a for a in ordered if a.shard_count == 1), None) + if baseline is None: + raise ValueError( + "no shard_count=1 baseline arm: speedup is meaningless without it" + ) + rows: list[dict[str, Any]] = [] + for arm in ordered: + speedup = arm.throughput / baseline.throughput + rows.append( + { + "shard_count": arm.shard_count, + "ticks": arm.ticks, + "wall_seconds": arm.wall_seconds, + "throughput": arm.throughput, + "speedup": speedup, + "efficiency": speedup / arm.shard_count, + "serial_fraction": karp_flatt(speedup, arm.shard_count), + "cpu_per_tick": arm.cpu_per_tick, + # Flat across N => same work per tick, so any wall-clock + # shortfall is core availability, not sharding overhead. + "cpu_per_tick_vs_baseline": ( + arm.cpu_per_tick / baseline.cpu_per_tick + if baseline.cpu_per_tick else None + ), + } + ) + return rows + + +def verdict( + rows: Sequence[dict[str, Any]], *, min_efficiency: float = DEFAULT_MIN_EFFICIENCY +) -> dict[str, Any]: + """Quasi-linear iff parallel efficiency at the LARGEST arm clears the bar. + + The largest arm is the honest place to judge: efficiency decays with N, so + a mid-range arm can look fine while the top one has already collapsed. + """ + if len(rows) < 2: + raise ValueError("need at least two arms (a baseline and one more)") + top = max(rows, key=lambda r: r["shard_count"]) + return { + "quasi_linear": bool(top["efficiency"] >= min_efficiency), + "max_shard_count": top["shard_count"], + "speedup": top["speedup"], + "efficiency": top["efficiency"], + "serial_fraction": top["serial_fraction"], + "min_efficiency": min_efficiency, + } + + +# --------------------------------------------------------------------------- # +# Child: one shard process +# --------------------------------------------------------------------------- # +def run_shard_workload( + dataset_slug: str, + zids: Sequence[int], + shard_index: int, + shard_count: int, + *, + n_cuts: int, + ready_path: Path | None = None, + go_path: Path | None = None, +) -> ShardResult: + """Replay each owned zid and report ONLY the compute phase. + + Imports, dataset load and conversation setup happen before the barrier, so + the measured window contains math and nothing else. + """ + # Imported here, not at module scope: the parent process orchestrates and + # must not pay numpy/scipy import cost, and the child must pay it BEFORE + # the barrier so it lands outside the timed window. + from polismath.replay.driver import run_replay + from polismath.replay.real_data import load_export_votes + from polismath.replay.schedule import ScheduleSpec + + owned = shard_workload(zids, shard_index, shard_count) + dataset = load_export_votes(dataset_slug) + n_votes = len(dataset.votes) + cuts = [round(n_votes * (i + 1) / n_cuts) for i in range(n_cuts)] + spec = ScheduleSpec( + dataset=dataset_slug, + schedule_id=f"shardbench{n_cuts}", + cuts={"mode": "vote-count", "at": cuts}, + ) + + # Barrier: every shard signals readiness, then waits to be released, so the + # arms' compute phases actually overlap. Without it a staggered start lets + # early shards run alone, understating contention and overstating speedup. + if ready_path is not None: + ready_path.write_text("ready", encoding="utf-8") + if go_path is not None: + while not go_path.exists(): + time.sleep(0.01) + + import resource + + ticks = 0 + ru0 = resource.getrusage(resource.RUSAGE_SELF) + t0 = time.perf_counter() + for _zid in owned: + ticks += len(run_replay(dataset, spec)) + compute = time.perf_counter() - t0 + ru1 = resource.getrusage(resource.RUSAGE_SELF) + cpu = (ru1.ru_utime - ru0.ru_utime) + (ru1.ru_stime - ru0.ru_stime) + + return ShardResult( + shard_index=shard_index, ticks=ticks, compute_seconds=compute, + cpu_seconds=cpu, + ) + + +# --------------------------------------------------------------------------- # +# Parent: orchestrate one arm, then the sweep +# --------------------------------------------------------------------------- # +def _child_cmd( + dataset: str, zid_count: int, shard_index: int, shard_count: int, + n_cuts: int, ready: Path, go: Path, +) -> list[str]: + return [ + sys.executable, "-m", "polismath.replay.shard_bench", + "--dataset", dataset, + "--zid-count", str(zid_count), + "--shard-index", str(shard_index), + "--shard-count", str(shard_count), + "--cuts", str(n_cuts), + "--ready-file", str(ready), + "--go-file", str(go), + ] + + +def run_arm( + dataset: str, zid_count: int, shard_count: int, *, n_cuts: int, pin: bool, + work_dir: Path, log: Any = None, +) -> ArmResult: + """Spawn ``shard_count`` real processes, barrier them, collect their reports.""" + work_dir.mkdir(parents=True, exist_ok=True) + go = work_dir / f"go-{shard_count}-{int(pin)}" + go.unlink(missing_ok=True) + env = blas_env(dict(os.environ), pin=pin) + + # Child output goes to FILES, never pipes. conversation.py logs several KB + # per tick to stderr; a 64KB pipe fills long before a shard finishes, and + # the child then blocks on write until the parent reads it. Because the + # parent drains shard-by-shard (communicate() below), shard 0 would run at + # full speed while every other shard sat blocked awaiting its turn — which + # serialises the arm and silently destroys the measurement. Measured on an + # IDLE 16-core r8g.4xlarge before this fix: 1.05x at N=2, cpu/tick flat. + procs: list[tuple[int, subprocess.Popen, Path, Path, Path, Any, Any]] = [] + for idx in range(shard_count): + ready = work_dir / f"ready-{shard_count}-{int(pin)}-{idx}" + ready.unlink(missing_ok=True) + out_path = work_dir / f"out-{shard_count}-{int(pin)}-{idx}.txt" + err_path = work_dir / f"err-{shard_count}-{int(pin)}-{idx}.txt" + out_fh = out_path.open("w", encoding="utf-8") + err_fh = err_path.open("w", encoding="utf-8") + proc = subprocess.Popen( + _child_cmd(dataset, zid_count, idx, shard_count, n_cuts, ready, go), + cwd=str(_DELPHI_ROOT), env=env, + stdout=out_fh, stderr=err_fh, text=True, + ) + procs.append((idx, proc, ready, out_path, err_path, out_fh, err_fh)) + + # Wait for every child to finish its setup, then release them together. + deadline = time.monotonic() + CHILD_TIMEOUT_SEC + while not all(r.exists() for _, _, r, _, _, _, _ in procs): + dead = [ + (i, p, ep) for i, p, _, _, ep, _, _ in procs if p.poll() is not None + ] + if dead: + for _, fh in [(p, fh) for _, p, _, _, _, fh, _ in procs]: + fh.close() + i, p, ep = dead[0] + err = ep.read_text(encoding="utf-8", errors="replace") if ep.exists() else "" + raise RuntimeError( + f"shard {i} died before the barrier (rc={p.returncode}):\n" + f"{err.strip()[-2000:]}" + ) + if time.monotonic() > deadline: + for _, p, _, _, _, _, _ in procs: + p.kill() + raise RuntimeError("timed out waiting for shards to become ready") + time.sleep(0.01) + go.write_text("go", encoding="utf-8") + + results: list[ShardResult] = [] + for idx, proc, _, out_path, err_path, out_fh, err_fh in procs: + proc.wait(timeout=CHILD_TIMEOUT_SEC) + out_fh.close() + err_fh.close() + if proc.returncode != 0: + err = err_path.read_text(encoding="utf-8", errors="replace") + raise RuntimeError( + f"shard {idx}/{shard_count} failed (rc={proc.returncode}):\n" + f"{err.strip()[-2000:]}" + ) + out = out_path.read_text(encoding="utf-8", errors="replace").strip() + if not out: + err = err_path.read_text(encoding="utf-8", errors="replace") + raise RuntimeError( + f"shard {idx}/{shard_count} produced no result line:\n" + f"{err.strip()[-2000:]}" + ) + payload = json.loads(out.splitlines()[-1]) + results.append(ShardResult(**payload)) + + arm = summarize_arm(shard_count, results) + if log is not None: + log( + f" N={shard_count:>2} ticks={arm.ticks:>4} " + f"wall={arm.wall_seconds:6.2f}s {arm.throughput:6.2f} ticks/s" + ) + return arm + + +def run_sweep( + *, dataset: str = DEFAULT_DATASET, zid_count: int = DEFAULT_ZID_COUNT, + shard_counts: Sequence[int] = DEFAULT_SHARD_COUNTS, n_cuts: int = 4, + pin: bool = True, work_dir: Path, min_efficiency: float = DEFAULT_MIN_EFFICIENCY, + repeats: int = 1, log: Any = None, +) -> dict[str, Any]: + """Run every arm ``repeats`` times, keep each arm's fastest, and judge. + + Load average is recorded because it is the single biggest confounder on a + developer machine: a sweep taken under heavy background load understates + scaling, and a reader cannot tell that from the table alone. + """ + for n in shard_counts: + if zid_count % n: + raise ValueError( + f"zid_count={zid_count} is not divisible by shard_count={n}: " + "an uneven split would measure a remainder, not scaling" + ) + load_before = os.getloadavg() + arms: list[ArmResult] = [] + for n in sorted(shard_counts): + repeats_for_n = [ + run_arm(dataset, zid_count, n, n_cuts=n_cuts, pin=pin, + work_dir=work_dir, log=log) + for _ in range(max(1, repeats)) + ] + arms.append(best_arm(repeats_for_n)) + rows = scaling_table(arms) + return { + "dataset": dataset, + "zid_count": zid_count, + "cuts_per_zid": n_cuts, + "blas_pinned": pin, + "repeats": repeats, + "cpu_count": os.cpu_count(), + "load_before": load_before, + "load_after": os.getloadavg(), + "rows": rows, + "verdict": verdict(rows, min_efficiency=min_efficiency), + } + + +# --------------------------------------------------------------------------- # +# Child entrypoint (python -m polismath.replay.shard_bench) +# --------------------------------------------------------------------------- # +def _main(argv: Sequence[str]) -> int: + import argparse + + ap = argparse.ArgumentParser(description="one shard of the scaling benchmark") + ap.add_argument("--dataset", required=True) + ap.add_argument("--zid-count", type=int, required=True) + ap.add_argument("--shard-index", type=int, required=True) + ap.add_argument("--shard-count", type=int, required=True) + ap.add_argument("--cuts", type=int, default=4) + ap.add_argument("--ready-file") + ap.add_argument("--go-file") + args = ap.parse_args(list(argv)) + + result = run_shard_workload( + args.dataset, + list(range(args.zid_count)), + args.shard_index, + args.shard_count, + n_cuts=args.cuts, + ready_path=Path(args.ready_file) if args.ready_file else None, + go_path=Path(args.go_file) if args.go_file else None, + ) + print(json.dumps(result.__dict__)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(_main(sys.argv[1:])) diff --git a/delphi/scripts/shard_scaling_bench.py b/delphi/scripts/shard_scaling_bench.py new file mode 100644 index 000000000..02f3fcf5a --- /dev/null +++ b/delphi/scripts/shard_scaling_bench.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Shard-scaling benchmark CLI — does throughput scale with the shard count? + +Answers HANDOFF_PYTHON_SHARDING.md §7's open acceptance criterion ("with N +shards on an N-core box, aggregate throughput should approach the measured +15.7x at 16 rather than the current 1.0x") for the SHIPPED ``zid % N`` filter, +which the cost study's arm never exercised (§8). + +Thin wrapper over ``polismath.replay.shard_bench`` — same split as +``scripts/poller_equiv.py`` and ``scripts/certify.py``: the library stays pure +/ side-effect-scoped, this script owns printing and process exit codes. + +Usage (from delphi/):: + + # Default sweep: biodiversity, 24 zids, N = 1,2,4,8, BLAS pinned. + uv run python scripts/shard_scaling_bench.py run + + # The §0 control arm: identical sweep with BLAS threads UNPINNED, which is + # what production runs today (nothing sets OMP_NUM_THREADS anywhere). + uv run python scripts/shard_scaling_bench.py run --no-pin + + # Both arms, so pinned vs unpinned is measured rather than asserted: + uv run python scripts/shard_scaling_bench.py run --both + +Each arm spawns real processes and takes ~a minute of CPU, so this is opt-in +tooling, never part of the pytest suite. Exit code is 1 if the pinned sweep +fails the quasi-linear bar. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import click + +from polismath.replay import shard_bench as sb + + +def _echo(msg: str) -> None: + click.echo(msg, err=True) + + +def _render(report: dict) -> list[str]: + """Compact table — one line per arm, plus the verdict.""" + pinned = "pinned" if report["blas_pinned"] else "UNPINNED" + lines = [ + f"dataset={report['dataset']} zids={report['zid_count']} " + f"cuts/zid={report['cuts_per_zid']} blas={pinned} " + f"cores={report['cpu_count']} repeats={report['repeats']} " + f"load={report['load_before'][0]:.2f}->{report['load_after'][0]:.2f}", + f"{'N':>3} {'ticks':>6} {'wall_s':>8} {'ticks/s':>8} " + f"{'speedup':>8} {'effic':>6} {'serial_f':>9} {'cpu/tick':>9} {'vs N=1':>7}", + ] + for r in report["rows"]: + sf = r["serial_fraction"] + rel = r["cpu_per_tick_vs_baseline"] + lines.append( + f"{r['shard_count']:>3} {r['ticks']:>6} {r['wall_seconds']:>8.2f} " + f"{r['throughput']:>8.2f} {r['speedup']:>7.2f}x " + f"{r['efficiency']:>6.2f} {'--' if sf is None else f'{sf:>9.4f}'} " + f"{r['cpu_per_tick']:>9.3f} " + f"{'--' if rel is None else f'{rel:>6.2f}x'}" + ) + warn = sb.load_warning( + load1=report["load_before"][0], cpu_count=report["cpu_count"], + shard_count=report["verdict"]["max_shard_count"], + ) + if warn: + lines.append(warn) + v = report["verdict"] + lines.append( + f"VERDICT: {'QUASI-LINEAR' if v['quasi_linear'] else 'NOT quasi-linear'} " + f"— {v['speedup']:.2f}x at N={v['max_shard_count']} " + f"(efficiency {v['efficiency']:.2f}, bar {v['min_efficiency']:.2f}; " + f"serial fraction {v['serial_fraction']:.4f})" + ) + return lines + + +@click.group() +def cli() -> None: + """Shard-scaling benchmark (see module docstring).""" + + +@cli.command() +@click.option("--dataset", default=sb.DEFAULT_DATASET, show_default=True, + help="Dataset slug replayed once per zid.") +@click.option("--zid-count", type=int, default=sb.DEFAULT_ZID_COUNT, + show_default=True, + help="Fixed workload size; must divide every shard count.") +@click.option("--shard-counts", default=",".join(map(str, sb.DEFAULT_SHARD_COUNTS)), + show_default=True, help="Comma-separated arms to run.") +@click.option("--cuts", type=int, default=4, show_default=True, + help="Schedule cuts per zid (= math ticks per zid).") +@click.option("--pin/--no-pin", default=True, show_default=True, + help="Pin BLAS/OpenMP threads to 1 in each shard (handoff §0).") +@click.option("--both", is_flag=True, + help="Run pinned AND unpinned sweeps, to measure the difference.") +@click.option("--repeats", type=int, default=1, show_default=True, + help="Run each arm this many times and keep the fastest — " + "background load only ever ADDS wall time.") +@click.option("--min-efficiency", type=float, default=sb.DEFAULT_MIN_EFFICIENCY, + show_default=True, help="Parallel-efficiency bar at the largest arm.") +@click.option("--out", type=click.Path(path_type=Path), default=None, + help="Write the full report JSON here.") +def run(dataset, zid_count, shard_counts, cuts, pin, both, repeats, + min_efficiency, out): + """Run the sweep and report speedup / efficiency / serial fraction.""" + counts = [int(x) for x in shard_counts.split(",") if x.strip()] + work_dir = Path("scratch/shard_bench") + arms = [True, False] if both else [pin] + + reports = [] + for do_pin in arms: + _echo(f"--- sweep: BLAS {'pinned to 1' if do_pin else 'UNPINNED'} ---") + report = sb.run_sweep( + dataset=dataset, zid_count=zid_count, shard_counts=counts, + n_cuts=cuts, pin=do_pin, work_dir=work_dir, + min_efficiency=min_efficiency, repeats=repeats, log=_echo, + ) + reports.append(report) + + for report in reports: + click.echo("") + for line in _render(report): + click.echo(line) + + if out is not None: + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(json.dumps(reports, indent=2), encoding="utf-8") + _echo(f"\nreport written to {out}") + + # Judge the PINNED sweep — the unpinned arm is a control, and is expected + # to scale badly. If only --no-pin was requested, judge that. + judged = reports[0] + return 0 if judged["verdict"]["quasi_linear"] else 1 + + +if __name__ == "__main__": + sys.exit(cli(standalone_mode=False) or 0) diff --git a/delphi/tests/poller/test_service.py b/delphi/tests/poller/test_service.py index d8721a90f..cc2046c42 100644 --- a/delphi/tests/poller/test_service.py +++ b/delphi/tests/poller/test_service.py @@ -99,6 +99,77 @@ def test_no_configured_mode_leaves_compute_default(self, monkeypatch): assert resolved == "improved" # engine_mode.ENGINE_MODE_DEFAULT +class TestShardedDispatch: + """Two shard processes over the SAME polled rows must partition the work: + disjoint (nothing double-processed, since per-zid serialisation does not + span processes) and total (nothing dropped).""" + + ZIDS = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] + + def _dispatch(self, **cfg_kwargs): + """Run one vote poll over ZIDS and return the zids actually submitted.""" + pg = MagicMock() + pg.poll_votes_since.return_value = [ + _vote_row(z, 100 + z) for z in self.ZIDS + ] + svc = MathPollerService(pg, PollerConfig(**cfg_kwargs)) + svc._ensure_runtime() + svc._vote_wm = 0 + submitted = [] + svc._pool.submit = lambda zid, mt, batch: submitted.append(zid) + svc._poll_votes_once() + return submitted, svc + + def test_two_shards_partition_the_polled_zids(self): + shard0, _ = self._dispatch(shard_index=0, shard_count=2) + shard1, _ = self._dispatch(shard_index=1, shard_count=2) + unsharded, _ = self._dispatch() + + # Disjoint: no zid dispatched by both shards. + assert set(shard0) & set(shard1) == set() + # Total: together they cover exactly the unsharded dispatch set. + assert set(shard0) | set(shard1) == set(unsharded) + # And each is a strict, non-empty subset -- proving the filter fired. + assert shard0 and shard1 + assert set(shard0) == {z for z in self.ZIDS if z % 2 == 0} + + def test_no_zid_is_dispatched_twice_across_the_fleet(self): + seen = [] + for idx in range(3): + dispatched, _ = self._dispatch(shard_index=idx, shard_count=3) + seen.extend(dispatched) + assert sorted(seen) == sorted(self.ZIDS) + assert len(seen) == len(set(seen)) + + def test_each_shard_still_advances_its_own_watermark_past_all_rows(self): + # Each shard owns its watermark in memory and discards rows belonging to + # its siblings -- so it must advance past them, exactly as the existing + # allowlist behaviour does (test_watermark_advances_past_all_rows...). + for idx in range(2): + _, svc = self._dispatch(shard_index=idx, shard_count=2) + assert svc._vote_wm == 100 + max(self.ZIDS) + + def test_moderation_dispatch_is_sharded_too(self): + pg = MagicMock() + pg.poll_moderation_since.return_value = [ + {"zid": z, "tid": 1, "modified": 200 + z, "mod": -1, "is_meta": False} + for z in self.ZIDS + ] + svc = MathPollerService(pg, PollerConfig(shard_index=1, shard_count=2)) + svc._ensure_runtime() + svc._mod_wm = 0 + submitted = [] + svc._pool.submit = lambda zid, mt, batch: submitted.append(zid) + + svc._poll_moderation_once() + + assert set(submitted) == {z for z in self.ZIDS if z % 2 == 1} + + def test_unsharded_default_dispatches_everything(self): + dispatched, _ = self._dispatch() + assert dispatched == self.ZIDS + + class TestConvCacheEviction: """T8: the in-memory conv registry never evicted (Clojure's 4h reboot was the de-facto cap, which we dropped). LRU-evict beyond a configurable cap; an diff --git a/delphi/tests/poller/test_watermark.py b/delphi/tests/poller/test_watermark.py index 453592c12..9f6fe77ab 100644 --- a/delphi/tests/poller/test_watermark.py +++ b/delphi/tests/poller/test_watermark.py @@ -5,7 +5,10 @@ and the allow/block cond (poller.clj:30-32). """ +import pytest + from polismath.poller.service import ( + PollerConfig, advance_watermark, should_process_zid, initial_watermark, @@ -72,3 +75,115 @@ def test_allowlist_takes_priority_over_blocklist(self): # Clojure cond: allowlist branch evaluated first. assert should_process_zid(42, [42], [42]) is True assert should_process_zid(7, [42], [7]) is False + + +class TestZidSharding: + """zid-sharding: one shard = one process, selected by ``zid % shard_count``. + + Threads cannot parallelise this workload (measured serial fraction 0.9884, + 1.0x from 1->16 workers), while N independent single-worker PROCESSES scale + near-linearly (0.0013, 15.7x at 16). Sharding is pure scheduling + scaffolding: it must never change what any single conversation computes. + """ + + def test_default_is_unsharded_and_identical_to_today(self): + # Regression guard: sharding is opt-in. With the default shard_count=1 + # every zid still passes, exactly as before the parameter existed. + for zid in range(0, 50): + assert should_process_zid(zid, [], []) is True + assert should_process_zid(zid, [], [], shard_index=0, shard_count=1) is True + + def test_partition_is_total_and_disjoint(self): + # Every zid must be accepted by EXACTLY ONE shard index -- no zid + # dropped (total) and none double-processed (disjoint). The range + # deliberately spans zids where zid % N == 0. + for shard_count in (2, 3, 4, 8): + for zid in range(0, 100): + accepted = [ + idx + for idx in range(shard_count) + if should_process_zid( + zid, [], [], shard_index=idx, shard_count=shard_count + ) + ] + assert accepted == [zid % shard_count], ( + f"zid={zid} shard_count={shard_count} accepted by {accepted}" + ) + + def test_shard_filter_beats_an_allowlist_naming_an_out_of_slice_zid(self): + # Ordering is a CORRECTNESS property, not style: the worker pool + # serialises per zid only WITHIN a process, so if two shards both + # accepted one zid they would run concurrent updates on the same + # conversation with no mutual exclusion. + assert should_process_zid(7, [7], [], shard_index=1, shard_count=2) is True + assert should_process_zid(7, [7], [], shard_index=0, shard_count=2) is False + + def test_blocklist_still_excludes_an_in_slice_zid(self): + # In-slice for shard 0 of 2, but blocked -> still excluded. + assert should_process_zid(8, [], [8], shard_index=0, shard_count=2) is False + assert should_process_zid(6, [], [8], shard_index=0, shard_count=2) is True + + +class TestShardConfigValidation: + """A silently out-of-range shard index is the worst failure mode here: the + shard processes NOTHING while looking healthy, so a slice of conversations + goes stale behind an apparently-up fleet. Fail loudly at config time.""" + + def test_index_equal_to_count_is_rejected(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "4") + monkeypatch.setenv("POLL_SHARD_INDEX", "4") + with pytest.raises(ValueError, match="shard_index"): + PollerConfig.from_env() + + def test_index_above_count_is_rejected(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "2") + monkeypatch.setenv("POLL_SHARD_INDEX", "9") + with pytest.raises(ValueError, match="shard_index"): + PollerConfig.from_env() + + def test_negative_index_is_rejected(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "4") + monkeypatch.setenv("POLL_SHARD_INDEX", "-1") + with pytest.raises(ValueError, match="shard_index"): + PollerConfig.from_env() + + def test_shard_count_below_one_is_rejected(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "0") + with pytest.raises(ValueError, match="shard_count"): + PollerConfig.from_env() + + def test_negative_shard_count_is_rejected(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "-3") + with pytest.raises(ValueError, match="shard_count"): + PollerConfig.from_env() + + def test_valid_shard_config_is_accepted(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "8") + monkeypatch.setenv("POLL_SHARD_INDEX", "7") + cfg = PollerConfig.from_env() + assert (cfg.shard_index, cfg.shard_count) == (7, 8) + + def test_defaults_are_unsharded(self, monkeypatch): + monkeypatch.delenv("POLL_SHARD_COUNT", raising=False) + monkeypatch.delenv("POLL_SHARD_INDEX", raising=False) + monkeypatch.delenv("MATH_SHARD_COUNT", raising=False) + monkeypatch.delenv("MATH_SHARD_INDEX", raising=False) + cfg = PollerConfig.from_env() + assert (cfg.shard_index, cfg.shard_count) == (0, 1) + + def test_math_prefixed_aliases_are_honored(self, monkeypatch): + # Dual-name convention, matching allowlist/blocklist (POLL_* preferred). + monkeypatch.delenv("POLL_SHARD_COUNT", raising=False) + monkeypatch.delenv("POLL_SHARD_INDEX", raising=False) + monkeypatch.setenv("MATH_SHARD_COUNT", "3") + monkeypatch.setenv("MATH_SHARD_INDEX", "2") + cfg = PollerConfig.from_env() + assert (cfg.shard_index, cfg.shard_count) == (2, 3) + + def test_poll_prefix_wins_over_math_alias(self, monkeypatch): + monkeypatch.setenv("POLL_SHARD_COUNT", "4") + monkeypatch.setenv("POLL_SHARD_INDEX", "1") + monkeypatch.setenv("MATH_SHARD_COUNT", "9") + monkeypatch.setenv("MATH_SHARD_INDEX", "8") + cfg = PollerConfig.from_env() + assert (cfg.shard_index, cfg.shard_count) == (1, 4) diff --git a/delphi/tests/replay_harness/test_shard_bench.py b/delphi/tests/replay_harness/test_shard_bench.py new file mode 100644 index 000000000..d854fcc99 --- /dev/null +++ b/delphi/tests/replay_harness/test_shard_bench.py @@ -0,0 +1,331 @@ +"""Unit surface for the shard-scaling benchmark (HANDOFF_PYTHON_SHARDING.md §7). + +The benchmark itself needs N real processes and ~a minute of CPU, so it is an +opt-in script. Every pure decision point it rests on is covered here with +canned numbers: the workload partition (which must go through the REAL +should_process_zid, not a reimplementation), the BLAS pinning env, and the +scaling arithmetic (throughput / speedup / efficiency / Karp-Flatt). +""" + +import pytest + +from polismath.replay import shard_bench as sb_mod +from polismath.poller.service import should_process_zid +from polismath.replay.shard_bench import ( + BLAS_ENV_VARS, + ShardResult, + best_arm, + blas_env, + karp_flatt, + load_warning, + scaling_table, + shard_workload, + summarize_arm, + verdict, +) + + +class TestShardWorkload: + """The benchmark must partition with the SAME function production uses -- + otherwise it measures a reimplementation and proves nothing about the + shipped filter.""" + + def test_partition_matches_the_real_filter(self): + zids = list(range(50)) + for shard_count in (1, 2, 4, 8): + for idx in range(shard_count): + assert shard_workload(zids, idx, shard_count) == [ + z for z in zids if should_process_zid(z, [], [], idx, shard_count) + ] + + def test_partition_is_total_and_disjoint(self): + zids = list(range(48)) + for shard_count in (1, 2, 3, 4, 8): + owned = [shard_workload(zids, i, shard_count) for i in range(shard_count)] + flat = [z for chunk in owned for z in chunk] + assert sorted(flat) == zids # total: nothing dropped + assert len(flat) == len(set(flat)) # disjoint: nothing doubled + + def test_balanced_when_divisible(self): + # A COST-balanced workload is the point: every zid replays the same + # dataset, so an even COUNT split is an even WORK split. Skew is a real + # production concern but would confound a scaling measurement. + zids = list(range(24)) + for shard_count in (1, 2, 4, 8): + sizes = {len(shard_workload(zids, i, shard_count)) for i in range(shard_count)} + assert sizes == {24 // shard_count} + + def test_single_shard_owns_everything(self): + zids = list(range(10)) + assert shard_workload(zids, 0, 1) == zids + + +class TestBlasEnv: + """Unpinned numpy fans one recompute across every core (measured cpu/wall + 8.75 on r8g; on a 10-core laptop it is SLOWER in wall time and burns ~7x + the CPU). N such shards on one box thrash, so every shard pins to 1.""" + + def test_pinning_sets_every_known_blas_var_to_one(self): + env = blas_env({"PATH": "/bin"}, pin=True) + for var in BLAS_ENV_VARS: + assert env[var] == "1", var + assert env["PATH"] == "/bin" # base env preserved + + def test_unpinned_removes_them_so_numpy_uses_its_default(self): + # Inherited values would silently pin the "unpinned" control arm and + # collapse the comparison the correction section is about. + base = {"PATH": "/bin", "OMP_NUM_THREADS": "1", "OPENBLAS_NUM_THREADS": "1"} + env = blas_env(base, pin=False) + for var in BLAS_ENV_VARS: + assert var not in env, var + assert env["PATH"] == "/bin" + + def test_base_env_is_not_mutated(self): + base = {"PATH": "/bin"} + blas_env(base, pin=True) + assert base == {"PATH": "/bin"} + + +class TestSummarizeArm: + def test_wall_is_the_slowest_shard_not_the_sum(self): + # Shards run concurrently: the arm finishes when the LAST one does. + results = [ + ShardResult(shard_index=0, ticks=12, compute_seconds=4.0), + ShardResult(shard_index=1, ticks=12, compute_seconds=5.0), + ] + arm = summarize_arm(2, results) + assert arm.ticks == 24 + assert arm.wall_seconds == 5.0 + assert arm.throughput == pytest.approx(24 / 5.0) + + def test_single_shard_arm(self): + arm = summarize_arm(1, [ShardResult(0, ticks=24, compute_seconds=20.0)]) + assert arm.throughput == pytest.approx(1.2) + + def test_cpu_seconds_sum_across_shards_and_per_tick(self): + """CPU per tick is what separates 'sharding costs extra work' from + 'the box has no free cores'. Wall can stall for want of a core while + CPU per tick stays flat — that is a machine limit, not a mechanism + limit, and only this statistic can tell them apart.""" + results = [ + ShardResult(0, ticks=12, compute_seconds=5.0, cpu_seconds=4.0), + ShardResult(1, ticks=12, compute_seconds=5.0, cpu_seconds=6.0), + ] + arm = summarize_arm(2, results) + assert arm.cpu_seconds == pytest.approx(10.0) + assert arm.cpu_per_tick == pytest.approx(10.0 / 24) + + def test_cpu_defaults_to_zero_when_unreported(self): + arm = summarize_arm(1, [ShardResult(0, ticks=24, compute_seconds=20.0)]) + assert arm.cpu_seconds == pytest.approx(0.0) + assert arm.cpu_per_tick == pytest.approx(0.0) + + def test_zero_wall_is_rejected_rather_than_dividing_by_zero(self): + with pytest.raises(ValueError, match="wall"): + summarize_arm(1, [ShardResult(0, ticks=5, compute_seconds=0.0)]) + + def test_empty_results_rejected(self): + with pytest.raises(ValueError, match="no shard results"): + summarize_arm(2, []) + + +class TestKarpFlatt: + """Karp-Flatt experimentally-determined serial fraction -- directly + comparable to the handoff's quoted serial fractions (py-threads 0.9884, + py-zid-shard 0.0013).""" + + def test_perfect_linear_speedup_is_zero_serial_fraction(self): + assert karp_flatt(8.0, 8) == pytest.approx(0.0, abs=1e-12) + + def test_no_speedup_at_all_is_fully_serial(self): + assert karp_flatt(1.0, 2) == pytest.approx(1.0) + + def test_half_speedup_is_intermediate(self): + # S=4 at N=8 -> e = (1/4 - 1/8) / (1 - 1/8) = 0.125/0.875 + assert karp_flatt(4.0, 8) == pytest.approx(0.125 / 0.875) + + def test_undefined_for_a_single_worker(self): + assert karp_flatt(1.0, 1) is None + + +class TestScalingTable: + def _arms(self): + # Ideal linear scaling: throughput doubles with each doubling of N. + return [ + summarize_arm(1, [ShardResult(0, 24, 24.0)]), + summarize_arm(2, [ShardResult(i, 12, 12.0) for i in range(2)]), + summarize_arm(4, [ShardResult(i, 6, 6.0) for i in range(4)]), + ] + + def test_speedup_is_relative_to_the_single_shard_arm(self): + rows = scaling_table(self._arms()) + assert [r["shard_count"] for r in rows] == [1, 2, 4] + assert [r["speedup"] for r in rows] == pytest.approx([1.0, 2.0, 4.0]) + assert [r["efficiency"] for r in rows] == pytest.approx([1.0, 1.0, 1.0]) + + def test_serial_fraction_reported_per_arm(self): + rows = scaling_table(self._arms()) + assert rows[0]["serial_fraction"] is None # undefined at N=1 + assert rows[2]["serial_fraction"] == pytest.approx(0.0, abs=1e-12) + + def test_sublinear_scaling_shows_lost_efficiency(self): + arms = [ + summarize_arm(1, [ShardResult(0, 24, 24.0)]), + summarize_arm(4, [ShardResult(i, 6, 12.0) for i in range(4)]), # 2x only + ] + rows = scaling_table(arms) + assert rows[1]["speedup"] == pytest.approx(2.0) + assert rows[1]["efficiency"] == pytest.approx(0.5) + assert rows[1]["serial_fraction"] == pytest.approx((0.5 - 0.25) / 0.75) + + def test_missing_baseline_is_rejected(self): + arms = [summarize_arm(2, [ShardResult(i, 12, 12.0) for i in range(2)])] + with pytest.raises(ValueError, match="baseline"): + scaling_table(arms) + + +class TestBestArm: + """Timing on a shared developer machine is noisy in ONE direction only: + background load can add wall time, never remove it. So across repeats of an + arm the fastest run is the closest estimate of the true cost, and taking a + mean would bake in whatever else the laptop was doing.""" + + def test_picks_the_fastest_repeat(self): + repeats = [ + summarize_arm(4, [ShardResult(i, 6, 12.0) for i in range(4)]), + summarize_arm(4, [ShardResult(i, 6, 6.0) for i in range(4)]), # best + summarize_arm(4, [ShardResult(i, 6, 9.0) for i in range(4)]), + ] + best = best_arm(repeats) + assert best.wall_seconds == 6.0 + assert best.throughput == pytest.approx(24 / 6.0) + + def test_single_repeat_passes_through(self): + only = summarize_arm(2, [ShardResult(i, 12, 8.0) for i in range(2)]) + assert best_arm([only]) is only + + def test_mixed_shard_counts_rejected(self): + with pytest.raises(ValueError, match="same shard_count"): + best_arm([ + summarize_arm(2, [ShardResult(0, 12, 8.0), ShardResult(1, 12, 8.0)]), + summarize_arm(4, [ShardResult(i, 6, 6.0) for i in range(4)]), + ]) + + def test_empty_rejected(self): + with pytest.raises(ValueError, match="no arms"): + best_arm([]) + + +# A stand-in shard: joins the barrier, floods stderr well past the 64KB pipe +# buffer, then reports a result. With stderr on a PIPE the parent — which only +# drains sequentially, at the end — leaves every child but the first blocked on +# write, serialising the arm. +_FLOODING_CHILD = """ +import sys, os, time, json +ready, go, idx = sys.argv[1], sys.argv[2], int(sys.argv[3]) +open(ready, "w").write("r") +while not os.path.exists(go): + time.sleep(0.01) +sys.stderr.write("x" * 300000) +sys.stderr.flush() +print(json.dumps({"shard_index": idx, "ticks": 4, + "compute_seconds": 0.5, "cpu_seconds": 0.4})) +""" + + +class TestChildOutputIsNotPiped: + """Regression guard for the bug that invalidated the first four sweeps. + + conversation.py logs several KB per tick to stderr. Piping that into a + buffer the parent only reads at the end blocks each child once 64KB fills, + and since the parent calls communicate() shard-by-shard, shard 0 runs at + full speed while the rest stall waiting their turn. Measured effect on an + IDLE 16-core r8g.4xlarge: 1.05x at N=2 with cpu/tick dead flat — the shards + were doing identical work and simply not running at the same time. + """ + + def test_arm_with_noisy_children_completes_and_collects(self, monkeypatch, tmp_path): + import sys + + def fake_cmd(dataset, zid_count, shard_index, shard_count, n_cuts, ready, go): + return [sys.executable, "-c", _FLOODING_CHILD, + str(ready), str(go), str(shard_index)] + + monkeypatch.setattr(sb_mod, "_child_cmd", fake_cmd) + arm = sb_mod.run_arm( + "vw", 8, 4, n_cuts=2, pin=True, work_dir=tmp_path / "wd" + ) + # All four shards reported despite each emitting ~300KB of stderr. + assert arm.shard_count == 4 + assert arm.ticks == 16 + + def test_stdout_and_stderr_are_never_subprocess_pipe(self, monkeypatch, tmp_path): + import subprocess as sp + import sys + + seen = [] + real_popen = sp.Popen + + def spy(cmd, **kw): + seen.append(kw) + return real_popen(cmd, **kw) + + def fake_cmd(dataset, zid_count, shard_index, shard_count, n_cuts, ready, go): + return [sys.executable, "-c", _FLOODING_CHILD, + str(ready), str(go), str(shard_index)] + + monkeypatch.setattr(sb_mod, "_child_cmd", fake_cmd) + monkeypatch.setattr(sb_mod.subprocess, "Popen", spy) + sb_mod.run_arm("vw", 4, 2, n_cuts=2, pin=True, work_dir=tmp_path / "wd") + + assert seen, "no child was spawned" + for kw in seen: + assert kw.get("stdout") is not sp.PIPE, "stdout must not be a pipe" + assert kw.get("stderr") is not sp.PIPE, "stderr must not be a pipe" + + +class TestLoadWarning: + """A sweep run on a busy box measures the box, not the mechanism — and the + table alone cannot show that. Discovered the hard way: a sweep at load 9 on + a 10-core machine reported 3.6x at N=8 with CPU/tick flat, i.e. the shards + were starved of cores rather than contending.""" + + def test_warns_when_load_leaves_too_few_free_cores(self): + warn = load_warning(load1=9.0, cpu_count=10, shard_count=8) + assert warn is not None + assert "9.0" in warn and "8" in warn + + def test_silent_on_a_quiet_machine(self): + assert load_warning(load1=0.4, cpu_count=10, shard_count=8) is None + + def test_unknown_cpu_count_does_not_crash(self): + assert load_warning(load1=9.0, cpu_count=None, shard_count=8) is None + + def test_warns_only_when_the_arm_actually_needs_the_cores(self): + # 1 free core is plenty for a single-shard arm. + assert load_warning(load1=9.0, cpu_count=10, shard_count=1) is None + + +class TestVerdict: + def test_quasi_linear_when_efficiency_holds_at_the_largest_arm(self): + rows = scaling_table([ + summarize_arm(1, [ShardResult(0, 24, 24.0)]), + summarize_arm(8, [ShardResult(i, 3, 3.3) for i in range(8)]), + ]) + v = verdict(rows, min_efficiency=0.8) + assert v["quasi_linear"] is True + assert v["max_shard_count"] == 8 + + def test_not_quasi_linear_when_the_largest_arm_degrades(self): + rows = scaling_table([ + summarize_arm(1, [ShardResult(0, 24, 24.0)]), + summarize_arm(8, [ShardResult(i, 3, 12.0) for i in range(8)]), # 2x + ]) + v = verdict(rows, min_efficiency=0.8) + assert v["quasi_linear"] is False + assert v["efficiency"] == pytest.approx(0.25) + + def test_verdict_needs_more_than_the_baseline_arm(self): + rows = scaling_table([summarize_arm(1, [ShardResult(0, 24, 24.0)])]) + with pytest.raises(ValueError, match="at least two"): + verdict(rows, min_efficiency=0.8)