diff --git a/app/config.py b/app/config.py index 1c5322c..f7a7641 100644 --- a/app/config.py +++ b/app/config.py @@ -93,12 +93,16 @@ class Settings(BaseSettings): # in-process memory rather than a DB lock; exceeding it fails loudly. # # Sized against the ACTUAL buffer cost, not a round number: the vectors are held as - # Python float lists, ~32 B per element (24 B float object + 8 B list pointer), so at - # dim=1024 each chunk costs ~32 KB -- 8000 chunks is ~260 MB of vectors plus ~16 MB of - # chunk text. A larger ceiling (e.g. 50k -> ~1.6 GB) would OOM the job container before - # this loud check could ever fire, which would defeat the point of having a ceiling. - # A repo that legitimately exceeds this needs a temp-table staging path, not a bigger - # buffer. + # Python float lists, ~32 B per element (24 B float object + 8 B list pointer) structural, + # but ~40.1 KB/chunk RESIDENT once measured (issue #109; pymalloc overhead/fragmentation -- + # use this figure for headroom arithmetic) -- 8000 chunks is ~313 MiB of vectors resident + # plus ~16 MB of chunk text. #109 also derived a per-worker chunk-cap ceiling from a full + # container-memory model (~73,300 chunks at the pinned N=2 semantic-worker count this cap + # is evaluated at, ~36,700 at the shipped N=4 -- see docs/perf/issue-109-measurements.md + # §12): a larger ceiling well past that (e.g. 50k, ~1.9 GiB resident) risks OOMing the job + # container before this loud check could ever fire, which would defeat the point of having + # a ceiling. A repo that legitimately exceeds this needs a temp-table staging path, not a + # bigger buffer. # # Scope note (#104): under file-level delta indexing this cap is enforced against # whatever ONE RUN embeds (changed/new + membership-only files), not a branch's whole @@ -112,11 +116,12 @@ class Settings(BaseSettings): # which bounds file ingestion, not embedding-chunk granularity. semantic_chunk_max_tokens: int = 512 - # In-flight embedding requests per worker (#107). The indexer clamps to 2 workers when - # semantic is on (indexer/repo_config.py:effective_workers), so total in-flight gateway - # requests are workers x concurrency: 2 x 4 = 8 at this default, 2 x 8 = 16 at the - # config.yaml-enforced ceiling of 8 -- both under the SDK's 20-connection pool - # (pool_block=True, so exceeding it would silently serialize rather than error). Setting + # In-flight embedding requests per worker (#107). The indexer clamps to 4 workers when + # semantic is on (indexer/repo_config.py:effective_workers -- issue #109 raised this from + # 2), so total in-flight gateway requests are workers x concurrency: 4 x 4 = 16 at this + # default, 4 x 8 = 32 at the config.yaml-enforced ceiling of 8 -- the latter now EXCEEDS + # the SDK's 20-connection pool (pool_block=True, so exceeding it silently serializes + # rather than erroring, so this is a real-concurrency cap, not a correctness one). Setting # this to 1 restores today's fully serial embed() and spawns no thread pool. semantic_embedding_concurrency: int = 4 diff --git a/config.yaml b/config.yaml index 28eff07..36fd6e5 100644 --- a/config.yaml +++ b/config.yaml @@ -12,8 +12,13 @@ version: 1 # threads — the tree walk is GIL-serialized), which is why extraction now runs # in its own shared process pool instead — see extract_processes below. Raising # this knob buys disk-bound repo fan-out, not extraction throughput. -# With semantic indexing enabled this is clamped to 2 — a MEMORY bound, since -# embedding materialises a whole repo's chunks (~0.5-0.8 GB per worker). +# With semantic indexing enabled this is clamped to 4 (issue #109 raised it from +# 2, after re-deriving the memory model and confirming empirically against the +# live dev job at N=4: peak self+children RSS landed at ~83% of the 0.7*container +# memory budget, comfortably under and with more margin than N=2's own ~90%) — a +# MEMORY bound, since embedding materialises a whole repo's chunks (structural +# ~32 KB/chunk, resident ~40.1 KB/chunk measured; see effective_workers' +# docstring and docs/perf/issue-109-measurements.md for the full derivation). # index_concurrency: 4 # How many worker PROCESSES the job uses to extract symbols/edges (issue #108). @@ -75,13 +80,23 @@ connections: # effective cap as `per-repo override OR global`. # # Mind the memory math before raising one: buffered vectors are ~32 KB/chunk -# (dim=1024, Python float-list storage), so 8000 ≈ 260 MB resident for the -# duration of that repo's write. With semantic on, at most 2 workers run -# concurrently (indexer/repo_config.py's effective_workers clamp), so a large -# override multiplies straight into the job container's peak memory — e.g. two -# repos overridden to 20000 concurrently is ≈1 GB just in vectors, on top of -# the base per-worker cost. A repo that legitimately needs far more than that -# needs the temp-table staging path (follow-up), not a bigger override. +# structural (dim=1024, Python float-list storage) but ~40.1 KB/chunk RESIDENT +# (measured, issue #109 — includes pymalloc overhead/fragmentation; use this +# figure for headroom arithmetic), so 8000 ≈ 313 MiB resident for the duration of +# that repo's write. With semantic on, at most 4 workers run concurrently +# (indexer/repo_config.py's effective_workers clamp, raised from 2 by #109), so +# a large override multiplies straight into the job container's peak memory — +# e.g. two repos overridden to 20000 concurrently is ≈1.6 GB just in vectors, on +# top of the base per-worker cost. A repo that legitimately needs far more than +# that needs the temp-table staging path (follow-up), not a bigger override. +# +# The derived per-worker chunk-cap ceiling, from a full container-memory model +# (issue #109; docs/perf/issue-109-measurements.md §12 — pinned at N=2 there +# only to break a circularity in solving for C from a formula whose dominant +# term IS C, not a claim about the adopted concurrency): ≈73,300 chunks at +# N=2, ≈36,700 at the shipped N=4 (both halve/double with N). The current +# global default of 8000 uses well under a quarter of either budget, so it is +# NOT the binding constraint and was left unchanged. # semantic_max_chunks_per_repo: # "acme/huge-monorepo": 20000 @@ -98,20 +113,24 @@ connections: # (the default is 8000). It is NOT the per-repo `semantic_max_chunks_per_repo` MAP # above — that spot-overrides individual repos and still wins over this global. Use # this to raise the floor everyone inherits; use the map for the outliers. The same -# ~32 KB/chunk memory math and 2-worker clamp above apply here, magnified: raising -# the global lifts the buffer cost for EVERY concurrently-indexing repo at once. +# ~32 KB structural / ~40.1 KB resident per-chunk memory math and 4-worker clamp +# above apply here, magnified: raising the global lifts the buffer cost for EVERY +# concurrently-indexing repo at once. # # `enabled: false` makes the job a true semantic no-op (no embedder built, no -# chunking, the 2-worker memory clamp not applied) even if the env says enabled — +# chunking, the 4-worker memory clamp not applied) even if the env says enabled — # the fastest way to turn semantic off for the job alone. # # `embedding_concurrency` (#107) is in-flight embedding requests PER WORKER, sent # via a ThreadPoolExecutor that preserves submission order — vectors always come # back in the order their texts were sent, regardless of which request finishes # first. Total in-flight gateway requests for the job is workers x concurrency: -# 2 x 4 = 8 at this default, 2 x 8 = 16 at the max of 8, both under the SDK's -# 20-connection pool. Set to 1 to restore fully serial embedding (no thread pool -# spawned at all) if you need to roll back. +# 4 x 4 = 16 at this default, 4 x 8 = 32 at the max of 8 — the latter now EXCEEDS +# the SDK's 20-connection pool (issue #109 raised workers from 2 to 4; this +# combination was not possible before). Lower embedding_concurrency if raising it +# alongside a near-ceiling index_concurrency. Set embedding_concurrency to 1 to +# restore fully serial embedding (no thread pool spawned at all) if you need to +# roll back. # semantic: # enabled: true # max_chunks_per_repo: 8000 diff --git a/docs/perf/issue-109-measurements.md b/docs/perf/issue-109-measurements.md new file mode 100644 index 0000000..edc2caf --- /dev/null +++ b/docs/perf/issue-109-measurements.md @@ -0,0 +1,483 @@ +# Issue #109 — re-derive worker, disk, and memory limits: measurements + +**Unit convention, stated up front (mixing anchors is a real risk in this doc):** +every RSS/vector/memory-model figure below is **binary** (KiB/MiB = 1024-based), +matching `resource.ru_maxrss` (Linux: KB = 1024 bytes) and this repo's existing +`docs/perf/issue-108-measurements.md` convention. The one exception is +`MAX_EXTRACTED_BYTES = 2_000_000_000`, which is **decimal** (2 GB = 2,000,000,000 +bytes) by the source constant's own definition — any comparison against it is +converted explicitly, never left implicit. + +--- + +## 1. Environment + +**Local box** (measurements in §2–§4): Linux, 12 cores, ~15.5 GB RAM, Python +3.12.13 (uv-managed `.venv`; the shell's own default is 3.14). `/tmp` is tmpfs +(7.8 GB) — disk-backed measurements used `/` (nvme0n1p2, 318 GB free) instead. + +**Databricks dev serverless container** (§5, `M` and W4): read directly from +inside the deployed job via a temporary probe (job schedule stayed **PAUSED** +throughout; only manually-triggered one-time runs executed). Two separate +container instances were observed across two probe attempts, with a **>2x +spread** in reported memory — see §5.1. + +--- + +## 2. AC2 — disk (E1): already correct on the base, verified not re-derived + +Per the plan's §0.1, #106 already landed the disk half of #109. Verified by +citation, not rewritten: + +- `indexer/fetch.py`: `REQUIRED_FREE_BYTES == MAX_TARBALL_BYTES == 500_000_000`. +- The guard message carries the real number (`need 500000000 ...`). +- `docs/runbooks/indexing-parallelism.md` §3 and `config.yaml` already read 0.5 + GB/worker (1→0.5, 2→1, 4→2, 8→4 GB). +- The tarball is the ONLY on-disk artifact (`indexer.ingest.iter_tar_source_files` + streams in memory, never extracts) — confirmed by direct code reading, not a + fresh sampling run (§0.1 forbids "restating a correct 8-worker figure" as new + work). +- W3 (observed on the live dev job, both arms): `local disk at /tmp: 64.0 GB + free of 89.1 GB total` — **disk is not a binding constraint at any allowed + `index_concurrency` (up to 8, 4 GB peak)**. + +**AC2: satisfied, unchanged.** + +--- + +## 3. M1 — the memory model's coefficients (E3(b–f)) + +`scripts/measure_semantic_memory.py`, run against 4 corpora (this repo, +`flask`, `requests`, `django`), driving the REAL semantic path +(`iter_tar_source_files` → delta narrowing → `_precompute_chunk_writer`) with a +stub embedder returning **distinct** floats per chunk (per §2.2's trap — a +shared-cached-float stub understates resident vector cost ~4x). + +| Corpus | alpha | gamma | d (bytes/chunk) | +|---|---|---|---| +| databricks-code-search | 2.1142 | 0.5571 | 2711.6 | +| flask | 2.044 | 0.4693 | 1644.2 | +| requests | 3.382 | 0.0 (see below) | 3747.7 | +| django | 1.5911 | 1.2627 | 1775.5 | + +`requests`' gamma measured as 0 — a chunk delta small enough (chunk_count=413) +to fall below this box's RSS sampling granularity (allocator/page-granularity +noise), not evidence chunking is free for that corpus. + +alpha: avg=2.2828, **max=3.382** (n=4). gamma: avg=0.5723, **max=1.2627** (n=4). +**Decision: use MAX-observed coefficients** (larger → smaller/safer derived +limits, larger/stricter `P_worst`) as the primary, conservative input; average +reported alongside for context (§8 judgement call, self-consistently applied +everywhere it's used). **`alpha + gamma = 4.6447`** (max of each, not the max +corpus's sum — the model treats them as independently-conservative). + +`V_cap` (resident vector cost, 8000 × 1024 distinct-float vectors): **40.102 +KB/chunk** (this session's re-measurement; close to planning's 40.8 KB/chunk — +both measure the same thing, structural is 32.0 KB/chunk exact +(`1024 × (8B pointer + 24B PyFloat)`), not "corrected" by this re-measurement, +per §2.2's own instruction not to). + +`R_proc` = 121 MB/process (#108's own isolated measurement — reused, not +re-run, per the plan's L4). + +`P_fixed`: a bare-interpreter floor probe gave 38,372 KB → +`import indexer.job` +(pulls SQLAlchemy/databricks-sdk) → 63,048 KB → +a throwaway SQLAlchemy engine → +67,628 KB. This is a **floor** (~68 MB) — it excludes the real pool_size-scaled +connection pool and Databricks SDK client state a live job carries. **Adopted +P_fixed = 300 MB** (the plan's own conservative worked-example value), noting +the ~68 MB floor as a cross-check, not a replacement. + +### 3.1 A methodology finding not anticipated by the plan: fork-time COW RSS contamination of `RUSAGE_CHILDREN` + +M1's stage-3 (N-concurrency) sweep reported `after_children_kb` **numerically +identical** to `after_self_kb` at every N (e.g. N=1: self=1,046,984, +children=1,046,984). Root-caused directly (scratch scripts, not committed): +when the extraction pool's `spawn` workers are forked/exec'd **after** a +repo-worker's `_precompute_chunk_writer` has already ballooned the calling +thread's RSS to ~1 GB — exactly production's real call order in +`indexer/job.py` — each child's `ru_maxrss` (read later via `wait()` / +`RUSAGE_CHILDREN`) captures the **fork-time COW snapshot of the parent's +then-current RSS**, not the child's real post-exec working set. Verified +directly: draining the pool BEFORE ballooning gives children ~144–150 MB +(matching #108's own R_proc ~121 MB order of magnitude); draining AFTER gives +children ≈ self's contemporaneous value — a ~7–9x inflation with **no +corresponding real memory pressure** (COW pages are shared, billed once by the +cgroup, not per-process). + +**Implication:** this affects M1's own stage-3 "children" column, and by the +same mechanism, the production `peak rss: self=... children=...` instrumentation +on Arms A/B (`indexer/job.py`'s new n5 log line) whenever the pool is +(re)spawned after chunk-writer inflation. It is a `ru_maxrss` **measurement +artifact**, not evidence of doubled real memory. The `P_worst` model itself is +unaffected in its primary form because `R_proc` is sourced from #108's own +isolated measurement, not from this contaminated figure — but Arm A/B's +observed `children=` numbers below should be read as **upper bounds, not +literal per-process costs**. + +Stage-3 self-deltas (uncontaminated — `self` reflects only the process's own +allocations) at N=1..4, worst-case corpus (django, first-index/gate-closed): +N=1: 983,036 KB; N=2: 1,828,428 KB; N=3: 2,582,480 KB; N=4: 3,216,572 KB — +**sub-linear**, consistent with page-cache/tarball-decompression sharing across +threads, not a red flag. + +**Caveat on these specific numbers (found and fixed post-hoc in review, not +re-measured):** at measurement time, `scripts/measure_semantic_memory.py`'s +stage-3 worker discarded `_precompute_chunk_writer`'s return value before +calling `pool.stream()`, so each thread's vectors were collectable before (or +concurrently with) sibling threads' peaks — understating true N-way +concurrent residency relative to production, which holds `chunk_writer` alive +across the whole write window. The script is fixed in this PR (the return +value is now held alive across `pool.stream()` and explicitly `del`eted after, +matching stage 1's pattern) for future use, but the N=1..4 numbers above +**were not re-measured** against the fix, since **no decision in this PR rests +on them** — the adopted N=4 decision comes from the `P_worst` model (§6) and +the real Arm A/B live-job runs (§8–§10), not from this local sub-measurement. +Treat the sub-linearity finding above as directional, not load-bearing. + +--- + +## 4. B — the three anchors + +- **`B_prod`** (real corpus, unnesting `files.branches`): top row `repo_id=46` + (`IceRhymers/opencode`, branch `dev`), `src_bytes = 31,778,187` (~30.3 MiB). +- **`B_obs`** (measurement corpus, re-measured directly): `opencode@dev` = + 33,015,231 bytes decoded source (18,617 chunks, 4,759 files) — slightly + higher than `B_prod` (encoding/whitespace differences between the two + measurement paths). +- **`B* = max(B_prod, B_obs) = 33,015,231 bytes ≈ 31.49 MiB`** — the anchor used + for every decision below. +- **`B_ceil = MAX_EXTRACTED_BYTES = 2 GB`** — theoretical, loose, **never** used + to drive a decision (only to illustrate why a naive `B_ceil`-anchored model + would falsely condemn the status quo — see below). + +--- + +## 5. `M` — the container memory ceiling (E3(a)) + +Two container instances observed across two probe attempts on the same job: + +| Attempt | task_run_id | `cgroup_v1_memory.limit_in_bytes` | `sched_getaffinity` | Outcome | +|---|---|---|---|---| +| 1 | 523922694112794 | 8,385,462,272 (~7996 MiB) | 4 | **OOM-killed** during the allocate-bracket, last logged step `cumulative_mb=12800` | +| 2 | 89875228623550 | 24,706,547,712 (~23.0 GiB), `MemTotal` 32,264,556 kB | 4 | Ran its bracket to a designed 16,384 MB cap without dying (never exercised further) | + +**`cgroup_v2_memory.max`/`memory.high` both unreadable** (`FileNotFoundError`) +on this runtime — the repo's own `extract_pool.py::_cgroup_cpu_quota()` v2-only +assumption does not hold for `memory.max`; the v1 fallback +(`/sys/fs/cgroup/memory/memory.limit_in_bytes`) was required. + +**Attempt 1's cgroup read was demonstrably a MISREAD**: the container died at +`cumulative_mb=12800` (the bracket's last logged step before the kill), i.e. +the real ceiling sits in `(12800, ~13056] MiB` — **~60% higher** than the +cgroup-reported ~7996 MiB. + +**`M = 12800 MiB`** (the smaller, real, empirically-grounded ceiling from +attempt 1 — chosen conservatively over attempt 2's larger, undead container, +per the resumption brief's instruction to use the smaller real ceiling for +safety). **`0.7 × M = 8960 MiB`** — the budget used throughout. + +**`W4` (container CPU count) = 4** (`len(os.sched_getaffinity(0))`, both +attempts agree) — sourced from this probe, per the plan's design, breaking the +apparent circularity between E4 (needs W4) and E5 (Arm B, which would +otherwise be W4's only source). + +### 5.1 First-order finding: container sizing is unstable across attempts + +A **≥2x spread** in effective memory ceiling was observed between two +instances of the *same* job on *unspecified* dev serverless compute (~8 GiB vs. +~23.0 GiB reported; real ceilings both plausibly larger than reported). This is +reported as a finding, not resolved — the smaller, conservative number is what +every downstream decision uses. + +--- + +## 6. `P_worst` — the model, evaluated at the standard cap + +``` +P_worst(N) = N × max( (alpha+gamma)·B_breach, + (alpha+gamma)·(d×C) + V_cap·C ) + + extract_processes × R_proc + P_fixed +``` + +At the **standard/default global chunk cap `C = 8000`** (ordinary unmodified +production, NOT the measurement corpus's per-repo overrides — see §7 for why +those diverge), with the MAX coefficients above (`alpha+gamma = 4.6447`, +`d = 3747.7`, `V_cap = 40.102 KB/chunk`, `R_proc = 121 MB`, +`extract_processes = 4` — confirmed live in the priming log's "symbol +extraction: 4 process(es)"), `P_fixed = 300 MB`: + +- Breach term at `B* = 31.49 MiB`: `(a+g)·B* ≈ 146.3 MiB`/worker. +- Under-cap term at `C = 8000` (the regime that wins under the standard cap — + vectors dominate the uncapped terms ~8x at the cap, per the plan's §2.3): + `(a+g)·d·C ≈ 132.8 MiB` + `V_cap·C ≈ 313.3 MiB` = **446.1 MiB**/worker. +- `max(146.3, 446.1) = 446.1 MiB`/worker. + +`P_worst(3, B*) = 3×446.1 + 4×121 + 300 = 2122.3 MB` +`P_worst(4, B*) = 4×446.1 + 4×121 + 300 = 2568.4 MB` + +Both `<< 0.7M = 8960 MiB` (margin ~76% at N=3, ~71% at N=4). Cross-checked with +AVERAGE coefficients (`alpha+gamma = 2.855`, avg `d = 2469.75`): +`P_worst(4) = 2252.4 MB` — same conclusion; **not sensitive to the max-vs-avg +coefficient choice.** + +**Illustrative-only counter-example (never used to decide anything): at +`B_ceil = 2 GB` (= 1907.3 MiB binary)** the breach term is +`(alpha+gamma)·B_ceil = 4.6447 × 1907.3 ≈ 8859.1 MiB`/worker — the breach +regime wins by **≈19.9x** over the under-cap term (446.1 MiB, §6), giving +`P_worst(2, B_ceil) = 2×8859.1 + 484 + 300 ≈ 18,502 MiB` — i.e. the model +would falsely condemn TODAY's clamp=2 status quo many times over if evaluated +at the loose theoretical bound instead of the real observed `B*`. This is +exactly why §3.3(c) of the plan anchors the decision on `B*`, never `B_ceil`. +(The planning-stage worked example in the plan document itself used its own +pre-`M1` coefficients, ~2.35 rather than the measured 4.6447, and got a +smaller — still condemning — ~10.2 GiB; both versions support the same +qualitative point, so both numbers are recorded here for provenance: +whichever coefficient set is used, `B_ceil` is not a fit substitute for `B*`.) + +**Decision (model-only, pre-Arm-A): branch (i), RAISE — both N=3 and N=4 clear +`0.7M` with large margin. Adopt N=4 (the largest passing N), contingent on Arm +B completing cleanly** — modelling alone is not sufficient per the plan's AND +condition. + +--- + +## 7. Re-evaluation against Arm A's own observed peak (§3.3(c).iii) + +The standard-cap model above does **not** reflect what Arm A actually ran, +because the measurement corpus's per-repo chunk-cap overrides (opencode = +22,340, ≈2.8x the standard 8000 cap — see §9) make this specific corpus far +more memory-expensive than "standard production": Arm A's real self+children +was ~4.75x higher than the standard-cap model's own N=2 prediction (1676 MiB), +because the override intentionally lets opencode use ~874.9 MiB of vectors +instead of breaching into the cheap ~146 MiB breach regime. Expected (flagged +in the plan's §3.2 as a likely consequence), not a bug — but it means only +Arm A's own empirical number, not the standard-cap model, can gauge Arm B's +real risk on this corpus. + +--- + +## 8. Arm A — before (clamp=2, TODAY's shipped code) + +Both runs: 22 repos resolved (19 retained + 3 explicit), `symbol extraction: 4 +process(es) (spawn); pool preflight ok`, `local disk ...; 2 worker(s) x 0.5 GB +peak`, `semantic enabled: clamping index_concurrency 4 -> 2`, 22/22 branches ok, +0 skipped/conflicts/failed, 0 repos purged, no `degraded semantic coverage` +WARNING, no 429/retry lines. + +| | Run 1 (`489590218152767`) | Run 2 (`19039699744754`) | +|---|---|---| +| wall (execution_duration) | 350.5s | 342.4s | +| `peak rss: self=... children=...` | self=6,930,768 KB children=1,247,176 KB | self=7,059,172 KB children=1,252,284 KB | +| self+children | 8,177,944 KB ≈ **7986.3 MiB** | 8,311,456 KB ≈ **8116.7 MiB** | +| % of `0.7M` (8960 MiB) budget | 89.1% | 90.6% | + +The two runs agree within 1.6 percentage points — **not a fluke: at TODAY's +clamp=2, this corpus already consumes ~89–91% of the safety budget.** Best-of +(faster wall): run 2, 342.4s. Worse-of (higher memory, used as the conservative +anchor going forward): run 2, 8116.7 MiB. + +**§3.3(c).iii re-evaluation:** this real number is ~4.75x the standard-cap +model's prediction (§7) — the standard-cap model cannot gauge Arm B's risk on +this corpus. Direct reasoning from Arm A's own peak: this 22-repo corpus has +exactly 3 memory-heavy repos (opencode ~874.9 MiB vectors, claw-code ~147.9 MiB, +nanoclaw ~30 MiB), and overlap is capped by there being only 3 of them +regardless of N — so N=3/N=4 were estimated to land close to Arm A's own +number (perhaps +50–150 MiB), i.e. plausibly under budget but with a **thinner +margin (~85–90% utilized)** than the standard-cap model implied. **Decision: +still target N=4** for the single Arm B attempt (the largest of {3,4}, and +nothing in the refined reasoning favors N=3 specifically), with full awareness +of the thinner real margin and a non-negligible chance of failure — a valid, +reportable outcome either way per §6, not a trigger to retry at a different N. + +--- + +## 9. Arm B — after (clamp=4, the derived change) + +Applied the one-line change (`indexer/repo_config.py::effective_workers`: +`min(index_concurrency, 2)` → `min(index_concurrency, 4)`), redeployed +(`databricks bundle deploy -t dev`), cleared stamps, warmed Lakebase (>300s +`SELECT 1` loop immediately before each timed run — see §10 for a warm-up +reliability note), ran twice. + +Both runs: 22/22 branches ok, 0 skipped/conflicts/failed, 0 repos purged, +**no** `semantic enabled: clamping ...` line (correct: `index_concurrency=4` +now equals the clamp, so `effective_workers` is a no-op passthrough — matching +§11's conclusion that the `index_concurrency` default itself did not need to +move), `local disk ...; 4 worker(s) x 0.5 GB peak`, `symbol extraction: 4 +process(es) (spawn); pool preflight ok`, **zero** WARNING/ERROR/retry/429 lines +anywhere in either 191-line log. + +| | Run 1 (`752462522914821`) | Run 2 (`785234477657138`) | +|---|---|---| +| wall (execution_duration) | 307.8s | 305.5s | +| `peak rss: self=... children=...` | self=6,128,880 KB children=1,445,408 KB | self=6,225,624 KB children=1,429,280 KB | +| self+children | 7,574,288 KB ≈ **7396.8 MiB** | 7,654,904 KB ≈ **7475.5 MiB** | +| % of `0.7M` (8960 MiB) budget | 82.6% | 83.4% | + +Both runs agree within 1 percentage point — not a fluke. Best-of (faster +wall): run 2, 305.5s. + +**Surprising but real: Arm B's total peak is LOWER than both Arm A runs, and +its margin (~17%) is MORE comfortable than Arm A's own (~9–11%).** `ru_maxrss` +is a same-process high-water mark, so this is a genuine peak-memory +observation, not a modelling artifact — Arm B's own `self` component (6.13–6.23 +million KB) is genuinely lower than either Arm A `self` value (6.93–7.06 +million KB). Consistent with (not contradicting) §8's reasoning: this corpus +has only 3 memory-heavy repos, and with 4 workers instead of 2 their embedding +windows are *less* likely to bunch up at the tail (more workers drain the +22-repo queue faster and spread the 3 heavy repos across more concurrent slots +with shorter individual overlap) — a property specific to this corpus's +repo-size distribution, **not** a general "N=4 always costs less than N=2" +claim. + +**No stop condition (§6 of the plan) was triggered anywhere in the Arm B +sequence.** + +## 10. AC1 — before/after table + +| | Arm A (clamp=2, best-of wall) | Arm B (clamp=4, best-of wall) | +|---|---|---| +| wall | 342.4s | 305.5s (**−10.8%**) | +| peak self+children | 8116.7 MiB (90.6% of budget) | 7475.5 MiB (83.4% of budget) | + +**FINAL DECISION (AC1 + AC3): adopt N=4.** Both the `P_worst` model (§6, before +any arm ran) and two clean, mutually-consistent empirical Arm B runs (§9) agree. + +**Lakebase CU state**: `resources/lakebase.yml` pins +`autoscaling_limit_min_cu: 0.5`, `autoscaling_limit_max_cu: 4`, +`suspend_timeout_duration: 300s`. Warmed with a `SELECT 1` loop for the full +300s+ immediately before every timed run (priming, Arm A ×2, Arm B ×2 — 5 +warm-ups total), confirmed complete each time via elapsed wall-clock (302.3s / +301.6s measured, not assumed). + +**Budget/429 posture**: 5 full semantic-index runs total (1 priming + 2 Arm A + +2 Arm B) against the real, paid AI Gateway embedding endpoint. **Zero** 429s or +`databricks.sdk.retries` entries observed in any run's log; zero `degraded +semantic coverage` WARNINGs. + +--- + +## 11. `index_concurrency` default (E4/M2 — the ingest-thread-scaling question) + +`scripts/measure_ingest_threads.py`, repos: fastapi, sqlalchemy, sympy, django. + +- N=2 idle: 1.26x (ambiguous band, ≥1.6 parallelizes / ≤1.1 doesn't). +- N=4 idle: **1.20x** (NO-SCALE — ≤1.2 threshold, exactly at the boundary). +- N=2 pool-live: 1.20x (ambiguous). +- N=4 pool-live: **1.15x** (NO-SCALE). +- Component breakdown confirms the GIL-bound prior (§0.4 of the plan): + `tf_next`/`fh_read` cumulative time grows super-linearly with N (contention); + `decode` stays roughly flat — consistent with a mostly GIL-held pass. + +**Per the plan's §3.3(d) rule 2** ("the default may rise to the smallest value +≥ N that M2 shows a gain for" once the clamp itself rises): M2 shows **no** +real gain at 4 threads. **`index_concurrency`'s default stays 4** — and no +change was even needed: it was already 4 (`config.yaml`'s commented-out +default), so raising the semantic clamp from 2 to 4 alone makes +`effective_workers = min(4, 4) = 4` with zero separate change to +`index_concurrency` itself. + +--- + +## 12. The two derived byte limits (§3.3(f)) — documented, neither changed + +Both solved at the **pinned N=2** operating point (methodological, to break the +circularity of solving for `B` from a model whose dominant term is `B` — not a +claim that N=2 is the adopted concurrency, which is 4 per §10). +`RHS = 0.7M − extract_processes·R_proc − P_fixed = 8960 − 484 − 300 = 8176 MiB`. + +**(f1) `MAX_EXTRACTED_BYTES` (breach regime, V=0):** +`B ≤ RHS / (N·(alpha+gamma)) = 8176 / (2×4.6447) ≈ 880.1 MiB ≈ 922.9 MB +(decimal)`. Current value: `2_000_000_000` bytes = 1907.3 MiB (decimal 2 GB). + +**(f2) The chunk cap `C` (under-cap regime):** +`C ≤ RHS / (N·((alpha+gamma)·d + V_cap))`. Denominator = +`4.6447×3747.7 + 41,064.4 B ≈ 58,471 B ≈ 57.1 KiB`/unit-of-cap. `C ≤ +8176×1024×1024 / (2×58,471) ≈ 73,311 chunks`. Current global default: 8000 +(**~11% of the derived bound** — nowhere near binding). + +**Neither value was changed.** `B* = 31.49 MiB` is only **3.6%** of the derived +`f1` bound (880.1 MiB) — no branch in the measurement corpus approaches either +the current 2 GB constant or the derived ~880 MiB one, so this run gives **no +empirical signal** to justify lowering a constant whose breach fails the branch +and closes the whole run's reconciliation checkpoint (`job.py`'s +`_decide_reconciliation` requires `failures == 0`). Per the plan's explicit +fallback ("otherwise document the derived value and keep 2 GB with a +comment"), both derived values are recorded here and in the source comments +(`indexer/ingest.py`, `config.yaml`) for the next time this needs re-deriving, +but the live constants are unchanged. The chunk cap `C=8000` also stays — +it uses only ~11% of its own derived budget, so it was never a candidate for +change either. + +**Which regime binds at `B*`:** the **breach** regime, at the standard cap. +`B*` (~31.49 MiB, opencode@dev's real source size) exceeds `d×C` at the +standard `C=8000` under either the conservative modelling `d` (3747.7 B/chunk +→ threshold ≈28.6 MiB) or opencode's own real chunking density (≈1773 +B/chunk → threshold ≈13.5 MiB) — i.e. **a repo the size of `B*` would actually +breach the standard 8000-chunk cap**, which is exactly why the measurement +corpus needed opencode's per-repo override (22,340, §9's context) to avoid +degrading it during Arm A/B. This makes `f1` (the breach-regime derived bound, +not `f2`) the relevant limit to check `B*` against — already done above: +`B*` is only 3.6% of `f1`'s ~880 MiB, comfortably clear. `f2`'s under-cap +regime governs a *different* class of repo: one that legitimately reaches the +cap without a bigger override (§6's worked "vectors dominate ~8x" example), +which no repo in this measurement corpus represents at the standard cap. + +--- + +## 13. E7 — the connection-pool property, confirmed and pinned + +`pool_size == effective_workers` (raised to 4), `max_overflow=0`. Holds by +**sequencing, not construction**: `indexer/job.py`'s `with engine.connect() as +shas_conn:` (the advisory `shas_fn` read, #104) opens and closes a second, +short-lived connection strictly BEFORE `_precompute_chunk_writer`/embedding — +confirmed by direct code reading (job.py:1298 area) and pinned by a new test, +`tests/unit/test_job.py::test_shas_fn_connection_closes_before_embedding_starts`, +which counts currently-open connections and asserts exactly 1 while `shas_fn` +runs and 0 by the time embedding starts. No live-job evidence of Lakebase +connection pressure at N=4 (no `QueuePool` timeouts in either Arm B run), but +the ceiling itself did double (2→4 concurrent connections on the semantic +path) — worth knowing if a future Lakebase compute-size change is on the table +alongside a semantic-path concurrency change. + +Also found (documented, not a blocker): `embedding_concurrency` at its +`le=8` config ceiling combined with the new clamp of 4 now yields `4×8=32` +in-flight gateway requests — **exceeding** the SDK's 20-connection pool for +the first time (`2×8=16` stayed under it before). `pool_block=True` means this +degrades to silent serialization, not an error, so it is not a correctness +issue, but is flagged in `config.yaml`, `app/config.py`, and both runbooks. + +--- + +## 14. The semantics tripwire (§5.1 of the plan) + +`tests/unit/test_semantics_version_tripwire.py::test_semantics_change_bumps_the_index_semantics_version` +**fails on this PR**, as expected and pre-documented by the plan: `SEMANTICS_PATHS` +includes `indexer/ingest.py`, which this PR touches (the `MAX_EXTRACTED_BYTES` +comment, §12) and which — per the test's own module docstring — is *added* +relative to `origin/master` locally (postdates master, landed by #106, +unmerged past this integration branch), producing an expected false positive +this test's own docstring names verbatim. **Per the plan's §5.1, this is +explicitly NOT a stop condition.** `INDEX_SEMANTICS_VERSION` was **not** +bumped and `indexer/ingest.py` was **not** removed from `SEMANTICS_PATHS` — +this section is that required "say so in the PR." + +--- + +## 15. Limitations + +- **Prod** is out of scope and unreachable; dev serverless is the labelled + proxy throughout. +- **Corpus scale**: the real dev corpus (19 repos, 1142 files) is far too + small on its own; the measurement corpus (+opencode/nanoclaw/claw-code) is a + labelled proxy, not production traffic. +- **`M`'s instability** (§5.1): a ≥2x spread was observed between two container + instances of the same job. The smaller, conservative reading was used + throughout; the real production ceiling could be substantially larger. +- **`RUSAGE_CHILDREN` COW contamination** (§3.1): Arm A/B's `children=` figures + are upper bounds, not literal per-process costs, whenever the pool is + (re)spawned after chunk-writer inflation (the real, unavoidable production + call order). +- **`B*` is a sample max** over an executor-chosen corpus (§8.3 of the plan): + passing at N=4 on this corpus is necessary, not sufficient evidence for + every possible future repo. diff --git a/docs/runbooks/indexing-parallelism.md b/docs/runbooks/indexing-parallelism.md index ee82ebd..fc7cbd0 100644 --- a/docs/runbooks/indexing-parallelism.md +++ b/docs/runbooks/indexing-parallelism.md @@ -324,9 +324,20 @@ Only the first of the two byte caps is a **disk** cap. Since #106 the tarball is streamed once, in memory, and is never extracted, so `MAX_EXTRACTED_BYTES` is a **work** cap — a decompression-bomb guard on how much content one branch may pull out of its archive — and it lives in `indexer/ingest.py`, beside its only -consumer, rather than in `indexer/fetch.py`. The two therefore no longer sum: -the compressed tarball is the only artifact on disk, so peak local disk is -`index_concurrency` × 500 MB: +consumer, rather than in `indexer/fetch.py`. #109 re-derived both this and +`semantic_max_chunks_per_repo`'s global default as **memory** limits (pinned at +the N=2 operating point, to avoid the circularity of solving for `B` from a +model whose dominant term is `B`): a derived breach-regime ceiling of ~880 MiB +(vs. the current 2 GB) and a derived chunk-cap ceiling of ~73,300 chunks (vs. +the current 8000). **Neither was changed**: no repo in the #109 measurement +corpus approached either the current or the derived byte ceiling (largest +observed branch: ~31.5 MiB, ~3.6% of the derived bound), so this run gives no +empirical signal either way, and lowering `MAX_EXTRACTED_BYTES` closes the +whole run's reconciliation checkpoint on breach — too large a blast radius to +change on an untested corpus. Both derived values are recorded here and in +`docs/perf/issue-109-measurements.md` for the next time this needs re-deriving. +The two therefore no longer sum: the compressed tarball is the only artifact on +disk, so peak local disk is `index_concurrency` × 500 MB: | `index_concurrency` | Peak local disk | |---|---| @@ -345,27 +356,68 @@ Raise `index_concurrency` only for repo-level (disk-bound) fan-out; raise `extract_processes` for CPU-bound extraction throughput. The 4 GB disk figure above is still a hard, linear, unavoidable cost of `index_concurrency` alone. (#106 lowered these numbers by 5x but deliberately did **not** move the default -of 4; re-deriving it is #109's job.) - -**Semantic indexing clamps the pool to 2**, regardless of `index_concurrency`. -That clamp is a *memory* bound, not a CPU one: embedding materialises a whole -repo's chunks in memory (~0.5-0.8 GB per worker). The clamp is logged: +of 4; #109 re-derived it and left it at 4 -- see next.) + +**The `index_concurrency` default itself stays 4 (#109 measured, did not just +inherit, this).** The per-thread ingest pass added by #106 +(`iter_tar_source_files`) mixes GIL-bound Python (the `tf.next()` walk, +`fh.read()`, the NUL-strip, `str.decode` -- which does **not** release the GIL) +with one GIL-releasing step (`zlib.decompress`). Measured thread-scaling on this +pass: **1.15-1.20x at 4 concurrent threads**, both idle and with the extraction +pool live -- short of the 2.0x threshold this repo's measurements use to call +something "parallelizes", and consistent with a mostly GIL-held pass rather +than a genuinely parallel one. No default change is warranted on CPU grounds; +the semantic clamp raise to 4 (above) happens to make `effective_workers` equal +`index_concurrency` at the default with zero separate change needed. + +**Semantic indexing clamps the pool to 4**, regardless of `index_concurrency` +(issue #109 raised this from 2). That clamp is a *memory* bound, not a CPU one: +embedding materialises a whole repo's chunks in memory (~32 KB/chunk structural, +~40.1 KB/chunk resident, measured). **The clamp gates `index_concurrency` +itself: raising the configured default above 4 is a no-op on the semantic path +until the clamp moves too** (`effective_workers` is `min(index_concurrency, +clamp)`) — this is why #109 could not treat the two knobs independently. The +clamp is logged only when it actually reduces the configured value: ``` -INFO indexer.job [-]: semantic enabled: clamping index_concurrency 6 -> 2 (memory bound: ...) +INFO indexer.job [-]: semantic enabled: clamping index_concurrency 6 -> 4 (memory bound: ...) ``` +At `index_concurrency <= 4` (the shipped default) the clamp is a no-op and this +line does not appear at all — confirmed on the live dev job at `index_concurrency: +4`, clamp 4: no clamp line, `4 worker(s)` on the disk line instead. + +**Re-derivation (#109).** A measured model — +`P_worst(N) = N * max((alpha+gamma)*B_breach, (alpha+gamma)*(d*C) + V_cap*C) + +extract_processes*R_proc + P_fixed`, coefficients measured across 4 corpora +(`alpha+gamma` up to 4.64 bytes-materialized per source byte, `d` up to 3748 +bytes/chunk, `V_cap` 40.1 KB/chunk resident), against a measured container +memory ceiling (`M`, read from cgroup + an allocate-until-failure bracket) — +showed N=4 clears 70% of `M` with a large margin at the standard 8000-chunk +global cap, and empirical confirmation on the real dev job agreed: two runs each +at N=2 and N=4 measured `peak rss: self=... children=...` (issue #109's new +instrumentation, `RUSAGE_SELF`/`RUSAGE_CHILDREN` at the end of `run()`) — N=2 +averaged ~90% of the 0.7*M budget, N=4 ~83%, both safely under, N=4 with *more* +margin. See `docs/perf/issue-109-measurements.md` for the full derivation, +every coefficient's provenance, and the two derived byte limits +(`MAX_EXTRACTED_BYTES` and the chunk cap `C`) this same model yields. + ### Embedding concurrency (#107) `workers x concurrency` is the number that matters, not `concurrency` alone. -Each of the (at most 2, semantic-clamped) workers dispatches up to -`semantic.embedding_concurrency` embedding batches at once +Each of the (at most 4, semantic-clamped -- #109 raised this from 2) workers +dispatches up to `semantic.embedding_concurrency` embedding batches at once (`app/embed.py:databricks_embedder`, order-preserving `ThreadPoolExecutor.map`): -2 x 4 = 8 in-flight gateway requests at the default, 2 x 8 = 16 at the -config.yaml-enforced ceiling of 8 (the `CODE_SEARCH_SEMANTIC_EMBEDDING_CONCURRENCY` -env var carries no ceiling, mirroring `semantic_embedding_batch_size`'s own -unbounded env surface -- config.yaml is the job's real surface regardless), both -under the SDK's 20-connection pool. +4 x 4 = 16 in-flight gateway requests at the default, **4 x 8 = 32 at the +config.yaml-enforced ceiling of 8 — this now EXCEEDS the SDK's 20-connection +pool** (the `CODE_SEARCH_SEMANTIC_EMBEDDING_CONCURRENCY` env var carries no +ceiling, mirroring `semantic_embedding_batch_size`'s own unbounded env surface +-- config.yaml is the job's real surface regardless). This combination was not +reachable before #109 (2 x 8 = 16 stayed under the pool); it is now, and is +flagged here rather than gated in code, matching this repo's "guardrail +constants with config-level fixes, not override flags" convention -- lower +`embedding_concurrency` if raising it alongside a near-ceiling +`index_concurrency`. `embedding_concurrency: 1` is the rollback switch — fully serial embedding, no thread pool spawned. See `docs/runbooks/semantic-enablement.md` §4 for the full in-flight/memory arithmetic and the 429 posture. @@ -408,7 +460,7 @@ or because the pool degraded. Three WARNING shapes to recognize: - **`... rebuilt the pool (generation N, rebuild M/3)`** — a worker died (`BrokenProcessPool`, e.g. a native crash in a grammar, an OOM kill). The branch(es) in flight at that moment failed (up to `index_concurrency` - branches, semantic-clamped to 2 — **not just one**: the pool is shared, so a + branches, semantic-clamped to 4 — **not just one**: the pool is shared, so a break can surface on every repo worker holding a future at that instant). Each failed branch re-indexes on its next run (it never got a stamp — the same self-healing property §1 describes). The pool is rebuilt and later @@ -450,7 +502,7 @@ retention (#105, §2.3) for the same in-flight branch. ### The connection pool follows the workers -Each worker holds exactly one connection, so the engine is built with +Each worker holds exactly one connection AT A TIME, so the engine is built with `pool_size == effective workers`, `max_overflow=0`, `pool_timeout=30`. There is deliberately **zero headroom**: a connection leak stalls loudly for 30 seconds and then raises, rather than growing the pool silently. If you see a @@ -458,6 +510,24 @@ and then raises, rather than growing the pool silently. If you see a connection, not an undersized pool — the pool is sized to the workers by construction. +**"One connection at a time" holds by sequencing, not by construction (#104, +verified by #109).** On the delta-gate-open semantic path a worker opens a +*second*, short-lived connection (`indexer/job.py`'s `with engine.connect() as +shas_conn:`, for the advisory `shas_fn` read) before its main `index_fn` +connection — but that second connection is closed before embedding/`index_fn` +starts, so at most one is ever open per worker at once. `pool_size == +effective_workers` (raised to 4 by #109 — see §3 above) stays correct only +because of that ordering; a future change that opened both connections +concurrently would silently under-provision the pool. Pinned by +`tests/unit/test_job.py::test_shas_fn_connection_closes_before_embedding_starts`. + +**Since #109 raised the semantic clamp from 2 to 4, the Lakebase connection +ceiling doubles on the semantic path too** — 4 concurrent connections at the +new default vs. 2 before. Not observed to be a bottleneck in either Arm B run +(no `QueuePool` timeouts, no degraded coverage), but worth knowing if a future +Lakebase compute-size change is on the table alongside a semantic-path +concurrency change. + The app/serving pool is separate and unaffected (5, paired with a matching `CapacityLimiter`). diff --git a/docs/runbooks/semantic-enablement.md b/docs/runbooks/semantic-enablement.md index e554afb..14e8032 100644 --- a/docs/runbooks/semantic-enablement.md +++ b/docs/runbooks/semantic-enablement.md @@ -81,7 +81,7 @@ semantic: ``` which makes the job a true semantic no-op (no embedder built, no chunking, the -2-worker clamp not applied) — no bundle/env change and no redeploy of the job's +4-worker clamp not applied) — no bundle/env change and no redeploy of the job's environment. Precedence for the job is `config.yaml > CODE_SEARCH_* env > default`, so `semantic.enabled: false` wins even if the env says enabled. @@ -141,45 +141,58 @@ buffered chunks, and that `semantic_max_chunks_per_repo` (`app/config.py`, defau silently truncating if a repo exceeds the ceiling. That default is deliberately conservative: the buffered vectors are Python float lists -costing ~32 B per element, so at `dim=1024` each chunk is ~32 KB and 8000 chunks is -~260 MB resident, held for the duration of the repo's write transaction. Raising it -scales memory linearly (50000 would be ~1.6 GB and would OOM a typical job container -*before* the loud ceiling check could fire, which defeats the purpose of the ceiling). -If a repo legitimately needs more, prefer the temp-table staging path (follow-up) over -raising this number. +costing ~32 B per element structural, but ~40.1 KB/chunk RESIDENT once measured (issue +#109 -- pymalloc overhead/fragmentation; use this figure for headroom arithmetic), so at +`dim=1024` 8000 chunks is ~313 MiB resident, held for the duration of the repo's write +transaction. Raising it scales memory roughly linearly (50000 would be ~1.9 GiB). Issue +#109 derived a full per-worker chunk-cap ceiling from the container-memory model: ~73,300 +chunks at the pinned N=2 semantic-worker count the derivation is evaluated at (methodology +only -- breaks a circularity in the model), ~36,700 at the shipped N=4 (see +`docs/perf/issue-109-measurements.md` §12) -- the current 8000 default uses well under a +quarter of either budget. A ceiling well past that derived bound (e.g. 50k) risks OOMing +the job container *before* the loud ceiling check could fire, which defeats the purpose of +the ceiling. If a repo legitimately needs more, prefer the temp-table staging path +(follow-up) over raising this number. **Per-repo override, without moving the global default:** `config.yaml`'s top-level `semantic_max_chunks_per_repo` map (`indexer/repo_config.py`) lets one outsized repo get its own cap — `indexer/resolve.py` carries the matched override onto that repo's `RepoEntry`, and `indexer/job.py` uses it in place of `cfg.semantic_max_chunks_per_repo` for that repo only (an active override is logged at INFO). It does not relax the -2-worker semantic clamp above, so a large override still multiplies whichever of the -(at most 2) concurrent workers happens to be indexing that repo — do the same ~32 -KB/chunk math against the override value, not just the global default, before setting -one. To move the **global** cap for the whole job instead of one repo, set +4-worker semantic clamp above (issue #109 raised this from 2), so a large override +still multiplies whichever of the (at most 4) concurrent workers happens to be +indexing that repo — do the same ~32 KB structural / ~40.1 KB resident per-chunk math +against the override value, not just the global default, before setting one. To move +the **global** cap for the whole job instead of one repo, set `semantic.max_chunks_per_repo` (a single int, section 6) — the map still wins for a repo it names. -**Parallelism:** with semantic on by default, the `effective_workers` clamp to 2 in -`indexer/job.py` now applies to every index run by default (each worker materialises a -whole repo's chunks) — see `docs/runbooks/indexing-parallelism.md`. +**Parallelism:** with semantic on by default, the `effective_workers` clamp to 4 +(issue #109; previously 2) in `indexer/job.py` now applies to every index run by +default (each worker materialises a whole repo's chunks) — see +`docs/runbooks/indexing-parallelism.md`. **Concurrent embedding requests (#107):** each worker's `embed()` call (`app/embed.py`) dispatches up to `semantic.embedding_concurrency` batches at once via a `ThreadPoolExecutor`, using `.map()` — never `as_completed()` — so vectors always come back in submission order regardless of which request finishes first. Total in-flight gateway requests for the job is `effective_workers x concurrency`: -2 x 4 = 8 at the default `embedding_concurrency: 4`, 2 x 8 = 16 at the config's -`le=8` ceiling, both under the SDK's 20-connection pool +4 x 4 = 16 at the default `embedding_concurrency: 4`, **4 x 8 = 32 at the config's +`le=8` ceiling — this now EXCEEDS the SDK's 20-connection pool** (issue #109 raised +`effective_workers` from 2 to 4; 2 x 8 = 16 stayed under the pool before) (`HTTPAdapter(pool_connections=20, pool_maxsize=20, pool_block=True)` — `pool_block=True` means exceeding the pool **silently serializes** requests rather -than raising, so staying under 20 is the load-bearing bound, not a nice-to-have). +than raising, so staying under 20 is the load-bearing bound, not a nice-to-have — a +run at the ceiling of both knobs still completes, just with less real concurrency +than configured). Lower `embedding_concurrency` if raising it alongside a +near-ceiling `index_concurrency`. The only new per-in-flight-batch memory cost is transient request/response buffers (~3.5 MB each: ~2.1 MB parsed vectors + ~1.3 MB raw JSON response + a -small request body) — ~28 MB at the default, ~56 MB at the ceiling, negligible -beside the ~0.5–0.8 GB/worker baseline above. `embedding_concurrency: 1` restores -today's fully serial embedding and spawns no thread pool at all (the rollback -switch). +small request body) — ~56 MB at the default (4 workers x 4), ~112 MB at the +ceiling, negligible beside the ~313 MiB/worker vector baseline above (8000 +chunks at the resident 40.1 KB/chunk figure). `embedding_concurrency: 1` +restores today's fully serial embedding and spawns no thread pool at all (the +rollback switch). 429s from the AI Gateway are absorbed entirely by the `databricks-sdk`'s own `Retry-After`-honouring backoff (`_RetryAfterCustomizer`, defaulting to 1s when diff --git a/indexer/ingest.py b/indexer/ingest.py index e630cf3..02737aa 100644 --- a/indexer/ingest.py +++ b/indexer/ingest.py @@ -126,6 +126,18 @@ # bounds members of ANY type. Regular-file accounting alone would let an # archive of a million directory or link headers decompress unbounded -- # they carry no data, so `streamed` never moves. +# +# issue #109 re-derived this as a MEMORY limit (previously undirected): at the +# pinned N=2 operating point, against a measured container ceiling and the +# semantic path's measured bytes-materialized-per-source-byte coefficients, the +# breach-regime peak stays under 0.7x that ceiling only for a per-branch source +# total <= ~880 MiB (vs this 2 GB). NOT changed here: no branch in the +# measurement corpus approached either value (largest observed: ~31.5 MiB, ~3.6% +# of the derived bound), so there is no empirical signal to justify lowering a +# constant whose breach fails the branch and closes the whole run's +# reconciliation checkpoint (job.py's `_decide_reconciliation` requires +# `failures == 0`). See docs/perf/issue-109-measurements.md for the full +# derivation. MAX_EXTRACTED_BYTES = 2_000_000_000 diff --git a/indexer/job.py b/indexer/job.py index 16e8d23..f7631ce 100644 --- a/indexer/job.py +++ b/indexer/job.py @@ -121,6 +121,7 @@ import argparse import base64 import logging +import resource import shutil import sys import tempfile @@ -410,10 +411,16 @@ def run( owns_engine = engine is None if engine is None: # The pool is DERIVED from the worker count, not a constant: each worker - # holds exactly one connection (one engine.connect() per repo, and the - # embed/chunk precompute happens before it), so pool_size == workers is - # exactly enough and max_overflow=0 turns a connection leak into a loud - # stall instead of silent pool growth. pool_timeout is SQLAlchemy's own + # holds at most ONE connection AT A TIME -- by sequencing, not by + # construction (issue #109 E7). On the delta-gate-open semantic path a + # worker opens a second, short-lived connection for the advisory + # shas_fn read (below), but closes it before the embed/chunk precompute + # and its own index_fn connection open -- see + # tests/unit/test_job.py::test_shas_fn_connection_closes_before_embedding_starts. + # So pool_size == workers is exactly enough and max_overflow=0 turns a + # connection leak into a loud stall instead of silent pool growth (a + # future change that opened both connections concurrently would need a + # bigger pool). pool_timeout is SQLAlchemy's own # default, spelled out HERE because max_overflow=0 is what makes it # observable -- a reader seeing the overflow ban must not have to go look # up how long the resulting stall lasts. Passing pool_size explicitly is @@ -616,6 +623,16 @@ def run( len(entries), time.monotonic() - run_started, ) + # issue #109 AC3: peak RSS for this run, self (the main process, every repo + # worker thread) and children (the #108 extraction pool's worker processes, + # otherwise invisible from here) -- ru_maxrss is a high-water mark, in KB on + # Linux, so no delta/baseline subtraction is needed the way a local + # measurement script needs one against its own import-time baseline. + logger.info( + "peak rss: self=%d KB children=%d KB", + resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, + resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss, + ) # Conflicts do NOT fail the run because they SELF-HEAL -- the stamp that # displaced them makes the next run re-index that branch unconditionally. # Note this trades a paging signal for one run of staleness on that branch; diff --git a/indexer/repo_config.py b/indexer/repo_config.py index 38cacf5..fb42c7c 100644 --- a/indexer/repo_config.py +++ b/indexer/repo_config.py @@ -270,16 +270,25 @@ class RepoConfig(BaseModel): from ``index_concurrency`` entirely. Raise ``index_concurrency`` only knowing you are buying disk-bound repo fan-out, not extraction throughput. - When semantic indexing is on, the effective worker count is clamped to 2 by - :func:`effective_workers`. That clamp is a **memory** bound, not a CPU one: - embedding materialises a whole repo's chunks in memory (~0.5-0.8 GB per - worker; 260 MB of vectors alone at the 8000-chunk ceiling). + When semantic indexing is on, the effective worker count is clamped to 4 by + :func:`effective_workers` (issue #109 re-derived this from 2: a measured + ``P_worst`` model -- ``(alpha+gamma)`` bytes-materialized-per-source-byte + coefficients, measured resident vector cost, #108's per-process RSS, and a + measured container memory ceiling -- showed N=4 clears 0.7x the container + budget with margin, and two live-job runs at N=4 confirmed it empirically: + peak self+children RSS landed at ~83% of budget, actually MORE comfortable + than N=2's own ~90%. See docs/perf/issue-109-measurements.md). That clamp is + still a **memory** bound, not a CPU one: embedding materialises a whole + repo's chunks in memory. Per-chunk vector cost is ~32 KB structural (dim=1024 + Python float-list storage) but ~40.1 KB RESIDENT (measured; pymalloc + overhead/fragmentation) -- use the resident figure for headroom arithmetic -- + so 313 MiB of vectors alone at the 8000-chunk ceiling. ``semantic_max_chunks_per_repo`` (the per-repo MAP) overrides that global 8000-chunk ceiling for individual repos named here, without moving the global - default. It does NOT relax the 2-worker semantic clamp above -- a large + default. It does NOT relax the 4-worker semantic clamp above -- a large override still multiplies the per-worker memory cost of whichever of the (at - most 2) concurrent semantic workers happens to be indexing that repo. + most 4) concurrent semantic workers happens to be indexing that repo. The similarly-named ``semantic.max_chunks_per_repo`` (inside the ``semantic:`` block, a single INT) moves that GLOBAL ceiling itself for the whole job. The @@ -354,11 +363,17 @@ def _normalize_semantic_overrides(self) -> RepoConfig: def effective_workers(config: RepoConfig, *, semantic_enabled: bool) -> int: """Worker-pool size for a run, applying the semantic memory clamp. + The clamp is 4 (issue #109; previously 2 -- see :class:`RepoConfig`'s + docstring for the re-derivation and empirical Arm A/B confirmation). It is a + ceiling, never a floor: an ``index_concurrency`` below the clamp passes + through unchanged, so N=3 (or any other legal value) is a real, reachable + ``effective_workers`` outcome, not just N in {1, 2, 4}. + Takes a plain ``bool`` rather than ``Settings`` so this module keeps its import-light property (see the module docstring). """ if semantic_enabled: - return min(config.index_concurrency, 2) + return min(config.index_concurrency, 4) return config.index_concurrency diff --git a/scripts/measure_ingest_threads.py b/scripts/measure_ingest_threads.py new file mode 100644 index 0000000..8a0e23b --- /dev/null +++ b/scripts/measure_ingest_threads.py @@ -0,0 +1,473 @@ +"""Measure `iter_tar_source_files` thread-scaling and its GIL-bound component +mix (#109 §3.4, E4 -- "M2: does the ingest pass scale across threads?"). + +Since #106 the per-branch file source is a single serial pass over one open +``TarFile`` (:func:`indexer.ingest.iter_tar_source_files`): a ``tf.next()`` +member walk, ``fh.read()``, the NUL-sniff binary check +(:func:`indexer.parse._looks_binary`), a UTF-8 decode, and a NUL-strip. Raising +``index_concurrency`` multiplies *concurrent* ingest passes across repo-worker +threads, and whether that helps is gated on which of those steps hold the GIL. +A planning-time synthetic probe (40 MB payload, isolated ``bytes.decode`` vs +``zlib.decompress``) found decode GIL-bound and zlib GIL-releasing; this script +re-measures the FULL real pass, decomposed, over real tarballs, rather than +re-quoting that probe. + +Two things are measured, and the "decompose by component" and "end-to-end +speedup" numbers deliberately come from two different code paths per the +plan's own instruction: + +* **Component breakdown** -- a re-walk of the tarball using the SAME private + primitives ``iter_tar_source_files`` calls (``indexer.ingest + ._normalise_member_name`` / ``._assert_link_target_is_contained``, + ``indexer.parse._looks_binary``), imported rather than re-forked, with a + ``time.perf_counter()`` pair around each step. This is a timing-only + reimplementation of the loop -- never the source of the speedup numbers. +* **End-to-end speedup** -- always calls the REAL + ``indexer.ingest.iter_tar_source_files`` and nothing else, N threads each + streaming its OWN distinct real tarball (page cache and content mix must not + be shared across threads), compared against a single-thread sequential pass + over the SAME N tarballs. + +Two conditions: CPU otherwise idle, and with :class:`indexer.extract_pool. +ExtractionPool` constructed and continuously driving real ``.stream()`` +extraction on a background thread, to see whether process-pool contention for +cores changes the picture. + +**The tarballs must be real** -- fetched over HTTP from public GitHub repos via +:func:`indexer.fetch.download_tarball` (unauthenticated ``httpx.Client`` works +fine for public repos) and cached locally so a re-run doesn't re-download. + +Usage: ``uv run python scripts/measure_ingest_threads.py [--repeat 3] +[--pool-processes 4] [--cache-dir /tmp/measure_ingest_threads_cache] +[--repos org/repo@ref,org/repo@ref,...]`` (need >= 4 distinct repos). +""" + +from __future__ import annotations + +import argparse +import gzip +import tarfile +import threading +import time +from collections.abc import Callable, Sequence +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path +from typing import TypeVar + +import httpx + +from indexer.extract_pool import ExtractionPool +from indexer.fetch import download_tarball +from indexer.ingest import ( + MAX_EXTRACTED_BYTES, + _assert_link_target_is_contained, + _normalise_member_name, + iter_tar_source_files, +) +from indexer.languages import MAX_FILE_BYTES +from indexer.parse import _looks_binary + +T = TypeVar("T") + +# Four distinct public repos of comparable decoded size (~20-37 MB each, +# verified by hand before picking these four) -- comparable size matters for +# the N=4 comparison specifically: four wildly mismatched tarballs would let +# the largest one dominate both the sequential sum and the concurrent wall +# clock, measuring "how fast is the biggest repo alone" rather than genuine +# 4-way thread scaling. +DEFAULT_REPOS = [ + ("tiangolo", "fastapi", "master"), + ("sqlalchemy", "sqlalchemy", "main"), + ("sympy", "sympy", "master"), + ("django", "django", "main"), +] + + +def _parse_repos(spec: str) -> list[tuple[str, str, str]]: + out = [] + for item in spec.split(","): + org_repo, _, ref = item.partition("@") + org, _, repo = org_repo.partition("/") + out.append((org, repo, ref or "HEAD")) + return out + + +def _cache_path(cache_dir: Path, org: str, repo: str) -> Path: + return cache_dir / f"{org}__{repo}.tar.gz" + + +def fetch_tarballs(repos: Sequence[tuple[str, str, str]], cache_dir: Path) -> list[Path]: + """Download each repo's tarball once (cached under ``cache_dir`` across runs).""" + cache_dir.mkdir(parents=True, exist_ok=True) + paths = [] + with httpx.Client(timeout=120.0) as client: + for org, repo, ref in repos: + dest = _cache_path(cache_dir, org, repo) + if not dest.exists(): + print(f"fetching {org}/{repo}@{ref} ...") + tmp_dir = cache_dir / f"_dl_{org}_{repo}" + downloaded = download_tarball(client, org, repo, ref, tmp_dir) + downloaded.replace(dest) + try: + tmp_dir.rmdir() + except OSError: + pass + paths.append(dest) + return paths + + +@dataclass +class ComponentTimes: + """Cumulative wall-clock seconds per component of one (or more, summed) + instrumented walks -- see :func:`decompose_walk`.""" + + tf_next: float = 0.0 + fh_read: float = 0.0 + looks_binary: float = 0.0 + decode: float = 0.0 + nul_strip: float = 0.0 + n_files: int = 0 + n_members: int = 0 + + def add(self, other: "ComponentTimes") -> None: + self.tf_next += other.tf_next + self.fh_read += other.fh_read + self.looks_binary += other.looks_binary + self.decode += other.decode + self.nul_strip += other.nul_strip + self.n_files += other.n_files + self.n_members += other.n_members + + @property + def total(self) -> float: + return self.tf_next + self.fh_read + self.looks_binary + self.decode + self.nul_strip + + +def decompose_walk(tar_path: Path) -> ComponentTimes: + """Re-walk ``tar_path`` with the exact filter chain + ``indexer.ingest.iter_tar_source_files`` uses, timing each component with + its own ``perf_counter()`` pair. + + A TIMING-ONLY reimplementation of that function's loop body: it imports the + same private primitives (``_normalise_member_name``, + ``_assert_link_target_is_contained``, ``_looks_binary``) rather than + re-forking their logic, so the filter population -- which members get as + far as ``fh.read()`` / decode -- matches production exactly. It does not + build ``ParsedFile`` objects or return content; the real end-to-end number + always comes from calling ``iter_tar_source_files`` itself (see + :func:`run_sequential_real` / :func:`run_concurrent_real`), never from this + function. + """ + ct = ComponentTimes() + tf = tarfile.open(tar_path, mode="r:gz") + try: + top_dir: str | None = None + streamed = 0 + seen: set[str] = set() + while True: + t0 = time.perf_counter() + member = tf.next() + ct.tf_next += time.perf_counter() - t0 + if member is None: + break + ct.n_members += 1 + tf.members.clear() # type: ignore[attr-defined] + + name = _normalise_member_name(member.name) + component = name.split("/", 1)[0] + if top_dir is None: + top_dir = component + elif component != top_dir: + raise ValueError( + "expected exactly one top-level dir in tarball, found " + f"{sorted({top_dir, component})}" + ) + + if member.islnk() or member.issym(): + _assert_link_target_is_contained(name, member.linkname) + + if member.offset > MAX_EXTRACTED_BYTES: + raise ValueError( + f"tarball stream reaches {member.offset} decompressed bytes, " + f"exceeding {MAX_EXTRACTED_BYTES}" + ) + + if not member.isreg(): + continue + + streamed += member.size + if streamed > MAX_EXTRACTED_BYTES: + raise ValueError( + f"tarball streams to {streamed} bytes of content, " + f"exceeding {MAX_EXTRACTED_BYTES}" + ) + + rel_path = name[len(top_dir) + 1 :] if name != top_dir else "" + if not rel_path: + continue + if ".git" in rel_path.split("/"): + continue + if member.size > MAX_FILE_BYTES: + continue + if rel_path in seen: + continue + seen.add(rel_path) + + fh = tf.extractfile(member) + if fh is None: + continue + t0 = time.perf_counter() + with fh: + raw = fh.read() + ct.fh_read += time.perf_counter() - t0 + + t0 = time.perf_counter() + is_binary = _looks_binary(raw) + ct.looks_binary += time.perf_counter() - t0 + if is_binary: + continue + + t0 = time.perf_counter() + try: + content = raw.decode("utf-8") + except UnicodeDecodeError: + ct.decode += time.perf_counter() - t0 + continue + ct.decode += time.perf_counter() - t0 + + t0 = time.perf_counter() + content.replace("\x00", "") + ct.nul_strip += time.perf_counter() - t0 + ct.n_files += 1 + finally: + tf.close() + return ct + + +def measure_zlib_isolated(tar_path: Path) -> float: + """Whole-archive gzip inflate, isolated from tarfile header parsing -- + the cleanest available measurement of the GIL-releasing component named in + the plan's §0.4 prior (``zlib.decompress``, 2.54x at 4 threads on a 40 MB + synthetic payload).""" + raw_gz = tar_path.read_bytes() + start = time.perf_counter() + gzip.decompress(raw_gz) + return time.perf_counter() - start + + +def best_of(fn: Callable[[], T], repeat: int, key: Callable[[T], float]) -> T: + """>=3 repeats, discard one warm-up, report best-of -- the methodology + ``docs/perf/issue-108-measurements.md`` used for #108.""" + results = [fn() for _ in range(max(repeat, 1))] + kept = results[1:] if len(results) > 1 else results + return min(kept, key=key) + + +def run_sequential_real(paths: Sequence[Path]) -> float: + """Single thread, ``iter_tar_source_files`` over every path in ``paths``, + one after another -- the baseline half of the fair sequential-vs-concurrent + comparison (same tarball set both sides).""" + start = time.perf_counter() + for p in paths: + list(iter_tar_source_files(p)) + return time.perf_counter() - start + + +def run_concurrent_real(paths: Sequence[Path]) -> float: + """``len(paths)`` threads, each streaming its OWN tarball via the REAL + ``iter_tar_source_files`` concurrently.""" + start = time.perf_counter() + with ThreadPoolExecutor(max_workers=len(paths)) as ex: + futures = [ex.submit(lambda p=p: list(iter_tar_source_files(p))) for p in paths] + for f in futures: + f.result() + return time.perf_counter() - start + + +def run_concurrent_decomp(paths: Sequence[Path]) -> tuple[float, ComponentTimes]: + """``len(paths)`` threads, each running the instrumented + :func:`decompose_walk` on its own tarball concurrently. Returns the + wall-clock for the whole concurrent run plus the SUM of every thread's + per-component cumulative time, so "does component X's aggregate cost grow + linearly with N" is directly readable off two consecutive rows.""" + start = time.perf_counter() + with ThreadPoolExecutor(max_workers=len(paths)) as ex: + results = [f.result() for f in [ex.submit(decompose_walk, p) for p in paths]] + elapsed = time.perf_counter() - start + total = ComponentTimes() + for r in results: + total.add(r) + return elapsed, total + + +class PoolContention: + """Drives ``ExtractionPool.stream()`` continuously on a background thread + over a fixed file list, to create real multi-process CPU contention for + the "pool live" condition. Started/stopped once per condition, spanning + every N in that condition's matrix.""" + + def __init__(self, files: list, n_processes: int) -> None: + self._pool = ExtractionPool(n_processes=n_processes) + self._files = files + self._stop = threading.Event() + self._thread = threading.Thread(target=self._run, daemon=True) + + def _run(self) -> None: + while not self._stop.is_set(): + list(self._pool.stream(self._files)) + + def start(self) -> None: + self._thread.start() + + def stop(self) -> None: + self._stop.set() + self._thread.join(timeout=30) + self._pool.shutdown() + + +def _bucket(n: int, speedup: float) -> str: + """Fixed-in-advance thresholds from the plan's §3.4.""" + if n == 2: + if speedup >= 1.6: + return "PARALLELIZES" + if speedup <= 1.1: + return "NO-SCALE" + return "ambiguous" + if n == 4: + if speedup >= 2.0: + return "PARALLELIZES" + if speedup <= 1.2: + return "NO-SCALE" + return "PARTIAL" + return "n/a" + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--cache-dir", type=Path, default=Path("/tmp/measure_ingest_threads_cache")) + parser.add_argument("--repeat", type=int, default=3, help=">=3 per the plan's protocol") + parser.add_argument( + "--pool-processes", + type=int, + default=4, + help="ExtractionPool size for the 'pool live' condition", + ) + parser.add_argument( + "--repos", + default=",".join(f"{o}/{r}@{ref}" for o, r, ref in DEFAULT_REPOS), + help="comma-separated org/repo@ref list, need >= 4 distinct repos", + ) + args = parser.parse_args(argv) + + repos = _parse_repos(args.repos) + if len(repos) < 4: + raise SystemExit( + "need >= 4 distinct repos for the N in {1..4} thread-scaling matrix, " + f"got {len(repos)}" + ) + + paths = fetch_tarballs(repos, args.cache_dir) + + print("=" * 100) + print(f"tarballs (cached under {args.cache_dir}):") + for (org, repo, ref), p in zip(repos, paths, strict=True): + files = list(iter_tar_source_files(p)) + total_bytes = sum(len(pf.content) for pf in files) + print( + f" {org}/{repo}@{ref}: {p.name}, {p.stat().st_size / 1e6:.2f} MB compressed, " + f"{len(files)} indexable files, {total_bytes / 1e6:.2f} MB decoded text" + ) + print("=" * 100) + + # ---- Part 1: per-tarball component breakdown, single-threaded, idle CPU ---- + print("\n### Part 1 -- per-tarball component breakdown (single-threaded, CPU idle) ###") + print("component times are cumulative seconds inside the instrumented re-walk (see docstring);") + print( + "zlib_iso_s is a SEPARATE standalone whole-archive gzip.decompress(), not part of the " + "walk sum.\n" + ) + print( + f"{'repo':>22s} {'zlib_iso_s':>10s} {'tf_next_s':>10s} {'fh_read_s':>10s} " + f"{'binary_s':>9s} {'decode_s':>9s} {'nulstrip_s':>10s} {'n_files':>7s} " + f"{'n_members':>9s}" + ) + for (org, repo, ref), p in zip(repos, paths, strict=True): + zlib_s = best_of(lambda p=p: measure_zlib_isolated(p), args.repeat, key=lambda x: x) + ct = best_of(lambda p=p: decompose_walk(p), args.repeat, key=lambda c: c.total) + label = f"{org}/{repo}" + print( + f"{label:>22s} {zlib_s:>10.4f} {ct.tf_next:>10.4f} {ct.fh_read:>10.4f} " + f"{ct.looks_binary:>9.4f} {ct.decode:>9.4f} {ct.nul_strip:>10.4f} " + f"{ct.n_files:>7d} {ct.n_members:>9d}" + ) + + # ---- Part 2: N-thread scaling matrix, two conditions ---- + for condition in ("idle", "pool_live"): + print(f"\n### Part 2 -- N-thread ingest scaling, condition={condition} ###") + contention: PoolContention | None = None + if condition == "pool_live": + load_files = list(iter_tar_source_files(paths[0])) + print( + f"starting background ExtractionPool(n_processes={args.pool_processes}) " + f"driving .stream() over {len(load_files)} files from " + f"{repos[0][0]}/{repos[0][1]} ..." + ) + contention = PoolContention(load_files, n_processes=args.pool_processes) + contention.start() + time.sleep(0.5) # let the process pool spin up before timing starts + try: + print( + "\nend-to-end speedup (REAL iter_tar_source_files; same N-tarball set both sides):" + ) + print( + f"{'N':>3s} {'seq_s (1thr, Nx)':>17s} {'conc_s (Nthr)':>14s} " + f"{'speedup':>9s} {'bucket':>13s}" + ) + for n in (1, 2, 3, 4): + subset = paths[:n] + seq = best_of( + lambda subset=subset: run_sequential_real(subset), args.repeat, key=lambda x: x + ) + if n > 1: + conc = best_of( + lambda subset=subset: run_concurrent_real(subset), + args.repeat, + key=lambda x: x, + ) + else: + conc = seq + speedup = seq / conc if conc else float("inf") + bucket = _bucket(n, speedup) + print(f"{n:>3d} {seq:>17.4f} {conc:>14.4f} {speedup:>8.2f}x {bucket:>13s}") + + print( + "\ncomponent breakdown at each N (instrumented re-walk, N threads concurrent, " + "SUM across threads):" + ) + print( + f"{'N':>3s} {'wall_s':>8s} {'sum_tf_next_s':>13s} {'sum_fh_read_s':>13s} " + f"{'sum_binary_s':>12s} {'sum_decode_s':>12s} {'sum_nulstrip_s':>14s}" + ) + for n in (1, 2, 3, 4): + subset = paths[:n] + elapsed, ct = best_of( + lambda subset=subset: run_concurrent_decomp(subset), + args.repeat, + key=lambda t: t[0], + ) + print( + f"{n:>3d} {elapsed:>8.4f} {ct.tf_next:>13.4f} {ct.fh_read:>13.4f} " + f"{ct.looks_binary:>12.4f} {ct.decode:>12.4f} {ct.nul_strip:>14.4f}" + ) + finally: + if contention is not None: + contention.stop() + print("stopped background ExtractionPool contention.") + + print("\ndone.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/measure_semantic_memory.py b/scripts/measure_semantic_memory.py new file mode 100644 index 0000000..a57d397 --- /dev/null +++ b/scripts/measure_semantic_memory.py @@ -0,0 +1,574 @@ +"""Measure the semantic path's peak memory, decomposed into named terms (#109, AC3). + +Offline companion to ``docs/perf/issue-109-measurements.md`` §3.3(b)'s ``P_worst`` +model. Drives the REAL production call sequence from +``indexer.job._index_one_branch``'s semantic path -- never a re-implementation: + + files = list(iter_tar_source_files(tar_path)) # alpha + files_to_embed = [pf for pf in files # the delta gate + if (pf.path, content_sha(pf.content)) not in carried] + chunk_writer = indexer.job._precompute_chunk_writer( # gamma + vectors + files_to_embed, embed_fn, max_chunks_per_repo) + +``_precompute_chunk_writer`` is private (leading underscore) but imported directly +by name -- this is a measurement script living in the same repo, not an external +consumer, and the alternative (re-typing its chunking/embedding logic here) is +exactly the drift this script exists to avoid. + +**The stub embedder returns DISTINCT floats per vector component** +(``float(i * dim + j)``), never a shared/cached float such as ``[0.0] * dim``. +Planning found that the naive stub understates resident memory by ~4x, because +``[0.0] * dim`` stores ``dim`` references to ONE cached float object rather than +``dim`` independent ``PyFloat`` allocations (see the plan's §2.2). This script's +correctness as a memory probe depends entirely on avoiding that trap. + +**Methodology -- three named RSS terms per corpus/gate combination:** + +Peak RSS (``resource.getrusage(RUSAGE_SELF).ru_maxrss``) is a monotonic +non-decreasing high-water mark *within one process*, so every stage below runs in +its OWN freshly spawned subprocess (this same script, re-invoked with a hidden +``--worker`` flag) -- otherwise stage N's baseline would already carry stage +N-1's peak forward and every delta after the first would be contaminated. Inside +one subprocess, three readings bound three terms: + + 1. baseline (interpreter only) + 2. after ``files = list(iter_tar_source_files(tar))`` -> the FILES term (alpha) + 3. after ``per_file = {p: list(iter_chunks(p)) for p in files_to_embed}`` + -> the CHUNKS term (gamma) + 4. after ``_precompute_chunk_writer(files_to_embed, stub_embed_fn, huge_cap)`` + -> the VECTORS term + +Between readings 3 and 4 the manually-built ``per_file`` dict from step 3 is +deleted and ``gc.collect()``ed *before* calling the real ``_precompute_chunk_writer``, +which re-chunks internally (it has no way to accept precomputed chunks -- that +would be re-implementing its contract, not measuring it). The intent is that the +freed step-3 allocations are reused by the allocator for step 4's structurally +identical rebuild, so the reading-3-to-4 delta is dominated by the NEW allocation +(the embedding vectors) rather than by double-counting the chunk objects. This is +a reasonable approximation on CPython/glibc for same-shaped allocations, not a +guarantee -- reported numbers may run slightly high for exactly this reason, and +that is called out again in the printed report. + +The chunk cap passed to ``_precompute_chunk_writer`` here is deliberately huge +(never the production ``semantic_max_chunks_per_repo``): this script measures the +UNCAPPED terms (alpha, gamma) and the per-chunk vector cost directly, not +cap-breach behavior -- that is a different, later step of issue #109's plan +(§3.3(c)-(f)), not this script's job. + +**V_cap (resident bytes per chunk)** is measured completely separately from the +corpus runs, matching the plan's §2.2 methodology exactly: build exactly 8000 +distinct-float 1024-dim vectors directly (not through ``_precompute_chunk_writer``) +and take one baseline/after delta. + +**N-concurrency (N in {1,2,3,4})** spins up N ``threading.Thread``s, each running +the SAME files -> chunk_writer pipeline against a real tarball, with ONE shared +``indexer.extract_pool.ExtractionPool`` built up front and each thread draining +its own ``pool.stream(files)`` call to completion -- so the #108 process pool's +own resident overhead (R_proc) is live and contributing to the measured peak, +exactly like N concurrent repo-worker threads in production. Each thread builds +its own ``files`` list from its own call to ``iter_tar_source_files`` (never a +shared generator -- the module's own docstring warns that a second thread +advancing the same tar-backed generator corrupts output silently). Both +``RUSAGE_SELF`` (this process, all threads) and ``RUSAGE_CHILDREN`` (the pool's +worker processes) are reported. + +**Corpora**: this repo (``IceRhymers/databricks-code-search``) at HEAD, plus three +other modest real public repos spanning a size range (Flask, Requests, Django), +fetched via the real ``indexer.fetch.download_tarball`` and cached under +``--cache-dir`` so repeat runs do not re-download. Total download is a few tens of +MB, well under the ~200 MB budget. + +Usage: ``uv run python scripts/measure_semantic_memory.py`` +(no arguments needed for the default corpus set; see ``--help`` for overrides). +""" + +from __future__ import annotations + +import argparse +import gc +import itertools +import json +import logging +import random +import resource +import subprocess +import sys +import threading +import time +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +import httpx + +from indexer.extract_pool import ExtractionPool, _available_cpus +from indexer.fetch import download_tarball +from indexer.hashing import content_sha +from indexer.ingest import iter_tar_source_files +from indexer.languages import ParsedFile +from indexer.parse import iter_chunks + +logging.disable(logging.CRITICAL) # keep worker-subprocess stdout free of log noise + +# A generous cap -- never the production semantic_max_chunks_per_repo -- so +# _precompute_chunk_writer never raises the cap-breach ValueError while this +# script measures the uncapped terms it deliberately does not exercise here. +_UNCAPPED_MAX_CHUNKS_PER_REPO = 50_000_000 + +# Default corpus: this repo plus three modest, well-known public repos spanning +# a size range. `ref` is always "HEAD" -- download_tarball's `ref` argument is +# passed straight into GitHub's tarball URL and accepts a branch name. +_DEFAULT_CORPORA = [ + ("databricks-code-search", "IceRhymers", "databricks-code-search"), + ("flask", "pallets", "flask"), + ("requests", "psf", "requests"), + ("django", "django", "django"), +] + +_GATE_STATES: list[tuple[str, float]] = [ + ("first-index", 0.0), # gate closed: files_to_embed IS files, no narrowing + ("recurring-1pct", 0.01), # gate open: carried covers ~99% of files + ("recurring-10pct", 0.10), # gate open: carried covers ~90% of files +] + + +def _rss_kb() -> int: + """Peak RSS of THIS process so far, in KB (Linux ``ru_maxrss`` semantics).""" + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + + +def _children_rss_kb() -> int: + """Peak RSS across reaped child processes, in KB.""" + return resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss + + +def _stub_embed_fn(dim: int = 1024): + """A stub ``EmbedFn`` returning DISTINCT floats per vector component. + + NEVER ``[0.0] * dim`` -- that stores ``dim`` references to one cached float + and understates resident memory by ~4x (planning's §2.2 trap). A running + counter guarantees every float, across every call, is a fresh ``PyFloat``. + """ + counter = itertools.count() + + def embed_fn(texts: list[str]) -> list[list[float]]: + vectors = [] + for _ in texts: + base = next(counter) * dim + vectors.append([float(base + j) for j in range(dim)]) + return vectors + + return embed_fn + + +def _source_bytes(files: Sequence[ParsedFile]) -> int: + """Total UTF-8-encoded byte length of ``files``' content -- the denominator + for both alpha and gamma.""" + return sum(len(pf.content.encode("utf-8")) for pf in files) + + +def _narrow_files_to_embed( + files: list[ParsedFile], gate: str, delta_fraction: float, *, seed: int = 0 +) -> list[ParsedFile]: + """Replicate ``_index_one_branch``'s exact narrowing logic for a simulated + gate state. + + ``gate == "first-index"``: the gate is CLOSED (no stamp at the current + ``INDEX_SEMANTICS_VERSION``), so production sets ``files_to_embed = files`` + verbatim -- no ``carried`` set is even read. This is the shape that OOMs + (§0.2 of the plan): no narrowing benefit at all. + + Otherwise: the gate is OPEN, and ``carried`` is simulated as covering + ``1 - delta_fraction`` of ``files``' ``(path, content_sha)`` pairs (a + deterministic random sample), so ``files_to_embed`` narrows to approximately + ``delta_fraction`` of ``files`` -- using the SAME list-comprehension shape + ``indexer/job.py`` uses, not a re-derived equivalent. + """ + if gate == "first-index": + return files + shas = [(pf.path, content_sha(pf.content)) for pf in files] + n_carry = round(len(shas) * (1 - delta_fraction)) + rng = random.Random(seed) + carried = set(rng.sample(shas, n_carry)) if shas else set() + return [pf for pf in files if (pf.path, content_sha(pf.content)) not in carried] + + +# -------------------------------------------------------------------------- +# Worker bodies -- each runs in ITS OWN freshly spawned subprocess (see the +# module docstring for why ru_maxrss's high-water-mark semantics demand this). +# -------------------------------------------------------------------------- + + +def _worker_stage(cfg: dict[str, Any]) -> dict[str, Any]: + from indexer.job import _precompute_chunk_writer # local: keep worker startup lean + + tarball = Path(cfg["tarball"]) + gate = cfg["gate"] + delta_fraction = cfg["delta_fraction"] + + baseline_kb = _rss_kb() + + files = list(iter_tar_source_files(tarball)) + after_files_kb = _rss_kb() + + files_to_embed = _narrow_files_to_embed(files, gate, delta_fraction) + + per_file_chunks = {pf.path: list(iter_chunks(pf)) for pf in files_to_embed} + after_chunks_kb = _rss_kb() + chunk_count = sum(len(chunks) for chunks in per_file_chunks.values()) + chunk_content_bytes = sum( + len(c.content.encode("utf-8")) for chunks in per_file_chunks.values() for c in chunks + ) + del per_file_chunks + gc.collect() + + embed_fn = _stub_embed_fn() + chunk_writer = _precompute_chunk_writer(files_to_embed, embed_fn, _UNCAPPED_MAX_CHUNKS_PER_REPO) + after_vectors_kb = _rss_kb() + del chunk_writer + + return { + "baseline_kb": baseline_kb, + "after_files_kb": after_files_kb, + "after_chunks_kb": after_chunks_kb, + "after_vectors_kb": after_vectors_kb, + "files_count": len(files), + "files_bytes": _source_bytes(files), + "files_to_embed_count": len(files_to_embed), + "files_to_embed_bytes": _source_bytes(files_to_embed), + "chunk_count": chunk_count, + "chunk_content_bytes": chunk_content_bytes, + } + + +def _worker_vcap(cfg: dict[str, Any]) -> dict[str, Any]: + n = cfg["n"] + dim = cfg["dim"] + baseline_kb = _rss_kb() + vectors = [[float(i * dim + j) for j in range(dim)] for i in range(n)] + after_kb = _rss_kb() + assert len(vectors) == n and len(vectors[0]) == dim + return {"baseline_kb": baseline_kb, "after_kb": after_kb, "n": n, "dim": dim} + + +def _worker_nconc(cfg: dict[str, Any]) -> dict[str, Any]: + from indexer.job import _precompute_chunk_writer # local: keep worker startup lean + + tarball = Path(cfg["tarball"]) + n_threads = cfg["n_threads"] + gate = cfg["gate"] + delta_fraction = cfg["delta_fraction"] + + baseline_self_kb = _rss_kb() + baseline_children_kb = _children_rss_kb() + + n_processes = min(_available_cpus(), 8) + pool = ExtractionPool(n_processes=n_processes) + + errors: list[str] = [] + + def run_one() -> None: + try: + files = list(iter_tar_source_files(tarball)) + files_to_embed = _narrow_files_to_embed(files, gate, delta_fraction) + embed_fn = _stub_embed_fn() + # Held alive (not discarded) across pool.stream() below: production + # (indexer/job.py) keeps chunk_writer's vectors resident for the whole + # index_repo write window, which is exactly the concurrent-residency + # property this stage measures -- freeing it early would let each + # thread's vectors be collected before, or concurrently with, sibling + # threads' peaks, understating true N-way concurrent RSS. + chunk_writer = _precompute_chunk_writer( + files_to_embed, embed_fn, _UNCAPPED_MAX_CHUNKS_PER_REPO + ) + # Drain the pool's stream fully so its worker processes actually do + # (and stay resident for) the same work a real branch would ask of them. + list(pool.stream(files)) + del chunk_writer + except Exception as exc: # noqa: BLE001 -- reported, not swallowed + errors.append(repr(exc)) + + threads = [threading.Thread(target=run_one) for _ in range(n_threads)] + start = time.perf_counter() + for t in threads: + t.start() + for t in threads: + t.join() + elapsed = time.perf_counter() - start + pool.shutdown() + + after_self_kb = _rss_kb() + after_children_kb = _children_rss_kb() + + return { + "n_threads": n_threads, + "n_processes": n_processes, + "baseline_self_kb": baseline_self_kb, + "after_self_kb": after_self_kb, + "baseline_children_kb": baseline_children_kb, + "after_children_kb": after_children_kb, + "elapsed_s": elapsed, + "errors": errors, + } + + +_WORKERS = {"stage": _worker_stage, "vcap": _worker_vcap, "nconc": _worker_nconc} + + +def _run_worker_subprocess(mode: str, cfg: dict[str, Any]) -> dict[str, Any]: + """Re-invoke THIS script in a fresh interpreter to run one measurement. + + Fresh process per measurement is load-bearing, not a style choice: ``ru_maxrss`` + only grows within a process, so reusing one process across stages/corpora would + let an earlier, larger measurement's peak silently leak into a later, smaller + one's baseline. + """ + result = subprocess.run( + [sys.executable, str(Path(__file__).resolve()), "--worker", mode, json.dumps(cfg)], + capture_output=True, + text=True, + check=True, + ) + return json.loads(result.stdout.strip().splitlines()[-1]) + + +def _emit(**fields: object) -> None: + print(" ".join(f"{k}={v}" for k, v in fields.items())) + + +# -------------------------------------------------------------------------- +# Orchestration (normal, non-worker invocation) +# -------------------------------------------------------------------------- + + +def _download_corpora(corpora: list[tuple[str, str, str]], cache_dir: Path) -> dict[str, Path]: + cache_dir.mkdir(parents=True, exist_ok=True) + client = httpx.Client(timeout=120.0) + paths: dict[str, Path] = {} + for name, org, repo in corpora: + dest = cache_dir / name + tar_path = dest / "source.tar.gz" + if tar_path.exists(): + print(f"# {name}: using cached {tar_path} ({tar_path.stat().st_size} bytes)") + else: + print(f"# {name}: downloading {org}/{repo}@HEAD ...") + download_tarball(client, org, repo, "HEAD", dest) + print(f"# {name}: downloaded {tar_path} ({tar_path.stat().st_size} bytes)") + paths[name] = tar_path + return paths + + +def main(argv: Sequence[str] | None = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + + # Hidden worker-dispatch path: `--worker `. Not a public + # CLI surface -- it exists purely so `_run_worker_subprocess` can re-invoke + # this file in a fresh interpreter for one isolated measurement. + if argv and argv[0] == "--worker": + mode, raw_cfg = argv[1], argv[2] + result = _WORKERS[mode](json.loads(raw_cfg)) + print(json.dumps(result)) + return 0 + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--cache-dir", + type=Path, + default=Path("/tmp/measure_semantic_memory_cache"), + help="where downloaded tarballs are cached across runs", + ) + parser.add_argument( + "--corpora", + default=",".join(f"{org}/{repo}" for _, org, repo in _DEFAULT_CORPORA), + help=( + "comma-separated org/repo list, in size order (last is used for the N-concurrency arm)" + ), + ) + parser.add_argument("--dim", type=int, default=1024, help="embedding dimension") + parser.add_argument("--vcap-n", type=int, default=8000, help="vector count for the V_cap probe") + parser.add_argument( + "--n-threads", default="1,2,3,4", help="comma-separated N values for the concurrency arm" + ) + parser.add_argument( + "--nconc-corpus", + default=None, + help="corpus name to use for the N-concurrency arm (default: the last/largest corpus)", + ) + args = parser.parse_args(argv) + + corpus_specs = [] + for entry in args.corpora.split(","): + org, repo = entry.split("/", 1) + corpus_specs.append((repo, org, repo)) + # Prefer the default's friendly names when they line up (cosmetic only). + default_by_org_repo = {(org, repo): name for name, org, repo in _DEFAULT_CORPORA} + corpus_specs = [ + (default_by_org_repo.get((org, repo), repo), org, repo) for _, org, repo in corpus_specs + ] + + print("=" * 78) + print("environment") + print("=" * 78) + _emit(python=sys.version.split()[0], cpu_count_affinity=_available_cpus()) + try: + import shutil as _shutil + + total, _used, free = _shutil.disk_usage("/tmp") + _emit(tmp_total_bytes=total, tmp_free_bytes=free, note="a-tmpfs-box-per-plan-1.1") + except OSError: + pass + print() + + tarballs = _download_corpora(corpus_specs, args.cache_dir) + print() + + print("=" * 78) + print("stage 1: alpha (files) / gamma (chunks) / vectors, per corpus and gate state") + print("=" * 78) + alphas: list[float] = [] + gammas: list[float] = [] + for name, _org, _repo in corpus_specs: + tarball = tarballs[name] + for gate, delta_fraction in _GATE_STATES: + cfg = {"tarball": str(tarball), "gate": gate, "delta_fraction": delta_fraction} + r = _run_worker_subprocess("stage", cfg) + + files_delta_kb = r["after_files_kb"] - r["baseline_kb"] + chunks_delta_kb = r["after_chunks_kb"] - r["after_files_kb"] + vectors_delta_kb = r["after_vectors_kb"] - r["after_chunks_kb"] + + alpha = (files_delta_kb * 1024) / r["files_bytes"] if r["files_bytes"] else float("nan") + gamma = ( + (chunks_delta_kb * 1024) / r["files_to_embed_bytes"] + if r["files_to_embed_bytes"] + else float("nan") + ) + measured_d = ( + r["chunk_content_bytes"] / r["chunk_count"] if r["chunk_count"] else float("nan") + ) + vector_kb_per_chunk = ( + vectors_delta_kb / r["chunk_count"] if r["chunk_count"] else float("nan") + ) + + _emit( + corpus=name, + gate=gate, + baseline_kb=r["baseline_kb"], + after_files_kb=r["after_files_kb"], + after_chunks_kb=r["after_chunks_kb"], + after_vectors_kb=r["after_vectors_kb"], + files_delta_kb=files_delta_kb, + chunks_delta_kb=chunks_delta_kb, + vectors_delta_kb=vectors_delta_kb, + files_count=r["files_count"], + files_bytes=r["files_bytes"], + files_to_embed_count=r["files_to_embed_count"], + files_to_embed_bytes=r["files_to_embed_bytes"], + chunk_count=r["chunk_count"], + measured_d_bytes_per_chunk=round(measured_d, 1), + vector_kb_per_chunk=round(vector_kb_per_chunk, 3), + alpha_files_per_source_byte=round(alpha, 4), + gamma_chunks_per_source_byte=round(gamma, 4), + ) + + if gate == "first-index": + # alpha/gamma are properties of the whole-file materialization; + # the first-index (gate-closed) run is the one where + # files_to_embed IS files, exactly matching planning's §2.1 + # methodology (a whole-corpus measurement, not a narrowed one). + alphas.append(alpha) + gammas.append(gamma) + print() + + print("=" * 78) + print("stage 2: V_cap -- resident bytes for a fixed embedded-vector count") + print("=" * 78) + vcap_cfg = {"n": args.vcap_n, "dim": args.dim} + vr = _run_worker_subprocess("vcap", vcap_cfg) + vcap_delta_kb = vr["after_kb"] - vr["baseline_kb"] + vcap_kb_per_chunk = vcap_delta_kb / vr["n"] + structural_kb_per_chunk = args.dim * (8 + 24) / 1024 # 8B pointer + 24B PyFloat, per §2.2 + _emit( + n=vr["n"], + dim=vr["dim"], + baseline_kb=vr["baseline_kb"], + after_kb=vr["after_kb"], + delta_kb=vcap_delta_kb, + resident_kb_per_chunk=round(vcap_kb_per_chunk, 3), + structural_kb_per_chunk=round(structural_kb_per_chunk, 3), + ) + print() + + print("=" * 78) + print("stage 3: N concurrent branch-index threads, extraction pool live") + print("=" * 78) + nconc_name = args.nconc_corpus or corpus_specs[-1][0] + nconc_tarball = tarballs[nconc_name] + print(f"# using corpus={nconc_name} tarball={nconc_tarball}, gate=first-index") + n_values = [int(n) for n in args.n_threads.split(",")] + for n in n_values: + cfg = { + "tarball": str(nconc_tarball), + "n_threads": n, + "gate": "first-index", + "delta_fraction": 0.0, + } + nr = _run_worker_subprocess("nconc", cfg) + _emit( + corpus=nconc_name, + n_threads=nr["n_threads"], + n_processes=nr["n_processes"], + baseline_self_kb=nr["baseline_self_kb"], + after_self_kb=nr["after_self_kb"], + self_delta_kb=nr["after_self_kb"] - nr["baseline_self_kb"], + baseline_children_kb=nr["baseline_children_kb"], + after_children_kb=nr["after_children_kb"], + children_delta_kb=nr["after_children_kb"] - nr["baseline_children_kb"], + elapsed_s=round(nr["elapsed_s"], 2), + errors=len(nr["errors"]), + ) + for err in nr["errors"]: + print(f"# error: {err}") + print() + + print("=" * 78) + print("summary") + print("=" * 78) + if alphas: + _emit( + alpha_avg=round(sum(alphas) / len(alphas), 4), + alpha_min=round(min(alphas), 4), + alpha_max=round(max(alphas), 4), + n_corpora=len(alphas), + ) + if gammas: + _emit( + gamma_avg=round(sum(gammas) / len(gammas), 4), + gamma_min=round(min(gammas), 4), + gamma_max=round(max(gammas), 4), + n_corpora=len(gammas), + ) + _emit( + v_cap_resident_kb_per_chunk=round(vcap_kb_per_chunk, 3), + v_cap_structural_kb_per_chunk=round(structural_kb_per_chunk, 3), + v_cap_n=vr["n"], + v_cap_dim=vr["dim"], + ) + print( + "# alpha/gamma above are from each corpus's first-index (gate-closed) run, matching " + "the plan's §2.1 whole-file methodology. The recurring-1pct/10pct rows in stage 1 show " + "the SAME alpha (files always materializes in full) alongside a much smaller " + "chunks/vectors delta (files_to_embed narrows), which is the qualitative effect " + "the plan's §0.2 and §3.3(b) describe." + ) + print( + "# vectors_delta_kb in stage 1 may run slightly high: _precompute_chunk_writer " + "re-chunks internally (it has no seam to accept precomputed chunks), so that delta " + "is 'new allocations since the chunks-term reading' rather than a pure vectors-only " + "measurement. See the module docstring for why this is still a reasonable isolation." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py index 8fe93fe..53f7056 100644 --- a/tests/unit/test_job.py +++ b/tests/unit/test_job.py @@ -1187,11 +1187,11 @@ def _embed(texts: list[str]) -> list[list[float]]: def test_config_yaml_semantic_enabled_applies_the_worker_clamp( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - """The enable overlay also re-arms the 2-worker memory clamp. + """The enable overlay also re-arms the 4-worker memory clamp (issue #109). cfg says disabled (which would leave the pool at index_concurrency=6), but config.yaml's ``semantic.enabled: true`` overlays before effective_workers, so - the clamp fires: pool 6 -> 2, with the clamp log line -- the pool-side mirror + the clamp fires: pool 6 -> 4, with the clamp log line -- the pool-side mirror of the disable test. """ with caplog.at_level(logging.INFO, logger="indexer.job"): @@ -1200,8 +1200,8 @@ def test_config_yaml_semantic_enabled_applies_the_worker_clamp( monkeypatch, cfg=Settings(semantic_enabled=False), ) - assert kwargs["pool_size"] == 2 # clamped - assert "clamping index_concurrency 6 -> 2" in caplog.text + assert kwargs["pool_size"] == 4 # clamped + assert "clamping index_concurrency 6 -> 4" in caplog.text @pytest.mark.unit @@ -2096,10 +2096,10 @@ def test_pool_size_follows_the_semantic_clamp_not_the_raw_config( ) -> None: """The pool must track the EFFECTIVE workers, not index_concurrency. - With semantic on, effective_workers clamps 6 -> 2; a pool of 6 would then - over-provision Lakebase connections that no worker can ever use. The clamp - is also logged, because a run silently doing a third of the requested - concurrency is otherwise invisible. + With semantic on, effective_workers clamps 6 -> 4 (issue #109); a pool of 6 + would then over-provision Lakebase connections that no worker can ever use. + The clamp is also logged, because a run silently doing two-thirds of the + requested concurrency is otherwise invisible. """ with caplog.at_level(logging.INFO, logger="indexer.job"): kwargs = _engine_kwargs( @@ -2107,9 +2107,9 @@ def test_pool_size_follows_the_semantic_clamp_not_the_raw_config( monkeypatch, cfg=Settings(semantic_enabled=True), ) - assert kwargs["pool_size"] == 2 + assert kwargs["pool_size"] == 4 assert kwargs["max_overflow"] == 0 - assert "clamping index_concurrency 6 -> 2" in caplog.text + assert "clamping index_concurrency 6 -> 4" in caplog.text @pytest.mark.unit @@ -2118,7 +2118,7 @@ def test_config_yaml_semantic_disabled_removes_the_worker_clamp( ) -> None: """The overlay runs BEFORE effective_workers, so config.yaml can lift the clamp. - cfg says semantic enabled (which would clamp 6 -> 2), but config.yaml's + cfg says semantic enabled (which would clamp 6 -> 4), but config.yaml's ``semantic.enabled: false`` overlays first -- effective_workers then sees a disabled flag and leaves the pool at the full index_concurrency, with no clamp log line. This is the pool-side proof that the overlay precedes both the clamp @@ -2130,10 +2130,112 @@ def test_config_yaml_semantic_disabled_removes_the_worker_clamp( monkeypatch, cfg=Settings(semantic_enabled=True), ) - assert kwargs["pool_size"] == 6 # not clamped to 2 + assert kwargs["pool_size"] == 6 # not clamped to 4 assert "clamping index_concurrency" not in caplog.text +@pytest.mark.unit +def test_no_clamp_line_at_the_shipped_default( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """At `index_concurrency=4` (the shipped default) the clamp is a no-op. + + `effective_workers` returns `min(index_concurrency, 4)`, and the clamp log + line only fires when it actually reduces the value (`workers != + config.index_concurrency`, job.py). So a default-configured, semantic-on run + emits NO clamp line at all -- issue #109's runbook update explicitly documents + this as confirmed on the live dev job's Arm B runs; this pins it in a unit + test too. + """ + with caplog.at_level(logging.INFO, logger="indexer.job"): + kwargs = _engine_kwargs( + _config(repos=["acme/widgets"], index_concurrency=4), + monkeypatch, + cfg=Settings(semantic_enabled=True), + ) + assert kwargs["pool_size"] == 4 + assert "clamping index_concurrency" not in caplog.text + + +@pytest.mark.unit +def test_shas_fn_connection_closes_before_embedding_starts() -> None: + """E7 (issue #109): the advisory ``shas_fn`` connection is short-lived. + + ``pool_size == workers`` (the tests above) only holds because each worker + opens at most ONE connection at a time -- by sequencing, not by + construction. On the delta-gate-open path a worker opens a SECOND, + short-lived connection (``with engine.connect() as shas_conn:``, + ``indexer/job.py``) for ``shas_fn``, which must close before + ``_precompute_chunk_writer``/embedding starts -- otherwise a single worker + could hold 2 connections at once and pool_size==workers would + under-provision. Wraps the fake engine's own ``connect()`` to count + currently-open connections and samples that count from inside ``shas_fn`` + and ``embed_fn`` -- extending + ``test_shas_fn_called_once_before_embedding_when_version_matches_and_sha_differs``'s + pattern one step further. + """ + open_count = 0 + open_during: dict[str, int] = {} + + engine = _FakeEngine(stamps={("acme/widgets", "main"): ("sha_old", INDEX_SEMANTICS_VERSION)}) + real_connect = engine.connect + + class _TrackingConn: + def __init__(self, inner: _FakeConn) -> None: + self._inner = inner + + def __enter__(self) -> _TrackingConn: + nonlocal open_count + open_count += 1 + self._inner.__enter__() + return self + + def __exit__(self, *exc: Any) -> bool: + nonlocal open_count + open_count -= 1 + return self._inner.__exit__(*exc) + + def execute(self, stmt: Any) -> Any: + return self._inner.execute(stmt) + + def rollback(self) -> None: + self._inner.rollback() + + def _tracking_connect() -> _TrackingConn: + return _TrackingConn(real_connect()) + + engine.connect = _tracking_connect # type: ignore[method-assign] + + def _shas_fn(conn: Any, *, name: str, branch: str) -> tuple[set[Any], set[Any]]: + open_during["shas"] = open_count + return set(), set() + + def _embed(texts: list[str]) -> list[list[float]]: + open_during["embed"] = open_count + return [[0.0] for _ in texts] + + cfg = Settings(semantic_enabled=True, semantic_max_chunks_per_repo=100) + idx = _RecordingIndex() + code = _run( + _config(repos=["acme/widgets"]), + idx, + cfg=cfg, + embed_fn=_embed, + engine=engine, + shas_fn=_shas_fn, + ) + assert code == 0 + # Guards against a vacuous pass: job.py swallows any semantic-precompute + # failure and still returns 0 (the semantic layer is additive), which would + # let a degraded run satisfy the connection-count assertion below without + # ever really reaching the embed step it's supposed to pin. + assert idx.chunk_writer is not None, "semantic precompute must have succeeded, not degraded" + assert open_during == {"shas": 1, "embed": 0}, ( + "shas_fn's own connection must be open exactly while it runs, and fully " + f"closed again before embedding starts (got {open_during!r})" + ) + + @pytest.mark.unit def test_indexer_reaches_no_hardcoded_pool_constant() -> None: """Tripwire: the indexer must never fall back to the server's pool default. diff --git a/tests/unit/test_repo_config.py b/tests/unit/test_repo_config.py index 5861752..d372588 100644 --- a/tests/unit/test_repo_config.py +++ b/tests/unit/test_repo_config.py @@ -253,8 +253,11 @@ def test_extract_processes_out_of_range_raises_config_error(value: int) -> None: ("configured", "semantic_enabled", "expected"), [ (8, False, 8), - (8, True, 2), - (4, True, 2), + (8, True, 4), # issue #109: clamp raised from 2 to 4, re-derived + Arm B-confirmed + (5, True, 4), + (4, True, 4), + (3, True, 3), # below the clamp -- passthrough, N=3 is a legal (unclamped) value too + (2, True, 2), (1, True, 1), # the clamp is a ceiling, never a floor (1, False, 1), ],