fix(db,resolution): WAL file cap + cgroup cache credit + pool/parse sizing corrections from the instrumented kernel-scale runs (#1335)
Four §7a.1 instrumented-run findings, each measured: 1. File-size trigger + truncate-at-barrier: a fully-backfilled WAL still grows the FILE without bound — the writer only restarts at frame 0 when a commit finds zero reader marks, which the instrumented run showed never happens (file marched 361→721MB through two COMPLETE backfills; 22GB by phase end). backpressure() now also trips at 4× the soft cap on raw file size and TRUNCATEs at the parked barrier; the timer path truncates opportunistically after complete backfills. Dubbo peak: 251MB → 69MB at the same 16MB valve; dumps byte-identical under aggressive folding. 2. cgroup memory credit: memory.current counts reclaimable page cache — a post-parse container read 57MB of headroom on a 6GB box and silently disabled the pool. inactive_file is credited back (the docker-stats working-set convention); the same run now reads a sane 4.4GB budget. 3. Pool at 2 cores reversed: sequential resolution measured FASTER than pooled-6-on-2 at kernel scale (853s vs 1,150s), and synthesis is Amdahl-bound by cFnPtrEdges (306s of 358s) so pooling it bought nothing. cpuCap = min(ap−1, 6), no floor: ap=2 → sequential is the fast path. 4. Parse floor of 2: one parse worker at a 2-cpuset measured 34% slower (493s vs 369s) — main + store-worker don't fill the second core. Floor restores the baseline (373.5s measured). Plus the observability §7a.1 burned three 25-minute cycles for: valve armed/fire/timer-pass/heartbeat lines, checkpoint-worker error capture, pool sizing decisions (incl. the disabled path), backpressure-hook presence — all behind CODEGRAPH_SYNTH_TIMINGS / CODEGRAPH_WAL_VALVE_DEBUG. Suite: 2,490 passed / 4 skipped (kernel required). Kernel-scale record runs with this build follow in the migration plan §7a.1. Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
8c1e821495
commit
ca88d3bd15
+11
-4
@@ -395,6 +395,7 @@ export class DatabaseConnection {
|
||||
const workerSource = `
|
||||
const { workerData, parentPort } = require('node:worker_threads');
|
||||
let row = null;
|
||||
let err = null;
|
||||
try {
|
||||
const { DatabaseSync } = require('node:sqlite');
|
||||
const db = new DatabaseSync(workerData.dbPath);
|
||||
@@ -402,10 +403,10 @@ export class DatabaseConnection {
|
||||
try {
|
||||
if (mode === 'TRUNCATE') db.exec('PRAGMA busy_timeout = 2000');
|
||||
row = db.prepare('PRAGMA wal_checkpoint(' + mode + ')').get();
|
||||
} catch {}
|
||||
} catch (e) { err = String(e && e.message || e); }
|
||||
try { db.close(); } catch {}
|
||||
} catch {}
|
||||
parentPort.postMessage({ row });
|
||||
} catch (e) { err = err || String(e && e.message || e); }
|
||||
parentPort.postMessage({ row, err });
|
||||
`;
|
||||
return await new Promise((resolve) => {
|
||||
let settled = false;
|
||||
@@ -416,7 +417,13 @@ export class DatabaseConnection {
|
||||
};
|
||||
try {
|
||||
const worker = new Worker(workerSource, { eval: true, workerData: { dbPath: this.dbPath, mode } });
|
||||
worker.once('message', (m: { row?: Record<string, number> | null }) => { void worker.terminate(); finish(m?.row ?? null); });
|
||||
worker.once('message', (m: { row?: Record<string, number> | null; err?: string | null }) => {
|
||||
if (m?.err && process.env.CODEGRAPH_WAL_VALVE_DEBUG) {
|
||||
console.error(`[wal-valve] checkpoint worker (${mode}): ${m.err}`);
|
||||
}
|
||||
void worker.terminate();
|
||||
finish(m?.row ?? null);
|
||||
});
|
||||
worker.once('error', () => { void worker.terminate(); finish(null); });
|
||||
worker.once('exit', () => finish(null));
|
||||
} catch {
|
||||
|
||||
+38
-2
@@ -50,6 +50,8 @@ import type { DatabaseConnection } from './index';
|
||||
const DEFAULT_WAL_VALVE_MB = 256;
|
||||
/** Hard cap = this × soft threshold; past it the writer pauses for a full backfill. */
|
||||
const HARD_CAP_MULTIPLIER = 2;
|
||||
/** File cap = this × soft threshold; past it the barrier also TRUNCATEs the file. */
|
||||
const FILE_CAP_MULTIPLIER = 4;
|
||||
/** Passes attempted per writer pause before giving up (a pinned reader could stall forever). */
|
||||
const MAX_PAUSED_BACKFILL_PASSES = 20;
|
||||
/** How often the timer looks at the WAL file size. */
|
||||
@@ -80,6 +82,7 @@ export class WalCheckpointValve {
|
||||
private sizeAtLastFullBackfill = 0;
|
||||
private readonly softBytes: number;
|
||||
private readonly hardBytes: number;
|
||||
private readonly fileCapBytes: number;
|
||||
|
||||
/**
|
||||
* Futility latch: consecutive backfill give-ups (a reader pinning the WAL)
|
||||
@@ -102,6 +105,7 @@ export class WalCheckpointValve {
|
||||
) {
|
||||
this.softBytes = softMb * 1024 * 1024;
|
||||
this.hardBytes = this.softBytes * HARD_CAP_MULTIPLIER;
|
||||
this.fileCapBytes = this.softBytes * FILE_CAP_MULTIPLIER;
|
||||
// CODEGRAPH_WAL_VALVE_DEBUG=1 surfaces valve decisions to stderr without
|
||||
// needing the caller's verbose plumbing — the observability gap that let
|
||||
// §7a.1 run 1 fail silently (give-ups were verbose-gated and invisible).
|
||||
@@ -124,7 +128,19 @@ export class WalCheckpointValve {
|
||||
/** Begin watching the WAL. Idempotent; the timer never holds the loop open. */
|
||||
start(): void {
|
||||
if (this.timer) return;
|
||||
this.timer = setInterval(() => this.check(), this.intervalMs);
|
||||
// One armed line per run under either diagnostics env: §7a.1's failed
|
||||
// kernel-scale runs burned three 25-minute cycles before "is the valve
|
||||
// even alive?" could be answered.
|
||||
if (process.env.CODEGRAPH_SYNTH_TIMINGS || process.env.CODEGRAPH_WAL_VALVE_DEBUG) {
|
||||
console.error(`[wal-valve] armed soft=${this.mb(this.softBytes)} hard=${this.mb(this.hardBytes)} wal=${this.mb(this.db.getWalSizeBytes())}`);
|
||||
}
|
||||
let ticks = 0;
|
||||
this.timer = setInterval(() => {
|
||||
if ((++ticks % 15) === 0) {
|
||||
this.log(`alive: wal=${this.mb(this.db.getWalSizeBytes())} baseline=${this.mb(this.sizeAtLastFullBackfill)} inflight=${this.inflight ? 'y' : 'n'} paused=${this.pause ? 'y' : 'n'}`);
|
||||
}
|
||||
this.check();
|
||||
}, this.intervalMs);
|
||||
this.timer.unref?.();
|
||||
}
|
||||
|
||||
@@ -150,7 +166,15 @@ export class WalCheckpointValve {
|
||||
backpressure(): Promise<void> | null {
|
||||
if (this.pause) return this.pause;
|
||||
if (Date.now() < this.futileUntil) return null; // pinned reader — parking is churn, not progress
|
||||
if (this.growthBytes() <= this.hardBytes) return null;
|
||||
// Two independent triggers:
|
||||
// - growth: un-backfilled BACKLOG past the hard cap (the original valve).
|
||||
// - file size: a WAL can stay fully backfilled and still grow without
|
||||
// bound — the writer only restarts at frame 0 if a commit finds no
|
||||
// reader marks, which §7a.1's instrumented run showed never happens
|
||||
// in practice (file marched 361→721MB through two COMPLETE
|
||||
// backfills). Past the file cap, park and TRUNCATE at the barrier —
|
||||
// the backfill part is instant when the backlog is already folded.
|
||||
if (this.growthBytes() <= this.hardBytes && this.db.getWalSizeBytes() <= this.fileCapBytes) return null;
|
||||
this.log(`backpressure: wal=${this.mb(this.db.getWalSizeBytes())} baseline=${this.mb(this.sizeAtLastFullBackfill)} — pausing writer for full backfill`);
|
||||
const t0 = Date.now();
|
||||
this.pause = this.backfillFully().finally(() => {
|
||||
@@ -226,9 +250,11 @@ export class WalCheckpointValve {
|
||||
}
|
||||
|
||||
private fire(): void {
|
||||
this.log(`fire: wal=${this.mb(this.db.getWalSizeBytes())} baseline=${this.mb(this.sizeAtLastFullBackfill)}`);
|
||||
const p = this.db
|
||||
.checkpointWalPassive()
|
||||
.then((res) => {
|
||||
this.log(`timer pass: ${res ? `busy=${res.busy} log=${res.log} checkpointed=${res.checkpointed}` : 'null (machinery unavailable)'} wal=${this.mb(this.db.getWalSizeBytes())}`);
|
||||
// Full backfill (busy 0, every log frame checkpointed) ⇒ the writer's
|
||||
// next commit wraps the WAL; the file's current size becomes the new
|
||||
// growth baseline. A partial pass (writer appended during it, or a
|
||||
@@ -237,6 +263,16 @@ export class WalCheckpointValve {
|
||||
// SQLite reports log = checkpointed = -1, which is harmless here.
|
||||
if (res && res.busy === 0 && res.log === res.checkpointed) {
|
||||
this.sizeAtLastFullBackfill = this.db.getWalSizeBytes();
|
||||
// Opportunistic file chop while everything is folded: best-effort —
|
||||
// an active writer/reader turns it into a busy no-op, and the
|
||||
// barrier-path truncate (backpressure file cap) remains the
|
||||
// deterministic bound.
|
||||
if (this.db.getWalSizeBytes() > this.softBytes * 2) {
|
||||
return this.db.checkpointWalTruncate().then((t) => {
|
||||
if (t) this.log(`timer truncate: busy=${t.busy} wal=${this.mb(this.db.getWalSizeBytes())}`);
|
||||
this.sizeAtLastFullBackfill = this.db.getWalSizeBytes();
|
||||
});
|
||||
}
|
||||
}
|
||||
})
|
||||
.catch(() => { /* best-effort */ })
|
||||
|
||||
@@ -1619,8 +1619,11 @@ export class ExtractionOrchestrator {
|
||||
// with cores from availableParallelism — cpuset/affinity-honest, where
|
||||
// os.cpus() enumerates the host's CPUs and spawned 8 wasm workers (and
|
||||
// their grammar heaps) inside a 2-CPU container for zero extra
|
||||
// throughput (§7a.1).
|
||||
const poolSize = resolveParsePoolSize(process.env.CODEGRAPH_PARSE_WORKERS, os.availableParallelism());
|
||||
// throughput (§7a.1). Floored so a 2-core box still gets 2 workers:
|
||||
// parse is worker-side CPU, and 1 worker measured 34% slower than the
|
||||
// old oversubscribed pool on the kernel-scale 2-cpuset envelope
|
||||
// (493s vs 369s) — main + store-worker don't fill the second core.
|
||||
const poolSize = resolveParsePoolSize(process.env.CODEGRAPH_PARSE_WORKERS, Math.max(3, os.availableParallelism()));
|
||||
// Read each needed grammar's WASM ONCE here and hand the bytes to every
|
||||
// worker, so spawns/respawns load grammars from memory instead of
|
||||
// re-reading them from disk (#1231: on an HDD, respawn re-reads amplify
|
||||
|
||||
@@ -1358,6 +1358,10 @@ export class ReferenceResolver {
|
||||
// window to fire; see ./cooperative-yield.
|
||||
const maybeYield = createYielder();
|
||||
|
||||
if (process.env.CODEGRAPH_SYNTH_TIMINGS) {
|
||||
console.error(`[pool-timing] backpressure hook: ${parallel?.backpressure ? 'present' : 'absent'}`);
|
||||
}
|
||||
|
||||
await this.warmCachesYielding(maybeYield);
|
||||
|
||||
const total = this.queries.getUnresolvedReferencesCount();
|
||||
|
||||
@@ -24,9 +24,26 @@ function readCgroupBytes(path: string): number | null {
|
||||
}
|
||||
}
|
||||
|
||||
/** `inactive_file` from a cgroup memory.stat file — reclaimable page cache. */
|
||||
function readInactiveFile(statPath: string): number {
|
||||
try {
|
||||
const m = /^inactive_file (\d+)$/m.exec(fs.readFileSync(statPath, 'utf8'));
|
||||
return m ? Number.parseInt(m[1]!, 10) : 0;
|
||||
} catch {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Available headroom under the cgroup memory limit (v2 then v1), or null
|
||||
* when uncontained (no limit, non-Linux, or unreadable). Never throws.
|
||||
*
|
||||
* Reclaimable page cache (`inactive_file`) is credited back: `memory.current`
|
||||
* counts it as usage, but the kernel reclaims it on demand — after a bulk
|
||||
* parse the cache is stuffed with the DB's own pages, and the naive
|
||||
* `max − current` read 57MB of headroom on a 6GB container and silently
|
||||
* disabled the resolver pool (§7a.1 diagnostic run). This is the same
|
||||
* working-set convention `docker stats` uses.
|
||||
*/
|
||||
export function cgroupMemoryAvailable(): number | null {
|
||||
if (process.platform !== 'linux') return null;
|
||||
@@ -34,7 +51,8 @@ export function cgroupMemoryAvailable(): number | null {
|
||||
const v2Max = readCgroupBytes('/sys/fs/cgroup/memory.max');
|
||||
if (v2Max !== null) {
|
||||
const current = readCgroupBytes('/sys/fs/cgroup/memory.current') ?? 0;
|
||||
return Math.max(0, v2Max - current);
|
||||
const reclaimable = readInactiveFile('/sys/fs/cgroup/memory.stat');
|
||||
return Math.max(0, v2Max - Math.max(0, current - reclaimable));
|
||||
}
|
||||
// v1
|
||||
const v1Limit = readCgroupBytes('/sys/fs/cgroup/memory/memory.limit_in_bytes');
|
||||
@@ -42,7 +60,8 @@ export function cgroupMemoryAvailable(): number | null {
|
||||
// anything at or beyond half the address-space-ish range as uncontained.
|
||||
if (v1Limit !== null && v1Limit < 2 ** 60) {
|
||||
const usage = readCgroupBytes('/sys/fs/cgroup/memory/memory.usage_in_bytes') ?? 0;
|
||||
return Math.max(0, v1Limit - usage);
|
||||
const reclaimable = readInactiveFile('/sys/fs/cgroup/memory/memory.stat');
|
||||
return Math.max(0, v1Limit - Math.max(0, usage - reclaimable));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -95,7 +95,12 @@ export class ResolverPool {
|
||||
return Math.min(n, 16);
|
||||
}
|
||||
}
|
||||
const cpuCap = Math.max(2, Math.min(opts.availableParallelism - 1, 6));
|
||||
// No floor: at ap=2 the pool LOSES to sequential outright — measured on
|
||||
// the kernel-scale 2-cpuset envelope: resolution 853s sequential vs
|
||||
// 1,150s pooled-6-on-2 (§7a.1), and synthesis is Amdahl-bound by its
|
||||
// dominant pass (cFnPtrEdges 306s of 358s) so pooling it bought nothing.
|
||||
// ap−1 < 2 ⇒ sequential is the fast path, not a fallback.
|
||||
const cpuCap = Math.min(opts.availableParallelism - 1, 6);
|
||||
const perWorker = Math.min(Math.max(opts.dbSizeBytes * 0.2, 256 * 1024 * 1024), 1.5 * 1024 * 1024 * 1024);
|
||||
const memCap = Math.floor((opts.memoryBudget * 0.7) / perWorker);
|
||||
const size = Math.min(cpuCap, memCap);
|
||||
@@ -117,18 +122,22 @@ export class ResolverPool {
|
||||
try {
|
||||
dbSizeBytes = fs.statSync(dbPath).size;
|
||||
} catch { /* fresh/missing file — the 256MB per-worker floor applies */ }
|
||||
const ap = os.availableParallelism();
|
||||
const budget = memoryBudgetBytes();
|
||||
const size = ResolverPool.resolvePoolSize({
|
||||
explicit: process.env.CODEGRAPH_RESOLVE_WORKERS,
|
||||
availableParallelism: os.availableParallelism(),
|
||||
memoryBudget: memoryBudgetBytes(),
|
||||
availableParallelism: ap,
|
||||
memoryBudget: budget,
|
||||
dbSizeBytes,
|
||||
});
|
||||
if (size === null) return null;
|
||||
// Both outcomes log under SYNTH_TIMINGS — a silent null is how §7a.1's
|
||||
// diagnostic run hid the memory-term misfire for a whole 25-minute cycle.
|
||||
if (process.env.CODEGRAPH_SYNTH_TIMINGS) {
|
||||
console.error(
|
||||
`[pool-timing] pool size=${size} (ap=${os.availableParallelism()} budget=${Math.round(memoryBudgetBytes() / 1024 / 1024)}MB db=${Math.round(dbSizeBytes / 1024 / 1024)}MB)`
|
||||
`[pool-timing] pool ${size === null ? 'disabled' : `size=${size}`} (ap=${ap} budget=${Math.round(budget / 1024 / 1024)}MB db=${Math.round(dbSizeBytes / 1024 / 1024)}MB)`
|
||||
);
|
||||
}
|
||||
if (size === null) return null;
|
||||
try {
|
||||
return new ResolverPool(workerScript, dbPath, projectRoot, size);
|
||||
} catch {
|
||||
|
||||
Reference in New Issue
Block a user