mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-08 03:08:13 +00:00
fix(parse): keep stable cache packs parallel (#3194)
Stable cache packs introduced by 9718e1247 often fit one worker job,
leaving most workers idle. Size jobs against live pool capacity and
reuse extraction queries per native grammar instead of recompiling
on every pack. Preserve recovery readiness across dispatches and
bound failed-thread termination acknowledgment.
Bisect confirms the scheduling regression. Controlled parsing of 951
TypeScript files falls from 56.39s to 26.66s with identical graph output;
peak RSS increases roughly 10%. Add scheduling and query reuse coverage
and repair recovery fixtures that assumed single-job dispatch.
Validation: build, typecheck, formatting and 229 focused tests pass.
Full suite was interrupted during lengthy native DB testing; full
release CI remains outstanding.
Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
This commit is contained in:
Gergő Magyar
GitHub
parent
a049b2dac6
commit
780cac7885
4 changed files with 193 additions and 23 deletions
|
|
@ -1543,6 +1543,10 @@ function reportWarning(message: string): void {
|
|||
}
|
||||
}
|
||||
|
||||
// Keep compiled queries across jobs in this worker. A language can select
|
||||
// multiple native grammars, so both grammar identity and query text matter.
|
||||
const compiledQueries = new WeakMap<object, Map<string, Parser.Query>>();
|
||||
|
||||
const processFileGroup = (
|
||||
files: ParseWorkerInput[],
|
||||
language: SupportedLanguages,
|
||||
|
|
@ -1553,7 +1557,14 @@ const processFileGroup = (
|
|||
let query: Parser.Query;
|
||||
try {
|
||||
const lang = parser.getLanguage();
|
||||
query = new Parser.Query(lang, queryString);
|
||||
let queries = compiledQueries.get(lang);
|
||||
if (!queries) {
|
||||
queries = new Map();
|
||||
compiledQueries.set(lang, queries);
|
||||
}
|
||||
const cached = queries.get(queryString);
|
||||
query = cached ?? new Parser.Query(lang, queryString);
|
||||
if (!cached) queries.set(queryString, query);
|
||||
} catch (err) {
|
||||
reportWarning(
|
||||
`Query compilation failed for ${language}: ${err instanceof Error ? err.message : String(err)}`,
|
||||
|
|
|
|||
|
|
@ -1316,9 +1316,16 @@ export const createWorkerPool = (
|
|||
}
|
||||
if (dispatchableItems.length === 0) return [];
|
||||
|
||||
// Stable cache packs can be much smaller than either job ceiling. Split
|
||||
// those packs across the live slots too, otherwise each serial dispatch
|
||||
// feeds only one worker. Keep both configured ceilings as upper bounds.
|
||||
const maxItemsPerJob = Math.min(
|
||||
poolOptions.subBatchSize,
|
||||
Math.max(1, Math.floor(dispatchableItems.length / activeSlots.size)),
|
||||
);
|
||||
const jobs = createJobs(
|
||||
dispatchableItems,
|
||||
poolOptions.subBatchSize,
|
||||
maxItemsPerJob,
|
||||
poolOptions.subBatchMaxBytes,
|
||||
poolOptions.subBatchIdleTimeoutMs,
|
||||
chunkHash,
|
||||
|
|
@ -1456,7 +1463,19 @@ export const createWorkerPool = (
|
|||
retireWorkerAfterTimeout(existing, workerIndex, reason);
|
||||
return;
|
||||
}
|
||||
await existing.terminate().catch(() => undefined);
|
||||
// Recovery must settle before dispatch returns, but a failed thread
|
||||
// may never acknowledge termination. Bound that wait as in shutdown.
|
||||
const termination = existing.terminate().then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
if (!(await settledWithin(termination, poolOptions.shutdownDrainMs))) {
|
||||
existing.unref?.();
|
||||
logger.warn(
|
||||
{ workerIndex, drainMs: poolOptions.shutdownDrainMs, reason },
|
||||
`Worker ${workerIndex} did not finish terminating within the shutdown drain; continuing recovery.`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
const replaceWorker = async (
|
||||
|
|
@ -1898,11 +1917,13 @@ export const createWorkerPool = (
|
|||
// (`error`, `exit`, msg-channel error). Bridges the per-job teardown
|
||||
// into the pool-level handleWorkerDeath recovery + breaker logic.
|
||||
const recoverAndResume = async (reason: string, excludePaths: readonly string[]) => {
|
||||
activeWorkers--;
|
||||
busySlots.delete(workerIndex);
|
||||
inFlightProgress[workerIndex] = 0;
|
||||
requeueRemainder(job, excludePaths);
|
||||
// Keep recovery in flight so another slot finishing cannot settle
|
||||
// this dispatch before the replacement is ready for the next one.
|
||||
await handleWorkerDeath(workerIndex, reason, excludePaths);
|
||||
activeWorkers--;
|
||||
if (stopped) return;
|
||||
// Slot may have been dropped or respawned. Kick the current slot
|
||||
// if still active, then wake any other idle live slots so the
|
||||
|
|
@ -1949,7 +1970,6 @@ export const createWorkerPool = (
|
|||
// is respawned (or dropped) and can dispatch the next
|
||||
// job deterministically.
|
||||
void (async () => {
|
||||
activeWorkers--;
|
||||
busySlots.delete(workerIndex);
|
||||
requeueRemainder(job, decision.excludePaths);
|
||||
await handleWorkerDeath(
|
||||
|
|
@ -1958,6 +1978,7 @@ export const createWorkerPool = (
|
|||
decision.excludePaths,
|
||||
'retire',
|
||||
);
|
||||
activeWorkers--;
|
||||
if (stopped) return;
|
||||
if (activeSlots.has(workerIndex)) runWorker(workerIndex);
|
||||
wakeIdleSlots();
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import {
|
|||
} from '../../src/core/ingestion/workers/worker-pool.js';
|
||||
import { pathToFileURL } from 'node:url';
|
||||
import { spawn } from 'node:child_process';
|
||||
import { createRequire } from 'node:module';
|
||||
import path from 'node:path';
|
||||
import fs from 'node:fs';
|
||||
import os from 'node:os';
|
||||
|
|
@ -245,10 +246,8 @@ describe('worker pool integration', () => {
|
|||
|
||||
const results = await pool.dispatch<any, any>(files);
|
||||
|
||||
// All 7 files fit one default sub-batch (size 200 / budget 8MB),
|
||||
// so the dispatch returns exactly one chunk result regardless of
|
||||
// pool size.
|
||||
expect(results).toHaveLength(1);
|
||||
// Small inputs must still split across the available workers.
|
||||
expect(results.length).toBeGreaterThan(1);
|
||||
|
||||
// Total files parsed should match input
|
||||
const totalParsed = results.reduce((sum: number, r: any) => sum + r.fileCount, 0);
|
||||
|
|
@ -793,6 +792,54 @@ describe('worker pool integration', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it.each([2, 4, 7, 9])(
|
||||
'uses available workers for a small %i-file cache pack',
|
||||
async (fileCount) => {
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-worker-small-pack-',
|
||||
`
|
||||
const { parentPort, threadId } = require('node:worker_threads');
|
||||
let paths = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
paths = msg.files.map((file) => file.path);
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: paths.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
} else if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({ type: 'result', data: { paths, threadId } });
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
pool = createWorkerPool(pathToFileURL(workerPath), 4, {
|
||||
subBatchMaxBytes: 256 * 1024,
|
||||
});
|
||||
|
||||
try {
|
||||
// Stable cache packs are often smaller than the byte budget. They must
|
||||
// still use the pool, including when a later dispatch has fewer files.
|
||||
for (const count of [fileCount, 1]) {
|
||||
const files = Array.from({ length: count }, (_, i) => ({
|
||||
path: `file-${i}.ts`,
|
||||
content: 'export const value = 1;',
|
||||
}));
|
||||
const progress: number[] = [];
|
||||
const results = await pool.dispatch<
|
||||
(typeof files)[number],
|
||||
{ paths: string[]; threadId: number }
|
||||
>(files, (completed) => progress.push(completed), `pack-${count}`);
|
||||
expect(new Set(results.map((result) => result.threadId)).size).toBe(Math.min(4, count));
|
||||
expect(results.flatMap((result) => result.paths)).toEqual(files.map((file) => file.path));
|
||||
expect(progress).toEqual([...progress].sort((a, b) => a - b));
|
||||
expect(progress.at(-1)).toBe(count);
|
||||
}
|
||||
} finally {
|
||||
await pool.terminate();
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it('bounds worker jobs by byte budget as well as file count', async () => {
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-worker-byte-budget-',
|
||||
|
|
@ -831,6 +878,69 @@ describe('worker pool integration', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it.skipIf(!hasDistWorker)(
|
||||
'reuses compiled queries across jobs while keeping TS and TSX grammars separate',
|
||||
async () => {
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-worker-query-cache-'));
|
||||
const workerPath = path.join(tempDir, 'worker.cjs');
|
||||
const parserPath = createRequire(import.meta.url).resolve('tree-sitter');
|
||||
fs.writeFileSync(
|
||||
workerPath,
|
||||
`
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
const Parser = require(${JSON.stringify(parserPath)});
|
||||
let queryCompilations = 0;
|
||||
Parser.Query = new Proxy(Parser.Query, {
|
||||
construct(target, args, newTarget) {
|
||||
queryCompilations++;
|
||||
return Reflect.construct(target, args, newTarget);
|
||||
},
|
||||
});
|
||||
const send = parentPort.postMessage.bind(parentPort);
|
||||
parentPort.postMessage = (message, ...args) => {
|
||||
if (message.type === 'result') message.data.queryCompilations = queryCompilations;
|
||||
return send(message, ...args);
|
||||
};
|
||||
import(${JSON.stringify(pathToFileURL(DIST_WORKER).href)});
|
||||
`,
|
||||
);
|
||||
pool = createWorkerPool(pathToFileURL(workerPath), 1, { workerReadyTimeoutMs: 30_000 });
|
||||
type QueryResult = {
|
||||
queryCompilations: number;
|
||||
fileCount: number;
|
||||
nodes: Array<{ properties: { name: string } }>;
|
||||
};
|
||||
try {
|
||||
const counts: number[] = [];
|
||||
for (const [extension, name] of [
|
||||
['ts', 'first'],
|
||||
['ts', 'second'],
|
||||
['tsx', 'view'],
|
||||
['tsx', 'otherView'],
|
||||
['ts', 'last'],
|
||||
]) {
|
||||
const file = {
|
||||
path: `${name}.${extension}`,
|
||||
content: `export function ${name}() { return ${extension === 'tsx' ? '<div />' : '1'}; }`,
|
||||
};
|
||||
const [result] = await pool.dispatch<typeof file, QueryResult>([file]);
|
||||
expect(result.fileCount).toBe(1);
|
||||
expect(result.nodes.map((node) => node.properties.name)).toContain(name);
|
||||
counts.push(result.queryCompilations);
|
||||
}
|
||||
expect(counts[0]).toBeGreaterThan(0);
|
||||
expect(counts[1]).toBe(counts[0]);
|
||||
expect(counts[2]).toBeGreaterThan(counts[1]);
|
||||
expect(counts[3]).toBe(counts[2]);
|
||||
expect(counts[4]).toBe(counts[2]);
|
||||
} finally {
|
||||
await pool.terminate();
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.skipIf(!hasDistWorker)('createWorkerPool with size 0 creates pool with zero workers', () => {
|
||||
const workerUrl = pathToFileURL(DIST_WORKER) as URL;
|
||||
const zeroPool = createWorkerPool(workerUrl, 0);
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ let workerInstances: FakeWorker[] = [];
|
|||
class FakeWorker extends EventEmitter {
|
||||
readonly seenMessages: unknown[] = [];
|
||||
|
||||
constructor() {
|
||||
constructor(startupExitCode?: number) {
|
||||
super();
|
||||
workerInstances.push(this);
|
||||
// Real Worker fires 'online' asynchronously after the runtime is ready;
|
||||
|
|
@ -67,6 +67,10 @@ class FakeWorker extends EventEmitter {
|
|||
// message instead — emit that too so replacement-worker tests don't
|
||||
// hit the WORKER_READY_TIMEOUT_MS budget (5s).
|
||||
queueMicrotask(() => {
|
||||
if (startupExitCode !== undefined) {
|
||||
this.emit('exit', startupExitCode);
|
||||
return;
|
||||
}
|
||||
this.emit('online');
|
||||
this.emit('message', { type: 'ready' });
|
||||
});
|
||||
|
|
@ -262,8 +266,8 @@ describe('worker pool resilience', () => {
|
|||
consecutiveFailureThreshold: 10,
|
||||
maxRespawnsPerSlot: 1,
|
||||
});
|
||||
// Slot 0 dies twice, exceeding budget=1; slot 1 succeeds with the
|
||||
// requeued remainder.
|
||||
// Separate dispatches target the first live slot twice, independently
|
||||
// of how multi-file packs are split across workers.
|
||||
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
|
||||
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
|
||||
nextActions.push({
|
||||
|
|
@ -272,9 +276,10 @@ describe('worker pool resilience', () => {
|
|||
result: { fileCount: 2 },
|
||||
});
|
||||
|
||||
await pool.dispatch([{ path: 'src/a.ts', content: '' }]);
|
||||
await pool.dispatch([{ path: 'src/b.ts', content: '' }]);
|
||||
expect(pool.getStats().activeSlots).toBe(1);
|
||||
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
|
||||
{ path: 'src/a.ts', content: '' },
|
||||
{ path: 'src/b.ts', content: '' },
|
||||
{ path: 'src/c.ts', content: '' },
|
||||
{ path: 'src/d.ts', content: '' },
|
||||
]);
|
||||
|
|
@ -495,18 +500,13 @@ describe('worker pool resilience', () => {
|
|||
await pool.terminate();
|
||||
});
|
||||
|
||||
it('drops slot when waitForWorkerOnline rejects (replaceWorker failure path)', async () => {
|
||||
it('drops slot when replacement readiness rejects', async () => {
|
||||
let factoryCallCount = 0;
|
||||
const pool = createWorkerPool(workerUrl, 2, {
|
||||
workerFactory: () => {
|
||||
factoryCallCount++;
|
||||
const worker = new FakeWorker();
|
||||
// Slot 0's initial worker is healthy; the replacement (3rd factory
|
||||
// call after slot 0 dies once) exits before emitting 'online'.
|
||||
if (factoryCallCount === 3) {
|
||||
// Override the queued 'online' microtask with an immediate 'exit'.
|
||||
queueMicrotask(() => worker.emit('exit', 1));
|
||||
}
|
||||
// The replacement exits INSTEAD OF reporting ready.
|
||||
const worker = new FakeWorker(factoryCallCount === 3 ? 1 : undefined);
|
||||
return worker as unknown as import('node:worker_threads').Worker;
|
||||
},
|
||||
consecutiveFailureThreshold: 10,
|
||||
|
|
@ -519,8 +519,9 @@ describe('worker pool resilience', () => {
|
|||
result: { fileCount: 2 },
|
||||
});
|
||||
|
||||
await pool.dispatch([{ path: 'src/a.ts', content: '' }]);
|
||||
expect(pool.getStats().activeSlots).toBe(1);
|
||||
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
|
||||
{ path: 'src/a.ts', content: '' },
|
||||
{ path: 'src/b.ts', content: '' },
|
||||
{ path: 'src/c.ts', content: '' },
|
||||
]);
|
||||
|
|
@ -532,6 +533,33 @@ describe('worker pool resilience', () => {
|
|||
await pool.terminate();
|
||||
});
|
||||
|
||||
it('finishes recovery when a failed worker never acknowledges termination', async () => {
|
||||
const pool = createWorkerPool(workerUrl, 2, {
|
||||
workerFactory: () => {
|
||||
const worker = new FakeWorker();
|
||||
if (workerInstances.length === 1) {
|
||||
worker.terminate = () => new Promise<number>(() => {});
|
||||
}
|
||||
return worker as unknown as import('node:worker_threads').Worker;
|
||||
},
|
||||
shutdownDrainMs: 10,
|
||||
});
|
||||
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
|
||||
nextActions.push({ kind: 'parse-ok', files: [{ path: 'src/good.ts' }] });
|
||||
try {
|
||||
const results = await pool.dispatch([
|
||||
{ path: 'src/bad.ts', content: '' },
|
||||
{ path: 'src/good.ts', content: '' },
|
||||
]);
|
||||
expect(results).toEqual([{ fileCount: 1 }]);
|
||||
expect(pool.getQuarantinedPaths()).toEqual(['src/bad.ts']);
|
||||
expect(pool.getStats().slotGenerations).toEqual([1, 0]);
|
||||
expect(pool.getStats().activeSlots).toBe(2);
|
||||
} finally {
|
||||
await pool.terminate();
|
||||
}
|
||||
}, 1000);
|
||||
|
||||
it('trips the breaker when all slots exhaust their respawn budget', async () => {
|
||||
const pool = createWorkerPool(workerUrl, 2, {
|
||||
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue