python-math #32: 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

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) 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.

… shard = one process ## 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) <noreply@anthropic.com> commit-id:88ae2a13

jucor force-pushed the spr/edge/62486a46 branch from 07b4090 to 767b2a1 Compare July 28, 2026 00:11

jucor force-pushed the spr/edge/88ae2a13 branch from 89611cf to 2d10ae0 Compare July 28, 2026 00:11

jucor changed the title ~~feat(delphi): zid-sharding for the math poller — one shard = one process~~ python-math #32: feat(delphi): zid-sharding for the math poller — one shard = one process Jul 28, 2026

jucor mentioned this pull request Jul 28, 2026

Clojure->Python CUTOVER: Step 4 — remove the Clojure tree; move the Python math engine to math/ #2689

Draft

3 tasks

jucor force-pushed the spr/edge/62486a46 branch from 767b2a1 to 78adde7 Compare July 28, 2026 01:10

jucor force-pushed the spr/edge/88ae2a13 branch from 2d10ae0 to 2762384 Compare July 28, 2026 01:10

Navigation Menu

Uh oh!

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

python-math #32: 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!

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!

Why: the poller's current parallelism mechanism does not work

What ships: opt-in sharding scaffolding only

The shard filter runs BEFORE the allow/blocklist — a correctness property

Watermarks: verified, no change needed

MATH_WORKER_POOL_SIZE stays at 4 — a deliberate, recorded decision

MEASURED: the §7 throughput criterion is now verified on prod-like hardware

DEPLOYMENT PREREQUISITE: pin BLAS/OpenMP threads per shard

Scaling tracks PHYSICAL cores, not vCPUs

Deliberately NOT done

Tests (TDD, RED first): 18 new

The benchmark that produced these numbers

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jucor commented Jul 25, 2026 •

edited

Loading