feat(delphi): zid-sharding for the math poller — one shard = one process by jucor · Pull Request #2658 · compdemocracy/polis

jucor · 2026-07-25T15:04:28Z

The poller's parallelism mechanism does not work. ConversationWorkerPool
uses 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: 0.0013 [0.0009, 0.0014], 26.09 ticks/s at 16
(15.7x). Free-threading is not the escape hatch and was measured, not
assumed: on 3.14t (GIL confirmed off) the same threaded code gives 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 from the cost-model study (r8g.4xlarge native,
free-threaded arm on laptop); see HANDOFF_PYTHON_SHARDING.md.

This adds the scheduling scaffolding only — no math, blob or write-path
change whatsoever:

should_process_zid() takes shard_index/shard_count, 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.
Startup log names the slice ("shard=1/4" or "shard=unsharded").

The shard test runs BEFORE the allow/blocklist, and that ordering is a
correctness property rather than 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 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 rather than a laptop. Sweeps on an idle r8g.4xlarge (16
Graviton4 cores, no SMT), best-of-3 repeats, BLAS pinned:

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.

Also 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, 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.

Also adds the shard-scaling benchmark that produced the numbers above:
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 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.

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) noreply@anthropic.com

commit-id:88ae2a13

Stack:

⚠️ Part of a stack created by spr. Do not merge manually using the UI - doing so may have unexpected results.

Copilot

Pull request overview

This PR adds opt-in zid-based sharding to the Delphi math poller so multiple processes can safely partition conversation work via zid % shard_count, and introduces a benchmarking harness + CLI to measure shard scaling (including BLAS/OpenMP thread pinning vs unpinned control).

Changes:

Add shard_index / shard_count to poller configuration and route them through both vote + moderation polling dispatch.
Extend should_process_zid() with sharding semantics (shard filter evaluated before allow/block lists) plus config validation and startup logging of the shard slice.
Add shard-scaling benchmark library + CLI and comprehensive unit coverage for the benchmark’s pure decision points.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
delphi/polismath/poller/service.py	Adds shard-aware dispatch, env-driven shard config, validation, and startup logging.
delphi/polismath/replay/shard_bench.py	Implements the multi-process shard scaling benchmark harness and its reporting/metrics.
delphi/scripts/shard_scaling_bench.py	Click CLI wrapper to run pinned/unpinned sweeps and render results.
delphi/tests/replay_harness/test_shard_bench.py	Unit tests covering benchmark partitioning, BLAS env handling, scaling math, and “no pipes” regression.
delphi/tests/poller/test_watermark.py	Adds sharding behavior + shard config validation tests for poller scheduling.
delphi/tests/poller/test_service.py	Adds integration-style tests ensuring sharded dispatch is total/disjoint and watermarks still advance.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+    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")


The poller's parallelism mechanism does not work. ConversationWorkerPool uses 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: 0.0013 [0.0009, 0.0014], 26.09 ticks/s at 16 (15.7x). Free-threading is not the escape hatch and was measured, not assumed: on 3.14t (GIL confirmed off) the same threaded code gives 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 from the cost-model study (r8g.4xlarge native, free-threaded arm on laptop); see HANDOFF_PYTHON_SHARDING.md. This adds the scheduling scaffolding only — no math, blob or write-path change whatsoever: - should_process_zid() takes shard_index/shard_count, 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. - Startup log names the slice ("shard=1/4" or "shard=unsharded"). The shard test runs BEFORE the allow/blocklist, and that ordering is a correctness property rather than 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 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 rather than a laptop. Sweeps on an idle r8g.4xlarge (16 Graviton4 cores, no SMT), best-of-3 repeats, BLAS pinned: 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. Also 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, 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. Also adds the shard-scaling benchmark that produced the numbers above: 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 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. 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) <noreply@anthropic.com> commit-id:88ae2a13

jucor mentioned this pull request Jul 25, 2026

feat(math): un-mirror priority_metric (#2571 resolved) + Q2 prev-tick group-votes #2645

Draft

jucor force-pushed the spr/edge/88ae2a13 branch 4 times, most recently from a8c16f1 to 7388d8b Compare July 25, 2026 22:10

jucor requested a review from Copilot July 25, 2026 22:11

Copilot started reviewing on behalf of jucor July 25, 2026 22:11 View session

Copilot AI reviewed Jul 25, 2026

View reviewed changes

jucor force-pushed the spr/edge/88ae2a13 branch from 7388d8b to 89611cf Compare July 25, 2026 22:24

jucor mentioned this pull request Jul 25, 2026

docs(delphi): overnight pre-cutover audit (s6) — CUTOVER_RUNBOOK + journal #2659

Draft

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(delphi): zid-sharding for the math poller — one shard = one process#2658

feat(delphi): zid-sharding for the math poller — one shard = one process#2658
jucor wants to merge 1 commit into
spr/edge/62486a46from
spr/edge/88ae2a13

jucor commented Jul 25, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

jucor commented Jul 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jucor commented Jul 25, 2026 •

edited

Loading