perf(resolution): worker connection recycling — WAL-depth writes-under-readers fix, superphase −11.4% at 8c (#1362)
The §7a.6 anomaly probed to its mechanism with five discriminating runs (§7a.7 table): main-thread B-tree writes triple under attached readers because READERS PIN WAL checkpoint progress — the deep WAL taxes every writer page operation (deletes 42.6s pool-off vs 118.8s pool-4 on identical hardware; an aggressive 64MB valve recovers the writes but overpays +129s in full-park folds; the v2 cache resurrection was falsified — long-tail name traffic is uncacheable at any capacity). Fix: workers close and reopen their read-only connections every 8 batches at the double-buffer's worker-idle boundary (ResolverPool.recycleWorkers + QueryBuilder.rebind + a cadence call). Reopens are sub-millisecond, resolver caches survive (only prepared statements re-prepare), and the existing checkpoints advance instead of parking. Failed recycle downgrades to sequential, same as a failed fan-out. Measured (8c pool-4, linux v7.2-rc2, cadence 25 → 8 iterated): resolution superphase 715.0 → 633.6s (−11.4%), envelope best 14.8min, recreate 59.7 → 45.3s. Byte-neutral everywhere: git dumps byte-identical old-vs-new, linux dump sha 6dd1185b reproduced (10,446,478 lines), counts 2,049,153/6,413,518, suite 2517 green. 2c unchanged by construction (no pool → no recycling). Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
5955d04c97
commit
971a5a0483
+34
-1
@@ -1408,7 +1408,7 @@ export class ReferenceResolver {
|
||||
// these counters name where the other ~340s goes (reads, edge build+insert,
|
||||
// deletes/marks, the per-batch count guard).
|
||||
const loopProf: Record<string, number> | null = process.env.CODEGRAPH_RESOLVE_PROFILE
|
||||
? { read: 0, settle: 0, backpressure: 0, createEdges: 0, insertEdges: 0, deletes: 0, marks: 0, countGuard: 0 }
|
||||
? { read: 0, settle: 0, backpressure: 0, recycle: 0, createEdges: 0, insertEdges: 0, deletes: 0, marks: 0, countGuard: 0 }
|
||||
: null;
|
||||
const lp = (k: string, t0: number): void => { if (loopProf) loopProf[k] = (loopProf[k] ?? 0) + (Date.now() - t0); };
|
||||
let tLp = 0;
|
||||
@@ -1450,6 +1450,14 @@ export class ReferenceResolver {
|
||||
// pending rows forward.
|
||||
let prevRemaining = Number.POSITIVE_INFINITY;
|
||||
|
||||
// Cadence for the worker connection recycling below — ~8 batches
|
||||
// ≈ 40k refs between recycles keeps the WAL shallow at kernel scale
|
||||
// while a small sync never recycles at all. (25 recovered only half
|
||||
// the write tax — the WAL re-deepened between recycles; reopens are
|
||||
// sub-millisecond so the shorter cadence is ~free.)
|
||||
const RECYCLE_EVERY_BATCHES = 8;
|
||||
let batchesSinceRecycle = 0;
|
||||
|
||||
// Fan-out result of ResolverPool.resolveBatch, settled (never rejecting)
|
||||
// so a fan-out begun before the previous batch's persist can't produce an
|
||||
// unhandled rejection while it waits to be awaited.
|
||||
@@ -1557,6 +1565,31 @@ export class ReferenceResolver {
|
||||
if (bp) await bp;
|
||||
lp('backpressure', tLp);
|
||||
|
||||
// Recycle the workers' read connections periodically at this same
|
||||
// worker-idle boundary (batch k settled, batch k+1 not yet fanned
|
||||
// out): a long-lived reader pins WAL checkpoint progress, and the
|
||||
// deep WAL that accumulates behind it taxes the writer's OWN page
|
||||
// operations — the §7a.6 writes-under-readers finding (deletes
|
||||
// 42.6s → 118.8s from 0 to 4 attached readers; an aggressive valve
|
||||
// recovered the writes but paid +129s in full-park folds). Releasing
|
||||
// the read marks every ~25 batches lets the existing checkpoints
|
||||
// advance instead, at ~milliseconds of reopen cost. A failed recycle
|
||||
// downgrades to sequential permanently, same as a failed fan-out.
|
||||
if (pool && poolReady && ++batchesSinceRecycle >= RECYCLE_EVERY_BATCHES) {
|
||||
batchesSinceRecycle = 0;
|
||||
tLp = Date.now();
|
||||
try {
|
||||
await pool.recycleWorkers();
|
||||
} catch (err) {
|
||||
logDebug('Worker connection recycle failed; falling back to sequential', {
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
await pool.destroy().catch(() => undefined);
|
||||
pool = null;
|
||||
}
|
||||
lp('recycle', tLp);
|
||||
}
|
||||
|
||||
// Persist in bounded sub-transactions with yields between: a whole
|
||||
// batch's edge insert / keyed deletes are otherwise one solid
|
||||
// synchronous span each on a multi-GB index, sitting BETWEEN the
|
||||
|
||||
@@ -63,6 +63,7 @@ export class ResolverPool {
|
||||
private nextId = 0;
|
||||
private waiters = new Map<number, { resolve: (r: ChunkResult) => void; reject: (e: Error) => void }>();
|
||||
private synthWaiters = new Map<number, { resolve: (r: SynthPassResult) => void; reject: (e: Error) => void }>();
|
||||
private recycleWaiters = new Map<number, () => void>();
|
||||
private failed: Error | null = null;
|
||||
|
||||
/**
|
||||
@@ -174,6 +175,10 @@ export class ResolverPool {
|
||||
const waiter = this.synthWaiters.get(msg.id);
|
||||
this.synthWaiters.delete(msg.id);
|
||||
waiter?.resolve({ edges: msg.edges ?? [], ms: msg.ms ?? 0 });
|
||||
} else if (msg.type === 'recycled' && msg.id !== undefined) {
|
||||
const waiter = this.recycleWaiters.get(msg.id);
|
||||
this.recycleWaiters.delete(msg.id);
|
||||
waiter?.();
|
||||
} else if (msg.type === 'error') {
|
||||
pw.busy--;
|
||||
const err = new Error(`resolver worker: ${msg.message}`);
|
||||
@@ -211,6 +216,10 @@ export class ResolverPool {
|
||||
this.waiters.clear();
|
||||
for (const [, waiter] of this.synthWaiters) waiter.reject(this.failed);
|
||||
this.synthWaiters.clear();
|
||||
// Pending recycles resolve rather than reject: their per-call timeout
|
||||
// owns rejection, and the recycle caller checks this.failed next round.
|
||||
for (const [, done] of this.recycleWaiters) done();
|
||||
this.recycleWaiters.clear();
|
||||
}
|
||||
|
||||
/** Whether this batch is worth fanning out. */
|
||||
@@ -273,6 +282,43 @@ export class ResolverPool {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Ask every worker to close and reopen its read-only connection, and wait
|
||||
* for all acks. MUST be called only at the pool-idle boundary (all fanned
|
||||
* chunks settled, next batch not yet dispatched) — the workers close their
|
||||
* connections in place. Why: a long-lived reader pins WAL checkpoint
|
||||
* progress, and the deep WAL behind it taxes every main-thread B-tree
|
||||
* page operation (writes-under-readers, plan §7a.6 — deletes 42.6→118.8s
|
||||
* from 0 to 4 attached readers). Releasing the read marks periodically
|
||||
* lets the existing checkpoints advance, keeping the WAL shallow WITHOUT
|
||||
* the full-park folds an aggressive valve pays (+129s measured at 64MB).
|
||||
* A recycle failure fails the pool — the caller's sequential fallback
|
||||
* covers the rest of the run.
|
||||
*/
|
||||
async recycleWorkers(): Promise<void> {
|
||||
if (this.failed) throw this.failed;
|
||||
await Promise.all(
|
||||
this.workers.map(
|
||||
(pw) =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
const id = this.nextId++;
|
||||
const t = setTimeout(() => {
|
||||
if (this.recycleWaiters.delete(id)) {
|
||||
const err = new Error('resolver worker recycle timed out');
|
||||
this.fail(err);
|
||||
reject(err);
|
||||
}
|
||||
}, 10_000);
|
||||
this.recycleWaiters.set(id, () => {
|
||||
clearTimeout(t);
|
||||
resolve();
|
||||
});
|
||||
pw.worker.postMessage({ type: 'recycle', id });
|
||||
})
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
async destroy(): Promise<void> {
|
||||
await Promise.all(
|
||||
this.workers.map(
|
||||
|
||||
@@ -39,15 +39,19 @@ let resolver: ReferenceResolver | null = null;
|
||||
|
||||
type InMessage =
|
||||
| { type: 'open'; dbPath: string; projectRoot: string }
|
||||
| { type: 'recycle'; id: number }
|
||||
| { type: 'resolve'; id: number; refs: UnresolvedReference[] }
|
||||
| { type: 'synth'; id: number; pass: string }
|
||||
| { type: 'close' };
|
||||
|
||||
let dbPath: string | null = null;
|
||||
|
||||
port.on('message', (msg: InMessage) => {
|
||||
try {
|
||||
switch (msg.type) {
|
||||
case 'open': {
|
||||
const tOpen = Date.now();
|
||||
dbPath = msg.dbPath;
|
||||
const created = createDatabase(msg.dbPath, { readOnly: true });
|
||||
db = created.db;
|
||||
db.pragma('busy_timeout = 5000');
|
||||
@@ -60,6 +64,25 @@ port.on('message', (msg: InMessage) => {
|
||||
port.postMessage({ type: 'ready' });
|
||||
break;
|
||||
}
|
||||
case 'recycle': {
|
||||
// Close and reopen the read-only connection so the WAL checkpoints
|
||||
// the writer runs can advance past this reader's mark (see
|
||||
// QueryBuilder.rebind). Everything above the connection survives —
|
||||
// the resolver keeps its warm caches; prepared statements re-prepare
|
||||
// lazily. Runs only at the pool-idle boundary, so no query is in
|
||||
// flight on this connection.
|
||||
if (!queries || !dbPath) throw new Error('resolver-worker: recycle before open');
|
||||
try {
|
||||
db?.close();
|
||||
} catch { /* already closed */ }
|
||||
const reopened = createDatabase(dbPath, { readOnly: true });
|
||||
db = reopened.db;
|
||||
db.pragma('busy_timeout = 5000');
|
||||
db.pragma('cache_size = -32000');
|
||||
queries.rebind(db);
|
||||
port.postMessage({ type: 'recycled', id: msg.id });
|
||||
break;
|
||||
}
|
||||
case 'resolve': {
|
||||
if (!resolver) throw new Error('resolver-worker: resolve before open');
|
||||
const tRes = Date.now();
|
||||
|
||||
Reference in New Issue
Block a user