From 27623840fa33ee6150e54f8eb391297e46a14cd9 Mon Sep 17 00:00:00 2001 From: Julien Cornebise Date: Sat, 25 Jul 2026 17:01:41 +0200 Subject: [PATCH] =?UTF-8?q?python-math=20#32:=20feat(delphi):=20zid-shardi?= =?UTF-8?q?ng=20for=20the=20math=20poller=20=E2=80=94=20one=20shard=20=3D?= =?UTF-8?q?=20one=20process?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Why: the poller's current parallelism mechanism does not work The math poller (the service that polls Postgres for new votes and runs math updates) parallelizes via `ConversationWorkerPool`, a `ThreadPoolExecutor` — and threads cannot execute this workload concurrently. Measured serial fraction: 0.9884 [0.976, 1.000]; throughput 1.66 ticks/s at 1 worker and 1.64 at 16 — `MATH_WORKER_POOL_SIZE` is effectively inert. N independent single-worker PROCESSES scale near-linearly instead: serial fraction 0.0013 [0.0009, 0.0014], 26.09 ticks/s at 16 (15.7x). Free-threading is not the escape hatch, and this was measured, not assumed: on Python 3.14t (GIL confirmed off) the same threaded code gives serial fraction 0.910 and a 1.1x ceiling, and at 8 workers is SLOWER than at 1 — numpy already releases the GIL for large array ops, so the GIL was never the only barrier. Measurements come from the cost-model study (r8g.4xlarge native; free-threaded arm on laptop); see `HANDOFF_PYTHON_SHARDING.md` (the handoff document from that study — the §N references below are its sections). ## What ships: opt-in sharding scaffolding only Scheduling scaffolding only — no math, blob (JSON math result) or write-path change whatsoever: - `should_process_zid()` takes `shard_index`/`shard_count` (a zid is a conversation ID; shard membership is `zid % shard_count == shard_index`), defaulting to 0/1 — a no-op — so sharding is strictly opt-in. - `PollerConfig` gains the two fields via the existing dual-name convention: `POLL_SHARD_INDEX` | `MATH_SHARD_INDEX`, `POLL_SHARD_COUNT` | `MATH_SHARD_COUNT`. - Both poll loops (votes, moderation) pass them through. - The startup log names the slice ("shard=1/4" or "shard=unsharded"). ## The shard filter runs BEFORE the allow/blocklist — a correctness property That ordering is not a style choice: `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. An allowlist naming an out-of-slice zid must therefore lose to the shard filter. Validation lives in `__post_init__`, not just `from_env`, so it covers direct construction too. An out-of-range index matches NO zid, so the process would start, poll, log happily and compute nothing — the fleet looks up while a slice of conversations silently goes stale. That fails loudly now. ## Watermarks: verified, no change needed Watermarks (the last-processed poll positions) need no change, and this was verified, not assumed: they are in-memory per-process attributes, never persisted, and already advance over ALL polled rows independently of the dispatch filter (the existing `test_watermark_advances_past_all_rows_even_filtered` pins this). Each shard therefore owns its own watermark and correctly advances past rows belonging to its siblings. ## MATH_WORKER_POOL_SIZE stays at 4 — a deliberate, recorded decision Threads cannot overlap math with math, but the Clojure implementation being replaced DOES parallelise per conversation, and a >1 pool may still overlap DB write I/O with math. The cost study measured a CPU-bound tick and cannot settle that, so it is documented in-code as a tuning parameter to measure once sharding is deployed. ## MEASURED: the §7 throughput criterion is now verified on prod-like hardware Sweeps on an idle r8g.4xlarge (16 Graviton4 cores, no SMT), best-of-3 repeats, BLAS pinned. (Karp-Flatt serial: the experimentally derived serial fraction of the workload; cpu/tick: CPU-seconds per math tick.) | N | ticks/s | speedup | efficiency | Karp-Flatt serial | cpu/tick | |---|---------|---------|------------|-------------------|----------| | 1 | 3.90 | 1.00x | 1.00 | — | 0.257 | | 2 | 7.78 | 2.00x | 1.00 | 0.0022 | 0.256 | | 4 | 15.53 | 3.98x | 1.00 | 0.0014 | 0.257 | | 8 | 30.91 | 7.93x | 0.99 | 0.0013 | 0.258 | | 16 | 61.26 | 15.71x | 0.98 | 0.0012 | 0.259 | The cost study measured 15.7x at 16 with serial fraction 0.0013 using N independent processes and NO `zid % N` filter — the ceiling, not an implementation. The shipped filter reaches it: 15.71x, 0.0012. cpu/tick is flat to 1.01x across the sweep, so sharding adds no measurable overhead. A second workload shape (`vw`, 69x125 vs biodiversity's 536x314) gives 15.52x at N=16, efficiency 0.97 — which retires §8's caveat that the exponents rested on ONE cell. ## DEPLOYMENT PREREQUISITE: pin BLAS/OpenMP threads per shard This is not an optimisation, it is a precondition, and the control arm measures why. Identical sweep with the threads UNPINNED — what production runs today; nothing in any Dockerfile, compose file or env file sets them: - N=1: 54.57s wall, cpu/tick 1.977 (pinned: 49.25s, 0.257) - N=2: 585.70s wall — 11x SLOWER than unpinned N=1 - N=16: 491.39s wall, 0.11x speedup, cpu/tick 40.9, load average 151.8 Pinned N=16 finishes the same work in 3.13s, so unpinned sharding is ~157x slower. Adding shards without pinning actively destroys throughput, because each shard fans every matmul across all 16 cores and they thrash. Note also that unpinning does not even pay for itself at N=1: it is 1.11x SLOWER for 7.7x the CPU (parallel efficiency ~0.117, reproducing the handoff's stated 0.11). §0's prose describes a "2.06x wall speedup", but 0.11 efficiency over 8.75 cores implies ~0.96x — no gain — and its own table shows 0.490s unpinned vs 0.238s pinned. The 2.06 is a slowdown ratio; that sentence reads inverted. Not included, deliberately: deployment wiring. There is no production entrypoint yet — `PollerConfig.from_env()` is called only from tests, with no `__main__` and no container command — so `cdk/autoscaling.ts` and the `after_install.sh` container caps have nothing to attach to. When that lands, `OMP_NUM_THREADS` / `OPENBLAS_NUM_THREADS` / MKL / VECLIB / NUMEXPR must all be set to 1 per shard container. ## Scaling tracks PHYSICAL cores, not vCPUs On c7i.8xlarge (32 vCPU = 16 cores x 2 SMT threads) the sweep is near-linear to N=16 (14.55x) and then plateaus at N=32 (15.66x, efficiency 0.49) with cpu/tick doubling to 1.98x — the signature of hyperthread contention, not a sharding limit. Size a shard fleet by physical cores or it over-provisions 2x on SMT hardware. ## Deliberately NOT done - Pushing `zid % N` into SQL. A modulo predicate typically cannot use an index, so it would trade cheap client-side filtering for a server-side scan. Client-side keeps the change inside `should_process_zid`, matching the allowlist/blocklist pattern already there. Each shard does poll all rows and discard ~(N-1)/N of them — measure that before optimising it. - Note `zid % N` balances COUNT, not COST: per-tick cost scales roughly with p_in_conv x c (in-conversation participants times comments), and that distribution is extremely skewed — of 15,575 prodclone conversations with votes, the median has 2 in-conv participants, the largest 23,354. Modulo is still the right first implementation — stateless, no coordination — but expect uneven utilisation and do not size capacity off the mean. The benchmark above deliberately uses a COST-balanced workload, so it measures the mechanism and not that skew. ## Tests (TDD, RED first): 18 new - Unit: default unsharded is identical to today for every zid; the partition is total and disjoint (every zid accepted by exactly one index, over ranges including `zid % N == 0`); shard beats allowlist; blocklist still excludes in-slice; `from_env` rejects index >= count, negative index, and count < 1; both env aliases and their precedence. - Integration: two service instances at indices 0 and 1 over the same polled rows dispatch disjoint sets whose union is the unsharded set; nothing dispatched twice across a 3-shard fleet; each shard still advances its own watermark; moderation sharded too. Golden snapshots untouched — sharding changes scheduling, never results. ## The benchmark that produced these numbers Also adds the shard-scaling benchmark: `polismath/replay/shard_bench.py` (library) + `scripts/shard_scaling_bench.py` (CLI), same split as `certify.py` / `poller_equiv.py`. Load-bearing design choices: a COST-balanced workload; BLAS pinned per shard with an explicit unpinned control arm so the §0 correction is measured rather than assumed; process startup excluded from the timed window via a ready/go barrier, so the arms' compute phases actually overlap; and wall time taken as the slowest shard, never the sum. Child stdout/stderr go to FILES, never pipes — a subtle but decisive point. `conversation.py` logs several KB per tick; piping that into a buffer the parent drains only at the end blocks each child once 64KB fills, and since the parent collects shard-by-shard, shard 0 runs at full speed while the rest sit blocked awaiting their turn. That serialises the arm: it reported 1.05x at N=2 on an IDLE 16-core box, with cpu/tick dead flat. A regression test spawns children that each flood 300KB to stderr and asserts neither stream is a pipe. The benchmark reports the Karp-Flatt serial fraction (directly comparable to the handoff's 0.9884 / 0.0013) and CPU-per-tick. The latter is what makes a disappointing speedup interpretable: flat cpu/tick as N grows means each shard does the same work and the ceiling is core availability (a property of the BOX), while rising cpu/tick means the shards interfere (a property of the MECHANISM) — it is what exposed both the pipe bug and the SMT plateau. A load-average guard warns when the machine cannot give each shard a core, so a sweep taken on a busy box cannot be misread as a scaling limit. 34 unit tests over every pure decision point (partition totality against the real filter, pinning env, best-of-K repeat selection, Karp-Flatt, verdict, load guard, no-pipes). The live sweep is opt-in tooling, never part of the pytest suite. Full delphi suite: 1187 passed, 22 skipped, 46 xfailed, zero failures. Co-Authored-By: Claude Opus 5 (1M context) commit-id:88ae2a13 --- delphi/polismath/poller/service.py | 89 +++- delphi/polismath/replay/shard_bench.py | 498 ++++++++++++++++++ delphi/scripts/shard_scaling_bench.py | 142 +++++ delphi/tests/poller/test_service.py | 71 +++ delphi/tests/poller/test_watermark.py | 115 ++++ .../tests/replay_harness/test_shard_bench.py | 331 ++++++++++++ 6 files changed, 1241 insertions(+), 5 deletions(-) create mode 100644 delphi/polismath/replay/shard_bench.py create mode 100644 delphi/scripts/shard_scaling_bench.py create mode 100644 delphi/tests/replay_harness/test_shard_bench.py 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)