mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-09-10 22:43:40 +00:00
perf(parse): batch cache packs into one dispatch round (#3196)
* perf(parse): batch cache packs into one dispatch round `WorkerPool.dispatch` is a barrier, so dispatching one parse-cache pack at a time leaves most slots idle for every round-trip. Packs are keyed by `(language, hash(path) % 128)`, so the byte budget rarely binds: this repo produces 1285 packs where the budget alone needs 16, and 549 of those hold a single file. In a real analyze, 76 of 221 dispatched chunks carried one file and cost 15.3s — 20% of the parse phase for 3.4% of the files. Chunks now accumulate into a round bounded by `GITNEXUS_PARSE_ROUND_BYTES` of cache-missing source (default: the chunk byte budget) and go out through a new `WorkerPool.dispatchGroups`. Jobs are still cut at pack boundaries, so each job carries exactly one `chunkHash` and every result stays attributable to the pack whose cache key owns it. Cache hits ride along as round entries, and rounds drain in `chunkIdx` order, so deferred aggregation stays deterministic. Cold `analyze --index-only` on this repo (2234 parseable files, 16 workers): 110.3s -> 70.5s total, parse phase 74.0s -> 40.5s, 221 dispatches -> 15. Graph output is unchanged: 51,286 nodes / 163,092 edges / 2106 clusters / 759 flows in both arms. Peak main-thread RSS 3372MB -> 3487MB (+3.4%). `dispatchGroups` also claims the pool synchronously and rejects a concurrent call. Two overlapping dispatches hand the same slots out twice and both stall; the first version of this change did exactly that, and the only symptom was every worker idle-timing out ~10s later with no indication of the cause. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(parse): bound what an open round holds, not just what it dispatches Follow-up to the review of #3196. Three reviewers independently found the same defect: `roundMissBytes` was the only in-loop close condition, but cache HITS were queued into the same round without contributing to it. A warm re-analyze misses nothing, so no round ever closed and every chunk's source plus its cached worker output stayed resident until the tail drain — the #2649 heap failure shape on a large repo. - Hit entries now carry a file COUNT, not the file array, so a replayed chunk never pins its source text. `applyChunkResults` only ever read `.length`. - Track `roundBufferedBytes` across hits and misses and close on either cap. Verified on a warm run: with the cap, draining starts as soon as 2MB is buffered; without it all 221 merges land in the final 10% of the phase. - Warm progress no longer freezes at the phase floor. `filesParsedSoFar` only advances at drain, so a new `queuedFilesSoFar` feeds the progress events while `filesParsedSoFar` stays the merge-accurate throughput number. - A throw from `drainRound` used to unwind straight to `terminate()` while the next round's workers were still busy — the #2432 mid-N-API abort hazard. Settle the in-flight round first, then propagate. - `dispatchGroups` returns one array per group; assert that length instead of `?? []`, which turned a contract break into a silently empty chunk. - Collapse `PendingWorkerChunk` into the `miss` RoundEntry it duplicated. - Repair two stale doc comments: `dispatch`'s JSDoc had been orphaned onto `dispatchGroups`, and `dispatchChunkParse` still described chunk overlap that now lives in parse-impl's round machinery. - New test: a round mixing a cache hit and a cache miss. `drainRound` walks entries in chunkIdx order but pulls results on a separate cursor, and no existing test put both kinds in one round with content assertions. Cold analyze unchanged: 71.3s, 15 rounds, 51,286 nodes / 163,092 edges. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Gergő Magyar
GitHub
parent
780cac7885
commit
8f006bd759
7 changed files with 891 additions and 176 deletions
|
|
@ -7,6 +7,7 @@ import { accumulateExportedTypesFromParsedNode, type ExportedTypeMap } from './c
|
|||
|
||||
import type { ParsedFile } from 'gitnexus-shared';
|
||||
import { WorkerPool } from './workers/worker-pool.js';
|
||||
import type { DispatchGroup } from './workers/worker-pool.js';
|
||||
import type { SkippedPath } from './workers/clone-safety.js';
|
||||
import type { CfgSkipCounts } from './cfg/collect.js';
|
||||
import { logger } from '../logger.js';
|
||||
|
|
@ -206,12 +207,12 @@ export const mergeChunkResults = (
|
|||
};
|
||||
|
||||
/**
|
||||
* Dispatch a chunk's files to the worker pool and return the RAW per-worker
|
||||
* results, WITHOUT merging them into the graph. Split out from
|
||||
* {@link processParsing} so the parse loop can overlap one chunk's
|
||||
* merge (main-thread, via {@link mergeChunkResults}) with the NEXT chunk's
|
||||
* worker parse — the merge is the only remaining serial main-thread step once
|
||||
* ParsedFile serialization moved into the workers (#worker-idle pipelining).
|
||||
* Dispatch ONE chunk's files to the worker pool and return the RAW per-worker
|
||||
* results, WITHOUT merging them into the graph. A thin single-group wrapper
|
||||
* over {@link dispatchChunkParseRound}, used by {@link processParsing}'s
|
||||
* one-shot path. The chunk-to-chunk overlap this once described now lives in
|
||||
* `parse-impl.ts` at ROUND granularity (`startRound` / `drainRound` /
|
||||
* `closeRound`), which batches several chunks into one dispatch.
|
||||
* Returns `[]` for an all-unparseable chunk (the caller merges `[]` → empty).
|
||||
*/
|
||||
export const dispatchChunkParse = async (
|
||||
|
|
@ -227,26 +228,56 @@ export const dispatchChunkParse = async (
|
|||
*/
|
||||
chunkHash?: string,
|
||||
): Promise<ParseWorkerResult[]> => {
|
||||
const parseableFiles: ParseWorkerInput[] = [];
|
||||
for (const file of files) {
|
||||
const lang = getLanguageFromFilename(file.path);
|
||||
if (lang) parseableFiles.push({ path: file.path, content: file.content });
|
||||
}
|
||||
if (parseableFiles.length === 0) return [];
|
||||
|
||||
const total = files.length;
|
||||
const chunkResults = await workerPool.dispatch<ParseWorkerInput, ParseWorkerResult>(
|
||||
parseableFiles,
|
||||
(filesProcessed) => {
|
||||
onFileProgress?.(Math.min(filesProcessed, total), total, 'Parsing...');
|
||||
},
|
||||
chunkHash,
|
||||
const [chunkResults = []] = await dispatchChunkParseRound(
|
||||
[{ items: files, chunkHash }],
|
||||
workerPool,
|
||||
onFileProgress,
|
||||
);
|
||||
|
||||
// Capture raw results for the incremental parse cache before merging.
|
||||
if (outRawResults) {
|
||||
for (const r of chunkResults) outRawResults.push(r);
|
||||
}
|
||||
return chunkResults;
|
||||
};
|
||||
|
||||
/**
|
||||
* Dispatch SEVERAL parse-cache chunks as one pool round and return their raw
|
||||
* results, one array per input group in input order.
|
||||
*
|
||||
* `WorkerPool.dispatch` is a barrier, so one round-trip per chunk leaves most
|
||||
* slots idle whenever a chunk is smaller than the pool — which stable
|
||||
* `(language, hash(path) % 128)` packs usually are. Batching chunks into one
|
||||
* `dispatchGroups` call removes those barriers; jobs are still cut at chunk
|
||||
* boundaries, so every result stays attributable to the chunk whose cache key
|
||||
* owns it.
|
||||
*/
|
||||
export const dispatchChunkParseRound = async (
|
||||
groups: ReadonlyArray<{
|
||||
items: { path: string; content: string }[];
|
||||
chunkHash?: string;
|
||||
}>,
|
||||
workerPool: WorkerPool,
|
||||
onFileProgress?: FileProgressCallback,
|
||||
): Promise<ParseWorkerResult[][]> => {
|
||||
const dispatchGroups: DispatchGroup<ParseWorkerInput>[] = groups.map((group) => {
|
||||
const items: ParseWorkerInput[] = [];
|
||||
for (const file of group.items) {
|
||||
const lang = getLanguageFromFilename(file.path);
|
||||
if (lang) items.push({ path: file.path, content: file.content });
|
||||
}
|
||||
return { items, chunkHash: group.chunkHash };
|
||||
});
|
||||
const total = groups.reduce((sum, group) => sum + group.items.length, 0);
|
||||
if (dispatchGroups.every((group) => group.items.length === 0)) return groups.map(() => []);
|
||||
|
||||
const perGroup = await workerPool.dispatchGroups<ParseWorkerInput, ParseWorkerResult>(
|
||||
dispatchGroups,
|
||||
(filesProcessed) => {
|
||||
onFileProgress?.(Math.min(filesProcessed, total), total, 'Parsing...');
|
||||
},
|
||||
);
|
||||
const chunkResults = perGroup.flat();
|
||||
|
||||
// Skipped-language telemetry (worker output, independent of the merge).
|
||||
const skippedLanguages = new Map<string, number>();
|
||||
|
|
@ -311,7 +342,7 @@ export const dispatchChunkParse = async (
|
|||
}
|
||||
|
||||
onFileProgress?.(total, total, 'done');
|
||||
return chunkResults;
|
||||
return perGroup;
|
||||
};
|
||||
|
||||
// ============================================================================
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ import {
|
|||
enrichExportedTypeMap,
|
||||
type BindingEntry,
|
||||
} from '../binding-accumulator.js';
|
||||
import { mergeChunkResults, dispatchChunkParse } from '../parsing-processor.js';
|
||||
import { mergeChunkResults, dispatchChunkParseRound } from '../parsing-processor.js';
|
||||
import {
|
||||
fileContentHash,
|
||||
computeChunkHash,
|
||||
|
|
@ -216,6 +216,28 @@ const TARGET_JOBS_PER_WORKER = 3;
|
|||
/** Floor for a derived sub-batch so jobs don't shrink to per-file IPC churn. */
|
||||
const MIN_SUB_BATCH_BYTES = 256 * 1024;
|
||||
|
||||
/**
|
||||
* Source bytes of cache-missing chunks allowed in flight in one pool round.
|
||||
*
|
||||
* A `dispatch` is a barrier, so one round-trip per cache pack leaves most slots
|
||||
* idle: packs are keyed by `(language, hash(path) % 128)` and routinely land far
|
||||
* under {@link DEFAULT_CHUNK_BYTE_BUDGET} (this repo: 1285 packs where the byte
|
||||
* budget alone needs 16, 549 of them holding a single file). Rounds batch packs
|
||||
* into one `dispatchGroups` call without touching pack identity.
|
||||
*
|
||||
* This is the in-flight cap, the same role Piscina's `maxQueue` plays: bigger
|
||||
* rounds remove more barriers but hold more file content and more un-merged
|
||||
* worker output on the main thread at once. Defaulting to one chunk budget
|
||||
* keeps in-flight source bytes at the magnitude the loop already prefetched
|
||||
* (`parseChunkConcurrency`, 2 chunks ahead). Override via
|
||||
* `GITNEXUS_PARSE_ROUND_BYTES`.
|
||||
*/
|
||||
function resolveParseRoundByteBudget(options?: PipelineOptions): number {
|
||||
const env = Number(process.env.GITNEXUS_PARSE_ROUND_BYTES);
|
||||
if (Number.isFinite(env) && env > 0) return env;
|
||||
return resolveChunkByteBudget(options);
|
||||
}
|
||||
|
||||
function resolveChunkByteBudget(options?: PipelineOptions): number {
|
||||
const opt = options?.chunkByteBudget;
|
||||
if (typeof opt === 'number' && Number.isFinite(opt) && opt > 0) return opt;
|
||||
|
|
@ -793,25 +815,73 @@ export async function runChunkedParseAndResolve(
|
|||
const verboseThroughputLog = isDev || isVerboseIngestionEnabled();
|
||||
const heapProbeEveryN = isDebugHeapEnabled() ? 25 : 0;
|
||||
|
||||
// ── Merge pipelining (#worker-idle) ──────────────────────────────────────
|
||||
// Merging a chunk's worker results into the graph is the only remaining
|
||||
// serial main-thread step (ParsedFile serialization now runs in workers).
|
||||
// To stop the whole pool idling during that merge, we OVERLAP it with the
|
||||
// NEXT chunk's worker parse: a freshly-dispatched worker chunk is parked in
|
||||
// `pendingWorkerChunk`, and we merge+finalize it only AFTER starting the
|
||||
// following chunk's dispatch — so the workers parse chunk N+1 while the
|
||||
// main thread merges chunk N. Chunk ORDER is preserved (N finalized before
|
||||
// N+1), which keeps the deferred aggregation deterministic. Cache-hit
|
||||
// chunks drain any pending chunk first, then finalize inline (no worker
|
||||
// dispatch to overlap).
|
||||
interface PendingWorkerChunk {
|
||||
readonly rawResults: ParseWorkerResult[];
|
||||
readonly chunkIdx: number;
|
||||
readonly chunkHash: string | null;
|
||||
readonly chunkFiles: Array<{ path: string; content: string }>;
|
||||
readonly chunkStartMs: number | null;
|
||||
}
|
||||
let pendingWorkerChunk: PendingWorkerChunk | null = null;
|
||||
// ── Dispatch rounds + merge pipelining (#worker-idle) ────────────────────
|
||||
// Two separate idle sources, handled together here.
|
||||
//
|
||||
// 1. Barrier per chunk. `dispatch` resolves only when every job it created
|
||||
// has committed, so dispatching one cache pack at a time strands the
|
||||
// pool whenever a pack is smaller than it — which stable packs usually
|
||||
// are. Chunks accumulate into a ROUND (bounded by `roundByteBudget` of
|
||||
// cache-missing source) and go out in one `dispatchGroups` call.
|
||||
// 2. Serial merge. Merging worker results into the graph is the only
|
||||
// remaining serial main-thread step (ParsedFile serialization now runs
|
||||
// in workers). A dispatched round is parked in `pendingRound` and
|
||||
// merged only AFTER the following round's dispatch has started, so the
|
||||
// workers parse round N+1 while the main thread merges round N.
|
||||
//
|
||||
// Chunk ORDER is preserved throughout — rounds drain in order and entries
|
||||
// inside a round finalize by `chunkIdx` — which keeps deferred aggregation
|
||||
// deterministic regardless of how chunks were batched. Cache hits ride
|
||||
// along as round entries so they observe the same ordering without forcing
|
||||
// a dispatch.
|
||||
/**
|
||||
* One chunk queued into the current round. A `hit` already has its worker
|
||||
* output (from the parse cache); a `miss` gets it from the round's single
|
||||
* `dispatchGroups` call. Both are finalized in `chunkIdx` order when the
|
||||
* round drains, which is what keeps deferred aggregation deterministic
|
||||
* regardless of how chunks were batched.
|
||||
*/
|
||||
type RoundEntry =
|
||||
| {
|
||||
readonly kind: 'hit';
|
||||
readonly chunkIdx: number;
|
||||
// A hit never reaches a worker, so it needs the file COUNT (progress,
|
||||
// throughput log) but never the source strings. Holding those would
|
||||
// pin the whole repo's text for a warm run, which is what the
|
||||
// buffered budget below exists to bound.
|
||||
readonly fileCount: number;
|
||||
readonly chunkStartMs: number | null;
|
||||
readonly cachedRaw: ParseWorkerResult[];
|
||||
}
|
||||
| {
|
||||
readonly kind: 'miss';
|
||||
readonly chunkIdx: number;
|
||||
readonly chunkHash: string | null;
|
||||
readonly chunkFiles: Array<{ path: string; content: string }>;
|
||||
readonly chunkStartMs: number | null;
|
||||
};
|
||||
|
||||
const roundByteBudget = resolveParseRoundByteBudget(options);
|
||||
let roundEntries: RoundEntry[] = [];
|
||||
let roundMissBytes = 0;
|
||||
/**
|
||||
* Bytes an open round is HOLDING, counting hits as well as misses.
|
||||
*
|
||||
* `roundMissBytes` alone bounds only what the workers are asked to do, so a
|
||||
* warm run — where nothing misses — would never reach the close condition
|
||||
* and would buffer every chunk's cached output until the tail drain. That
|
||||
* is the #2649 heap failure on a large repo. Closing on either cap keeps a
|
||||
* hits-only run draining at the same cadence as a cold one; `startRound`
|
||||
* already supports a round with no misses.
|
||||
*/
|
||||
let roundBufferedBytes = 0;
|
||||
/**
|
||||
* Files QUEUED into rounds so far. `filesParsedSoFar` only advances when a
|
||||
* round drains, so it is the right number for the throughput log but would
|
||||
* pin a warm run's progress bar at the phase floor for the whole loop.
|
||||
*/
|
||||
let queuedFilesSoFar = 0;
|
||||
let pendingRound: { entries: RoundEntry[]; missResults: ParseWorkerResult[][] } | null = null;
|
||||
|
||||
// Apply one chunk's merged worker data: per-chunk aggregation into the
|
||||
// run-level accumulators + the throughput log. Shared by the cache-hit
|
||||
|
|
@ -821,7 +891,7 @@ export async function runChunkedParseAndResolve(
|
|||
const applyChunkResults = async (
|
||||
chunkWorkerData: WorkerExtractedData | null,
|
||||
chunkIdx: number,
|
||||
chunkFiles: Array<{ path: string; content: string }>,
|
||||
fileCount: number,
|
||||
chunkStartMs: number | null,
|
||||
): Promise<void> => {
|
||||
if (chunkWorkerData) {
|
||||
|
|
@ -898,18 +968,18 @@ export async function runChunkedParseAndResolve(
|
|||
}
|
||||
}
|
||||
|
||||
filesParsedSoFar += chunkFiles.length;
|
||||
filesParsedSoFar += fileCount;
|
||||
|
||||
if (verboseThroughputLog && chunkStartMs !== null) {
|
||||
const elapsedMs = Date.now() - chunkStartMs;
|
||||
const filesPerSec = elapsedMs > 0 ? (chunkFiles.length * 1000) / elapsedMs : 0;
|
||||
const filesPerSec = elapsedMs > 0 ? (fileCount * 1000) / elapsedMs : 0;
|
||||
const stats = workerPool?.getStats?.();
|
||||
const poolFrag = stats
|
||||
? ` pool: ${stats.activeSlots}/${stats.size} active, ` +
|
||||
`${stats.quarantined} quarantined${stats.poolBroken ? ', BROKEN' : ''}`
|
||||
: ' (cache replay)';
|
||||
logger.info(
|
||||
`📊 chunk ${chunkIdx + 1}/${numChunks}: ${chunkFiles.length} files in ${elapsedMs}ms ` +
|
||||
`📊 chunk ${chunkIdx + 1}/${numChunks}: ${fileCount} files in ${elapsedMs}ms ` +
|
||||
`(${filesPerSec.toFixed(1)} files/s)${poolFrag}`,
|
||||
);
|
||||
}
|
||||
|
|
@ -917,12 +987,15 @@ export async function runChunkedParseAndResolve(
|
|||
|
||||
// Merge + finalize a parked worker chunk: graph merge (the overlapped
|
||||
// main-thread step) → parse-cache write-guard → run-level aggregation.
|
||||
const finalizeWorkerChunk = async (p: PendingWorkerChunk): Promise<void> => {
|
||||
const chunkWorkerData = mergeChunkResults(graph, symbolTable, p.rawResults, exportedTypeMap);
|
||||
const finalizeWorkerChunk = async (
|
||||
p: Extract<RoundEntry, { kind: 'miss' }>,
|
||||
rawResults: ParseWorkerResult[],
|
||||
): Promise<void> => {
|
||||
const chunkWorkerData = mergeChunkResults(graph, symbolTable, rawResults, exportedTypeMap);
|
||||
// Persist raw results for this chunk hash (skipping when any chunk file
|
||||
// was worker-quarantined, so the narrower rawResults isn't cached under
|
||||
// the full-chunk key — see the original inline note / U20.U2).
|
||||
if (parseCache && p.chunkHash && p.rawResults.length > 0) {
|
||||
if (parseCache && p.chunkHash && rawResults.length > 0) {
|
||||
const quarantineSet = new Set(workerPool?.getQuarantinedPaths?.() ?? []);
|
||||
const chunkHadQuarantine = p.chunkFiles.some((f) => quarantineSet.has(f.path));
|
||||
if (chunkHadQuarantine) {
|
||||
|
|
@ -935,7 +1008,7 @@ export async function runChunkedParseAndResolve(
|
|||
);
|
||||
}
|
||||
} else {
|
||||
await persistParseCacheChunk(parseCache, p.chunkHash, p.rawResults);
|
||||
await persistParseCacheChunk(parseCache, p.chunkHash, rawResults);
|
||||
if (isDev) {
|
||||
logger.info(
|
||||
`📦 parse-cache MISS+store: chunk ${p.chunkIdx + 1}/${numChunks} (${p.chunkFiles.length} files, ${p.chunkHash.slice(0, 8)})`,
|
||||
|
|
@ -943,7 +1016,171 @@ export async function runChunkedParseAndResolve(
|
|||
}
|
||||
}
|
||||
}
|
||||
await applyChunkResults(chunkWorkerData, p.chunkIdx, p.chunkFiles, p.chunkStartMs);
|
||||
await applyChunkResults(chunkWorkerData, p.chunkIdx, p.chunkFiles.length, p.chunkStartMs);
|
||||
};
|
||||
|
||||
/**
|
||||
* Dispatch a round's cache misses as ONE pool round. Returns the parked
|
||||
* round; the caller drains it after starting the next one so the workers
|
||||
* parse round N+1 while the main thread merges round N (the same overlap
|
||||
* the per-chunk loop had, at round granularity).
|
||||
*/
|
||||
const startRound = async (
|
||||
entries: RoundEntry[],
|
||||
): Promise<{ entries: RoundEntry[]; results: Promise<ParseWorkerResult[][]> } | null> => {
|
||||
if (entries.length === 0) return null;
|
||||
const misses = entries.filter((entry) => entry.kind === 'miss');
|
||||
if (misses.length === 0) {
|
||||
return { entries, results: Promise.resolve([]) };
|
||||
}
|
||||
for (const miss of misses) {
|
||||
if (durableParsedFileDir !== undefined && miss.chunkHash !== null) {
|
||||
try {
|
||||
await prepareDurableParsedFileChunk(durableParsedFileDir, miss.chunkHash);
|
||||
} catch (err) {
|
||||
// The durable store is an optimization — degrade like the restore
|
||||
// path does instead of failing the analyze. Workers recreate the
|
||||
// directory on write, so at worst the old generation lingers.
|
||||
logger.warn(
|
||||
{ err, chunkHash: miss.chunkHash.slice(0, 8) },
|
||||
'parsedfile-cache: could not reset durable chunk generation; continuing',
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
const roundFiles = misses.reduce((sum, miss) => sum + miss.chunkFiles.length, 0);
|
||||
const firstIdx = misses[0].chunkIdx;
|
||||
const lastIdx = misses[misses.length - 1].chunkIdx;
|
||||
const progressForRound = (current: number, _total: number, filePath: string) => {
|
||||
// Rounds queued before this one are already counted in
|
||||
// `queuedFilesSoFar`; `current` is this round's own worker progress.
|
||||
const globalCurrent = queuedFilesSoFar - roundFiles + current;
|
||||
// Parse phase covers 20-70 (M2). Deferred extraction handles 70-95.
|
||||
const parsingProgress = 20 + (globalCurrent / totalParseable) * 50;
|
||||
onProgress({
|
||||
phase: 'parsing',
|
||||
percent: Math.round(parsingProgress),
|
||||
message:
|
||||
firstIdx === lastIdx
|
||||
? `Parsing chunk ${firstIdx + 1}/${numChunks}...`
|
||||
: `Parsing chunks ${firstIdx + 1}-${lastIdx + 1}/${numChunks}...`,
|
||||
detail: filePath,
|
||||
stats: {
|
||||
filesProcessed: globalCurrent,
|
||||
totalFiles: totalParseable,
|
||||
nodesCreated: graph.nodeCount,
|
||||
},
|
||||
});
|
||||
};
|
||||
const activeWorkerPool = getOrCreateWorkerPool();
|
||||
if (verboseThroughputLog) {
|
||||
logger.info(
|
||||
`🚚 round: ${misses.length} chunk(s) ${firstIdx + 1}-${lastIdx + 1}/${numChunks}, ` +
|
||||
`${roundFiles} files in one dispatch`,
|
||||
);
|
||||
}
|
||||
const results = dispatchChunkParseRound(
|
||||
misses.map((miss) => ({
|
||||
items: miss.chunkFiles,
|
||||
chunkHash: miss.chunkHash ?? undefined,
|
||||
})),
|
||||
activeWorkerPool,
|
||||
progressForRound,
|
||||
);
|
||||
// Mark handled so a rejection during the overlap drain below isn't
|
||||
// flagged as unhandled; the `await` in drainRound re-throws it for real
|
||||
// handling.
|
||||
results.catch(() => {});
|
||||
return { entries, results };
|
||||
};
|
||||
|
||||
/**
|
||||
* Merge + finalize every chunk of a parked round, in `chunkIdx` order.
|
||||
* Takes RESOLVED worker output: the round's dispatch must already have
|
||||
* settled before this runs, because the pool allows only one dispatch in
|
||||
* flight at a time (see `closeRound`).
|
||||
*/
|
||||
const drainRound = async (round: {
|
||||
entries: RoundEntry[];
|
||||
missResults: ParseWorkerResult[][];
|
||||
}): Promise<void> => {
|
||||
const missResults = round.missResults;
|
||||
const missCount = round.entries.reduce(
|
||||
(sum, entry) => sum + (entry.kind === 'miss' ? 1 : 0),
|
||||
0,
|
||||
);
|
||||
// `dispatchGroups` returns one array per input group. If that contract
|
||||
// ever breaks, every later entry in this round would silently merge the
|
||||
// wrong chunk's results and skip its cache write, with a clean exit.
|
||||
if (missResults.length !== missCount) {
|
||||
throw new Error(
|
||||
`Parse round result mismatch: ${missResults.length} result group(s) for ${missCount} dispatched chunk(s).`,
|
||||
);
|
||||
}
|
||||
let missIdx = 0;
|
||||
for (const entry of round.entries) {
|
||||
if (entry.kind === 'hit') {
|
||||
const chunkWorkerData = mergeChunkResults(
|
||||
graph,
|
||||
symbolTable,
|
||||
entry.cachedRaw,
|
||||
exportedTypeMap,
|
||||
);
|
||||
await applyChunkResults(
|
||||
chunkWorkerData,
|
||||
entry.chunkIdx,
|
||||
entry.fileCount,
|
||||
entry.chunkStartMs,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
await finalizeWorkerChunk(entry, missResults[missIdx++]);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Close the accumulated round.
|
||||
*
|
||||
* `WorkerPool.dispatch`/`dispatchGroups` is NOT reentrant — concurrent
|
||||
* calls race on the shared per-slot busy/in-flight state and wedge the
|
||||
* pool until every worker idle-times out. So exactly one dispatch is in
|
||||
* flight here: start this round, merge the PREVIOUS round (whose results
|
||||
* are already resolved) while these workers run, then await this round and
|
||||
* park it resolved for the next close to merge.
|
||||
*/
|
||||
const closeRound = async (): Promise<void> => {
|
||||
const started = await startRound(roundEntries);
|
||||
roundEntries = [];
|
||||
roundMissBytes = 0;
|
||||
roundBufferedBytes = 0;
|
||||
const previous = pendingRound;
|
||||
pendingRound = null;
|
||||
if (previous) {
|
||||
try {
|
||||
await drainRound(previous);
|
||||
} catch (err) {
|
||||
// The round started above is still on the workers. Unwinding now
|
||||
// reaches this function's `finally`, which calls `terminate()` — and
|
||||
// terminate kills busy workers outright, which is the #2432
|
||||
// mid-N-API SIGABRT hazard. Let the in-flight round settle first so
|
||||
// the pool is idle, then propagate the original failure.
|
||||
await started?.results.catch(() => undefined);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
if (!started) return;
|
||||
let missResults: ParseWorkerResult[][];
|
||||
try {
|
||||
missResults = await started.results;
|
||||
} catch (err) {
|
||||
if (!(err instanceof WorkerPoolInitializationError)) throw err;
|
||||
// Every worker crashed during startup and the pool's bounded self-heal
|
||||
// was exhausted. Fail fast (#1741) — there is no sequential parser to
|
||||
// degrade to. `handleWorkerStartupFailure` always throws, so
|
||||
// `missResults` stays definitely assigned for the parked round below.
|
||||
handleWorkerStartupFailure(err);
|
||||
}
|
||||
pendingRound = { entries: started.entries, missResults };
|
||||
};
|
||||
|
||||
for (let chunkIdx = 0; chunkIdx < numChunks; chunkIdx++) {
|
||||
|
|
@ -1042,13 +1279,8 @@ export async function runChunkedParseAndResolve(
|
|||
// Cache hit: replay cached worker output. Finalize any parked worker
|
||||
// chunk FIRST so deferred aggregation stays in chunk order, then merge
|
||||
// + apply this hit inline (no worker dispatch to overlap).
|
||||
if (pendingWorkerChunk) {
|
||||
await finalizeWorkerChunk(pendingWorkerChunk);
|
||||
pendingWorkerChunk = null;
|
||||
}
|
||||
chunkCacheHits++;
|
||||
parseCacheHitFileCount += chunkFiles.length;
|
||||
const chunkWorkerData = mergeChunkResults(graph, symbolTable, cachedRaw, exportedTypeMap);
|
||||
if (isDev) {
|
||||
logger.info(
|
||||
`📦 parse-cache HIT: chunk ${chunkIdx + 1}/${numChunks} (${chunkFiles.length} files, ${chunkHash?.slice(0, 8) ?? 'unknown'})`,
|
||||
|
|
@ -1062,90 +1294,44 @@ export async function runChunkedParseAndResolve(
|
|||
// takes 70-95 so the UI advances through the (potentially long)
|
||||
// resolution stages instead of holding at 82 (M2 from PR #1693
|
||||
// review).
|
||||
percent: Math.round(20 + ((filesParsedSoFar + cachedFiles) / totalParseable) * 50),
|
||||
percent: Math.round(20 + ((queuedFilesSoFar + cachedFiles) / totalParseable) * 50),
|
||||
message: `Parsing chunk ${chunkIdx + 1}/${numChunks} (cache)...`,
|
||||
stats: {
|
||||
filesProcessed: filesParsedSoFar + cachedFiles,
|
||||
filesProcessed: queuedFilesSoFar + cachedFiles,
|
||||
totalFiles: totalParseable,
|
||||
nodesCreated: graph.nodeCount,
|
||||
},
|
||||
});
|
||||
// The durable gate already snapshotted warm `.v8` shards into the
|
||||
// run-scoped store for scope resolution.
|
||||
await applyChunkResults(chunkWorkerData, chunkIdx, chunkFiles, chunkStartMs);
|
||||
// run-scoped store for scope resolution. Queue into the round so this
|
||||
// hit still finalizes in `chunkIdx` order relative to its neighbours.
|
||||
roundEntries.push({
|
||||
kind: 'hit',
|
||||
chunkIdx,
|
||||
fileCount: chunkFiles.length,
|
||||
chunkStartMs,
|
||||
cachedRaw,
|
||||
});
|
||||
for (const file of chunkFiles) roundBufferedBytes += file.content.length;
|
||||
queuedFilesSoFar += chunkFiles.length;
|
||||
} else {
|
||||
// Cache miss: dispatch to workers, capture the raw results, store
|
||||
// them under the chunk hash for the next run.
|
||||
// Cache miss: queue for the round's single dispatch; the raw results
|
||||
// are stored under the chunk hash when the round drains.
|
||||
chunkCacheMisses++;
|
||||
reparsedFileCount += chunkFiles.length;
|
||||
if (durableParsedFileDir !== undefined && chunkHash !== null) {
|
||||
try {
|
||||
await prepareDurableParsedFileChunk(durableParsedFileDir, chunkHash);
|
||||
} catch (err) {
|
||||
// The durable store is an optimization — degrade like the restore
|
||||
// path does instead of failing the analyze. Workers recreate the
|
||||
// directory on write, so at worst the old generation lingers.
|
||||
logger.warn(
|
||||
{ err, chunkHash: chunkHash.slice(0, 8) },
|
||||
'parsedfile-cache: could not reset durable chunk generation; continuing',
|
||||
);
|
||||
}
|
||||
roundEntries.push({ kind: 'miss', chunkIdx, chunkHash, chunkFiles, chunkStartMs });
|
||||
for (const file of chunkFiles) {
|
||||
roundMissBytes += file.content.length;
|
||||
roundBufferedBytes += file.content.length;
|
||||
}
|
||||
const progressForChunk = (current: number, _total: number, filePath: string) => {
|
||||
const globalCurrent = filesParsedSoFar + current;
|
||||
// Parse phase covers 20-70 (M2). Deferred extraction handles 70-95.
|
||||
const parsingProgress = 20 + (globalCurrent / totalParseable) * 50;
|
||||
onProgress({
|
||||
phase: 'parsing',
|
||||
percent: Math.round(parsingProgress),
|
||||
message: `Parsing chunk ${chunkIdx + 1}/${numChunks}...`,
|
||||
detail: filePath,
|
||||
stats: {
|
||||
filesProcessed: globalCurrent,
|
||||
totalFiles: totalParseable,
|
||||
nodesCreated: graph.nodeCount,
|
||||
},
|
||||
});
|
||||
};
|
||||
const activeWorkerPool = getOrCreateWorkerPool();
|
||||
// Worker path — PIPELINE: kick off this chunk's dispatch, merge the
|
||||
// PREVIOUS chunk while these workers parse, then park this chunk for
|
||||
// the next iteration to merge (overlapping its parse). The deferred
|
||||
// merge + parse-cache write-guard + aggregation all run in
|
||||
// `finalizeWorkerChunk`, in chunk order. The pool is the sole parse
|
||||
// path — `getOrCreateWorkerPool` returns a pool or throws.
|
||||
const dispatchPromise = dispatchChunkParse(
|
||||
chunkFiles,
|
||||
activeWorkerPool,
|
||||
progressForChunk,
|
||||
undefined,
|
||||
chunkHash ?? undefined,
|
||||
);
|
||||
// Mark handled so a rejection during the overlap drain below isn't
|
||||
// flagged as unhandled; the `await` re-throws it for real handling.
|
||||
dispatchPromise.catch(() => {});
|
||||
if (pendingWorkerChunk) {
|
||||
await finalizeWorkerChunk(pendingWorkerChunk);
|
||||
pendingWorkerChunk = null;
|
||||
}
|
||||
let chunkResults: ParseWorkerResult[];
|
||||
try {
|
||||
chunkResults = await dispatchPromise;
|
||||
} catch (err) {
|
||||
if (!(err instanceof WorkerPoolInitializationError)) throw err;
|
||||
// Every worker crashed during startup and the pool's bounded
|
||||
// self-heal was exhausted. Fail fast (#1741) — there is no sequential
|
||||
// parser to degrade to. `handleWorkerStartupFailure` always throws, so
|
||||
// `chunkResults` stays definitely assigned for the parked chunk below.
|
||||
handleWorkerStartupFailure(err);
|
||||
}
|
||||
pendingWorkerChunk = {
|
||||
rawResults: chunkResults,
|
||||
chunkIdx,
|
||||
chunkHash,
|
||||
chunkFiles,
|
||||
chunkStartMs,
|
||||
};
|
||||
queuedFilesSoFar += chunkFiles.length;
|
||||
}
|
||||
|
||||
// Close on EITHER cap. `roundMissBytes` sizes the worker round;
|
||||
// `roundBufferedBytes` bounds what the main thread is holding, which is
|
||||
// the only cap a warm run can ever reach.
|
||||
if (roundMissBytes >= roundByteBudget || roundBufferedBytes >= roundByteBudget) {
|
||||
await closeRound();
|
||||
}
|
||||
|
||||
// (Per-chunk aggregation + parse-cache write + throughput log now run in
|
||||
|
|
@ -1155,11 +1341,13 @@ export async function runChunkedParseAndResolve(
|
|||
// scope-resolution phase, RING4-2 #943.)
|
||||
}
|
||||
|
||||
// Drain the final parked worker chunk — the last pipelined chunk has no
|
||||
// successor to overlap its merge with, so merge + finalize it here.
|
||||
if (pendingWorkerChunk) {
|
||||
await finalizeWorkerChunk(pendingWorkerChunk);
|
||||
pendingWorkerChunk = null;
|
||||
// Drain the tail: close the partially-filled round, then drain the round
|
||||
// it parked — the last round has no successor to overlap its merge with.
|
||||
if (roundEntries.length > 0) await closeRound();
|
||||
if (pendingRound) {
|
||||
const last = pendingRound;
|
||||
pendingRound = null;
|
||||
await drainRound(last);
|
||||
}
|
||||
|
||||
if (isDev && parseCache && (chunkCacheHits > 0 || chunkCacheMisses > 0)) {
|
||||
|
|
|
|||
|
|
@ -96,10 +96,45 @@ export function buildDispatchMessage<T>(items: readonly T[]): {
|
|||
transferList,
|
||||
};
|
||||
}
|
||||
/**
|
||||
* One content-addressed parse-cache chunk's worth of work inside a pool round.
|
||||
* See {@link WorkerPool.dispatchGroups}.
|
||||
*/
|
||||
export interface DispatchGroup<TInput> {
|
||||
readonly items: readonly TInput[];
|
||||
/**
|
||||
* Chunk hash tagged onto every job derived from `items`, exactly as the
|
||||
* `chunkHash` argument of {@link WorkerPool.dispatch} does for a lone chunk.
|
||||
*/
|
||||
readonly chunkHash?: string;
|
||||
}
|
||||
|
||||
export interface WorkerPool {
|
||||
/**
|
||||
* Dispatch items across workers. Items are split into bounded jobs, each job
|
||||
* is committed independently, and stalled jobs are split/retried locally.
|
||||
* Dispatch several content-addressed chunks in ONE pool round.
|
||||
*
|
||||
* `dispatch` is a barrier: it resolves only once every job it created has
|
||||
* committed, so dispatching one small parse-cache pack at a time leaves most
|
||||
* slots idle for the whole round-trip. Stable packs are keyed by
|
||||
* `(language, hash(path) % 128)`, which routinely yields packs far below the
|
||||
* byte budget — on this repo, 1285 packs where the budget alone needs 16, and
|
||||
* 549 of them hold a single file. Batching packs into one round removes those
|
||||
* barriers without touching pack identity: jobs are still cut at group
|
||||
* boundaries, so each job carries exactly one `chunkHash` and every result
|
||||
* stays attributable to the pack that owns its cache key.
|
||||
*
|
||||
* Returns one result array per input group, in input order. A group whose
|
||||
* items were all quarantined yields an empty array.
|
||||
*/
|
||||
dispatchGroups<TInput, TResult>(
|
||||
groups: readonly DispatchGroup<TInput>[],
|
||||
onProgress?: (filesProcessed: number) => void,
|
||||
): Promise<TResult[][]>;
|
||||
|
||||
/**
|
||||
* Dispatch ONE chunk across workers — {@link WorkerPool.dispatchGroups} with
|
||||
* a single group. Items are split into bounded jobs, each job is committed
|
||||
* independently, and stalled jobs are split/retried locally.
|
||||
*
|
||||
* Files in {@link WorkerPool.getQuarantinedPaths} are filtered out before
|
||||
* dispatch — they have already caused a worker death this pool lifetime and
|
||||
|
|
@ -863,15 +898,21 @@ function inFlightExcludePath<TInput>(job: WorkerJob<TInput>, lastProgress: numbe
|
|||
return path ? [path] : [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Cut `items` into bounded jobs. `startIndexOffset` places those jobs on a
|
||||
* shared index space so several groups can be laid out end to end in one
|
||||
* dispatch round and every result still sorts back into global input order.
|
||||
*/
|
||||
function createJobs<TInput>(
|
||||
items: TInput[],
|
||||
items: readonly TInput[],
|
||||
maxItems: number,
|
||||
maxBytes: number,
|
||||
timeoutMs: number,
|
||||
chunkHash?: string,
|
||||
startIndexOffset = 0,
|
||||
): WorkerJob<TInput>[] {
|
||||
const jobs: WorkerJob<TInput>[] = [];
|
||||
let startIndex = 0;
|
||||
let startIndex = startIndexOffset;
|
||||
let batch: TInput[] = [];
|
||||
let batchBytes = 0;
|
||||
|
||||
|
|
@ -1263,11 +1304,44 @@ export const createWorkerPool = (
|
|||
workers.map((_, i) => bringSlotReady(i)),
|
||||
).then(() => undefined);
|
||||
|
||||
const dispatch = async <TInput, TResult>(
|
||||
items: TInput[],
|
||||
/**
|
||||
* Guards the one-dispatch-at-a-time contract. The dispatch machinery keeps
|
||||
* its jobs/busy-slot/in-flight state per call, so two concurrent dispatches
|
||||
* hand the same slots out twice: both stall, and the failure surfaces only
|
||||
* when every worker hits its idle timeout (10s+ of a wedged pool with no
|
||||
* indication of the cause). Fail loudly at the call instead.
|
||||
*/
|
||||
let dispatchInFlight = false;
|
||||
|
||||
/**
|
||||
* Claim the pool synchronously, then run the dispatch. The claim CANNOT be
|
||||
* taken inside `dispatchGroupsInner`: its first statement awaits the
|
||||
* readiness gate, so two calls made in the same tick would both get past the
|
||||
* check before either set the flag.
|
||||
*/
|
||||
const dispatchGroups = <TInput, TResult>(
|
||||
groups: readonly DispatchGroup<TInput>[],
|
||||
onProgress?: (filesProcessed: number) => void,
|
||||
chunkHash?: string,
|
||||
): Promise<TResult[]> => {
|
||||
): Promise<TResult[][]> => {
|
||||
if (dispatchInFlight) {
|
||||
return Promise.reject(
|
||||
new WorkerPoolDispatchError(
|
||||
'Worker pool dispatch is already in flight. `dispatch`/`dispatchGroups` is not ' +
|
||||
'reentrant — await the previous call before starting another on the same pool.',
|
||||
[],
|
||||
),
|
||||
);
|
||||
}
|
||||
dispatchInFlight = true;
|
||||
return dispatchGroupsInner<TInput, TResult>(groups, onProgress).finally(() => {
|
||||
dispatchInFlight = false;
|
||||
});
|
||||
};
|
||||
|
||||
const dispatchGroupsInner = async <TInput, TResult>(
|
||||
groups: readonly DispatchGroup<TInput>[],
|
||||
onProgress?: (filesProcessed: number) => void,
|
||||
): Promise<TResult[][]> => {
|
||||
// Await the initial-spawn readiness gate (F13). On first dispatch
|
||||
// this blocks for up to poolOptions.workerReadyTimeoutMs while every initial
|
||||
// worker's `{type:'ready'}` handshake is checked; on subsequent
|
||||
|
|
@ -1285,7 +1359,8 @@ export const createWorkerPool = (
|
|||
[],
|
||||
);
|
||||
}
|
||||
if (items.length === 0) return [];
|
||||
const emptyPerGroup = (): TResult[][] => groups.map(() => []);
|
||||
if (groups.every((group) => group.items.length === 0)) return emptyPerGroup();
|
||||
if (activeSlots.size === 0) {
|
||||
const detail =
|
||||
initialReadinessFailures.length > 0
|
||||
|
|
@ -1308,30 +1383,51 @@ export const createWorkerPool = (
|
|||
// Layer 3: filter out quarantined paths so a known-bad file never reaches
|
||||
// a worker again this pool lifetime. The caller queries
|
||||
// `getQuarantinedPaths` after dispatch to route filtered items.
|
||||
const dispatchableItems: TInput[] = [];
|
||||
for (const item of items) {
|
||||
const path = itemPath(item);
|
||||
if (path !== undefined && quarantine.has(path)) continue;
|
||||
dispatchableItems.push(item);
|
||||
}
|
||||
if (dispatchableItems.length === 0) return [];
|
||||
const dispatchableGroups = groups.map((group) => {
|
||||
const items: TInput[] = [];
|
||||
for (const item of group.items) {
|
||||
const path = itemPath(item);
|
||||
if (path !== undefined && quarantine.has(path)) continue;
|
||||
items.push(item);
|
||||
}
|
||||
return { items, chunkHash: group.chunkHash };
|
||||
});
|
||||
const dispatchableCount = dispatchableGroups.reduce(
|
||||
(sum, group) => sum + group.items.length,
|
||||
0,
|
||||
);
|
||||
if (dispatchableCount === 0) return emptyPerGroup();
|
||||
|
||||
// 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,
|
||||
maxItemsPerJob,
|
||||
poolOptions.subBatchMaxBytes,
|
||||
poolOptions.subBatchIdleTimeoutMs,
|
||||
chunkHash,
|
||||
Math.max(1, Math.floor(dispatchableCount / activeSlots.size)),
|
||||
);
|
||||
// Lay the groups end to end on one index space and cut jobs at every group
|
||||
// boundary. A job therefore belongs to exactly one group, which is what
|
||||
// lets a result be attributed back to the parse-cache chunk that owns it
|
||||
// (and what keeps `chunkHash` a per-job constant through splits/requeues).
|
||||
const jobs: WorkerJob<TInput>[] = [];
|
||||
const groupEnds: number[] = [];
|
||||
let groupStart = 0;
|
||||
for (const group of dispatchableGroups) {
|
||||
for (const job of createJobs(
|
||||
group.items,
|
||||
maxItemsPerJob,
|
||||
poolOptions.subBatchMaxBytes,
|
||||
poolOptions.subBatchIdleTimeoutMs,
|
||||
group.chunkHash,
|
||||
groupStart,
|
||||
)) {
|
||||
jobs.push(job);
|
||||
}
|
||||
groupStart += group.items.length;
|
||||
groupEnds.push(groupStart);
|
||||
}
|
||||
|
||||
return new Promise<TResult[]>((resolve, reject) => {
|
||||
return await new Promise<TResult[][]>((resolve, reject) => {
|
||||
const results: WorkerJobResult<TResult>[] = [];
|
||||
const inFlightProgress = new Array(size).fill(0);
|
||||
// Tracks which slots are currently mid-job so the "wake idle slots"
|
||||
|
|
@ -1359,10 +1455,7 @@ export const createWorkerPool = (
|
|||
const reportProgress = () => {
|
||||
if (!onProgress) return;
|
||||
const inFlight = inFlightProgress.reduce((sum, value) => sum + value, 0);
|
||||
const next = Math.min(
|
||||
dispatchableItems.length,
|
||||
Math.max(maxReported, completedFiles + inFlight),
|
||||
);
|
||||
const next = Math.min(dispatchableCount, Math.max(maxReported, completedFiles + inFlight));
|
||||
if (next === maxReported) return;
|
||||
maxReported = next;
|
||||
onProgress(next);
|
||||
|
|
@ -1551,9 +1644,19 @@ export const createWorkerPool = (
|
|||
if (jobs.length === 0 && activeWorkers === 0) {
|
||||
stopped = true;
|
||||
results.sort((a, b) => a.startIndex - b.startIndex);
|
||||
if (onProgress && maxReported < dispatchableItems.length)
|
||||
onProgress(dispatchableItems.length);
|
||||
resolve(results.map((result) => result.data));
|
||||
if (onProgress && maxReported < dispatchableCount) onProgress(dispatchableCount);
|
||||
// Partition back per group. Job (and split sub-job) start indices
|
||||
// stay inside their group's span, so a single forward walk over the
|
||||
// sorted results assigns every result to exactly one group.
|
||||
const perGroup: TResult[][] = groupEnds.map(() => []);
|
||||
let groupIdx = 0;
|
||||
for (const result of results) {
|
||||
while (groupIdx < groupEnds.length - 1 && result.startIndex >= groupEnds[groupIdx]) {
|
||||
groupIdx++;
|
||||
}
|
||||
perGroup[groupIdx].push(result.data);
|
||||
}
|
||||
resolve(perGroup);
|
||||
}
|
||||
};
|
||||
|
||||
|
|
@ -2294,8 +2397,18 @@ export const createWorkerPool = (
|
|||
activeSlots.clear();
|
||||
};
|
||||
|
||||
const dispatch = async <TInput, TResult>(
|
||||
items: TInput[],
|
||||
onProgress?: (filesProcessed: number) => void,
|
||||
chunkHash?: string,
|
||||
): Promise<TResult[]> => {
|
||||
const [result] = await dispatchGroups<TInput, TResult>([{ items, chunkHash }], onProgress);
|
||||
return result ?? [];
|
||||
};
|
||||
|
||||
return {
|
||||
dispatch,
|
||||
dispatchGroups,
|
||||
terminate,
|
||||
size,
|
||||
getQuarantinedPaths: () => quarantine.snapshot(),
|
||||
|
|
|
|||
225
gitnexus/test/integration/parse-impl-dispatch-rounds.test.ts
Normal file
225
gitnexus/test/integration/parse-impl-dispatch-rounds.test.ts
Normal file
|
|
@ -0,0 +1,225 @@
|
|||
/**
|
||||
* Dispatch rounds — batching cache packs into one pool round.
|
||||
*
|
||||
* `WorkerPool.dispatch` is a barrier, so one round-trip per parse-cache pack
|
||||
* strands the pool whenever a pack is smaller than it. Packs are keyed by
|
||||
* `(language, hash(path) % 128)`, so on a real repo most of them are: this
|
||||
* repository produces 1285 packs where the byte budget alone needs 16, and 549
|
||||
* of those hold a single file. Chunks now accumulate into a round bounded by
|
||||
* `GITNEXUS_PARSE_ROUND_BYTES` and go out in one `dispatchGroups` call.
|
||||
*
|
||||
* Batching must be invisible to the graph. These tests pin the two ways it
|
||||
* could stop being invisible:
|
||||
* 1. Ordering — deferred aggregation runs in `chunkIdx` order, so the graph
|
||||
* must not depend on how chunks were grouped into rounds.
|
||||
* 2. Attribution — a round returns one result array per pack, so a pack's
|
||||
* parse-cache entry must hold ITS OWN worker output. Mis-attribution would
|
||||
* survive a cold run and only surface as a corrupted warm replay, which is
|
||||
* what the second test exercises.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
|
||||
import fs from 'node:fs';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
|
||||
import { runChunkedParseAndResolve } from '../../src/core/ingestion/pipeline-phases/parse-impl.js';
|
||||
import { createKnowledgeGraph } from '../../src/core/graph/graph.js';
|
||||
import { PARSE_CACHE_VERSION, packParseCacheChunks } from '../../src/storage/parse-cache.js';
|
||||
import type { ParseWorkerResult } from '../../src/core/ingestion/workers/parse-worker.js';
|
||||
|
||||
const ORIGINAL_ROUND_BYTES = process.env.GITNEXUS_PARSE_ROUND_BYTES;
|
||||
|
||||
/**
|
||||
* Enough files, across enough languages, that `(language, bucket)` packing
|
||||
* yields many more packs than the byte budget would — the shape that makes
|
||||
* per-pack dispatch a barrier problem in the first place.
|
||||
*/
|
||||
const FIXTURE: ReadonlyArray<[string, string]> = [
|
||||
...Array.from({ length: 12 }, (_, i): [string, string] => [
|
||||
`src/mod${i}.ts`,
|
||||
`export function ts${i}() { return ${i}; }\n`,
|
||||
]),
|
||||
...Array.from({ length: 8 }, (_, i): [string, string] => [
|
||||
`src/mod${i}.py`,
|
||||
`def py${i}():\n return ${i}\n`,
|
||||
]),
|
||||
...Array.from({ length: 6 }, (_, i): [string, string] => [
|
||||
`src/Mod${i}.java`,
|
||||
`public class Mod${i} { public int go() { return ${i}; } }\n`,
|
||||
]),
|
||||
...Array.from({ length: 6 }, (_, i): [string, string] => [
|
||||
`src/mod${i}.go`,
|
||||
`package main\n\nfunc Go${i}() int { return ${i} }\n`,
|
||||
]),
|
||||
];
|
||||
|
||||
describe('parse-impl dispatch rounds', () => {
|
||||
let repoPath = '';
|
||||
let storageDir = '';
|
||||
|
||||
beforeEach(() => {
|
||||
repoPath = fs.mkdtempSync(path.join(os.tmpdir(), 'parse-impl-dispatch-rounds-'));
|
||||
storageDir = fs.mkdtempSync(path.join(os.tmpdir(), 'parse-impl-rounds-storage-'));
|
||||
for (const [rel, content] of FIXTURE) {
|
||||
const full = path.join(repoPath, rel);
|
||||
fs.mkdirSync(path.dirname(full), { recursive: true });
|
||||
fs.writeFileSync(full, content);
|
||||
}
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
for (const dir of [repoPath, storageDir]) {
|
||||
if (dir && fs.existsSync(dir)) fs.rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
if (ORIGINAL_ROUND_BYTES === undefined) delete process.env.GITNEXUS_PARSE_ROUND_BYTES;
|
||||
else process.env.GITNEXUS_PARSE_ROUND_BYTES = ORIGINAL_ROUND_BYTES;
|
||||
});
|
||||
|
||||
const files = () =>
|
||||
FIXTURE.map(([rel]) => ({ path: rel, size: fs.statSync(path.join(repoPath, rel)).size }));
|
||||
|
||||
/**
|
||||
* Order-independent fingerprint of the graph. Counts alone would let a
|
||||
* mis-attributed chunk (right totals, wrong contents) pass.
|
||||
*/
|
||||
const fingerprint = (graph: ReturnType<typeof createKnowledgeGraph>): string =>
|
||||
Array.from(graph.nodes.values())
|
||||
.map((node) => {
|
||||
const props = node.properties as { name?: string; filePath?: string } | undefined;
|
||||
return `${node.label}|${props?.name ?? ''}|${props?.filePath ?? ''}`;
|
||||
})
|
||||
.sort()
|
||||
.join('\n');
|
||||
|
||||
const run = async (parseCache?: {
|
||||
version: string;
|
||||
entries: Map<string, ParseWorkerResult[]>;
|
||||
usedKeys: Set<string>;
|
||||
storagePath: string;
|
||||
onDiskKeys: Set<string>;
|
||||
}) => {
|
||||
const scan = files();
|
||||
const rels = scan.map((f) => f.path);
|
||||
const graph = createKnowledgeGraph();
|
||||
await runChunkedParseAndResolve(
|
||||
graph,
|
||||
scan,
|
||||
rels,
|
||||
scan.length,
|
||||
repoPath,
|
||||
Date.now(),
|
||||
() => {},
|
||||
parseCache ? { parseCache } : {},
|
||||
);
|
||||
return graph;
|
||||
};
|
||||
|
||||
it('the fixture really does split into more packs than the byte budget needs', () => {
|
||||
// Guards the premise: if packing ever stopped over-splitting, the tests
|
||||
// below would still pass while measuring nothing.
|
||||
const packs = packParseCacheChunks(
|
||||
files().map((f) => ({
|
||||
path: f.path,
|
||||
size: f.size,
|
||||
language: f.path.slice(f.path.lastIndexOf('.') + 1),
|
||||
})),
|
||||
2 * 1024 * 1024,
|
||||
);
|
||||
const totalBytes = files().reduce((sum, f) => sum + f.size, 0);
|
||||
expect(totalBytes).toBeLessThan(2 * 1024 * 1024);
|
||||
expect(packs.length).toBeGreaterThan(1);
|
||||
});
|
||||
|
||||
it('produces the same graph whether chunks are batched into rounds or dispatched one by one', async () => {
|
||||
// 1 byte closes a round after every cache-missing chunk — the pre-round
|
||||
// behaviour, and the control arm for the batched default.
|
||||
process.env.GITNEXUS_PARSE_ROUND_BYTES = '1';
|
||||
const perChunk = await run();
|
||||
|
||||
delete process.env.GITNEXUS_PARSE_ROUND_BYTES;
|
||||
const batched = await run();
|
||||
|
||||
expect(batched.nodeCount).toBe(perChunk.nodeCount);
|
||||
expect(batched.relationshipCount).toBe(perChunk.relationshipCount);
|
||||
expect(fingerprint(batched)).toBe(fingerprint(perChunk));
|
||||
// Pin real symbols so an empty-graph regression cannot satisfy the above.
|
||||
const names = fingerprint(batched);
|
||||
expect(names).toContain('ts0');
|
||||
expect(names).toContain('py0');
|
||||
expect(names).toContain('Mod0');
|
||||
expect(names).toContain('Go0');
|
||||
});
|
||||
|
||||
it('keeps hit and miss chunks attributed to their own files inside one round', async () => {
|
||||
// The realistic incremental shape: some packs warm, some cold, batched into
|
||||
// the SAME round. `drainRound` walks the round's entries in `chunkIdx`
|
||||
// order but pulls worker output with a separate `missIdx` cursor, so a
|
||||
// hit sitting between two misses is exactly where that cursor can slip.
|
||||
// Cold-then-warm alone never exercises it -- every entry is the same kind.
|
||||
const cache = {
|
||||
version: PARSE_CACHE_VERSION,
|
||||
entries: new Map<string, ParseWorkerResult[]>(),
|
||||
usedKeys: new Set<string>(),
|
||||
storagePath: storageDir,
|
||||
onDiskKeys: new Set<string>(),
|
||||
};
|
||||
|
||||
const cold = await run(cache);
|
||||
const cachedPacks = cache.onDiskKeys.size + cache.entries.size;
|
||||
expect(cachedPacks).toBeGreaterThan(1);
|
||||
|
||||
// Edit ONE file. Its pack now misses; every other pack still hits, so the
|
||||
// next run's rounds carry both kinds together.
|
||||
fs.writeFileSync(
|
||||
path.join(repoPath, 'src/mod0.ts'),
|
||||
'export function ts0() { return 999; }\nexport function ts0Extra() { return 1; }\n',
|
||||
);
|
||||
|
||||
const mixed = await run(cache);
|
||||
|
||||
// The edited file's NEW symbol must be present, proving the miss chunk's
|
||||
// fresh worker output landed under its own file...
|
||||
const mixedPrint = fingerprint(mixed);
|
||||
expect(mixedPrint).toContain('ts0Extra');
|
||||
// ...and every untouched file's symbols must still be present and attached
|
||||
// to their own paths, proving no hit chunk was overwritten by, or swapped
|
||||
// with, a neighbouring miss chunk's results.
|
||||
const coldPrint = fingerprint(cold);
|
||||
const untouched = coldPrint
|
||||
.split('\n')
|
||||
.filter((entry) => !entry.endsWith('|src/mod0.ts'))
|
||||
.sort();
|
||||
const mixedUntouched = mixedPrint
|
||||
.split('\n')
|
||||
.filter((entry) => !entry.endsWith('|src/mod0.ts'))
|
||||
.sort();
|
||||
expect(mixedUntouched).toEqual(untouched);
|
||||
});
|
||||
|
||||
it('stores each pack’s own worker output, so a warm replay reproduces the cold graph', async () => {
|
||||
const cache = {
|
||||
version: PARSE_CACHE_VERSION,
|
||||
entries: new Map<string, ParseWorkerResult[]>(),
|
||||
usedKeys: new Set<string>(),
|
||||
storagePath: storageDir,
|
||||
onDiskKeys: new Set<string>(),
|
||||
};
|
||||
|
||||
// Cold: every pack misses, and the round writes each pack's results under
|
||||
// that pack's own hash.
|
||||
const cold = await run(cache);
|
||||
// With a storagePath the chunk bodies land on disk and the hash is tracked
|
||||
// in `onDiskKeys`; without one they stay in `entries`. Count both so the
|
||||
// assertion pins "more than one pack was cached", not the storage route.
|
||||
expect(cache.onDiskKeys.size + cache.entries.size).toBeGreaterThan(1);
|
||||
expect(cache.usedKeys.size).toBe(cache.onDiskKeys.size + cache.entries.size);
|
||||
|
||||
// Warm: every pack replays from its cache entry with no worker dispatch.
|
||||
// If a round had attributed pack A's results to pack B's key, the replayed
|
||||
// graph would differ here even though the cold run looked correct.
|
||||
const warm = await run(cache);
|
||||
expect(fingerprint(warm)).toBe(fingerprint(cold));
|
||||
expect(warm.nodeCount).toBe(cold.nodeCount);
|
||||
expect(warm.relationshipCount).toBe(cold.relationshipCount);
|
||||
});
|
||||
});
|
||||
|
|
@ -50,11 +50,13 @@ function scanned(repo: string, files: string[]) {
|
|||
}
|
||||
|
||||
/**
|
||||
* Capture every per-chunk progress message emitted during a run.
|
||||
* parse-impl emits one per chunk in the "Parsing chunk X/Y" form, so
|
||||
* counting unique chunk indices in the captured stream is a stable
|
||||
* proxy for the number of chunks the loop actually produced. Avoids
|
||||
* exposing internal counter state from parse-impl.
|
||||
* Read the chunk count out of the progress stream. parse-impl reports progress
|
||||
* as "Parsing chunk X/Y" for a single chunk and "Parsing chunks X-Z/Y" when a
|
||||
* dispatch round batches several — so the DENOMINATOR, not the number of
|
||||
* distinct messages, is the count of packs the loop produced. Reading `Y`
|
||||
* keeps this independent of how chunks are grouped into rounds while still
|
||||
* exercising the real budget-resolution path inside
|
||||
* `runChunkedParseAndResolve`, rather than re-deriving packs in the test.
|
||||
*/
|
||||
async function countChunksFromProgress(
|
||||
repoPath: string,
|
||||
|
|
@ -63,7 +65,7 @@ async function countChunksFromProgress(
|
|||
): Promise<number> {
|
||||
const scan = scanned(repoPath, files);
|
||||
const graph = createKnowledgeGraph();
|
||||
const chunkIndices = new Set<string>();
|
||||
const totals = new Set<number>();
|
||||
await runChunkedParseAndResolve(
|
||||
graph,
|
||||
scan,
|
||||
|
|
@ -73,8 +75,8 @@ async function countChunksFromProgress(
|
|||
Date.now(),
|
||||
(p) => {
|
||||
if (typeof p.message !== 'string') return;
|
||||
const m = /Parsing chunk (\d+)\/(\d+)/.exec(p.message);
|
||||
if (m !== null) chunkIndices.add(`${m[1]}/${m[2]}`);
|
||||
const m = /Parsing chunks? \d+(?:-\d+)?\/(\d+)/.exec(p.message);
|
||||
if (m !== null) totals.add(Number(m[1]));
|
||||
},
|
||||
// Chunk count is byte-budget-driven and emitted before the pool runs, so it
|
||||
// is independent of worker vs sequential. Sequential parsing was removed, so
|
||||
|
|
@ -82,7 +84,10 @@ async function countChunksFromProgress(
|
|||
// integration tier.
|
||||
{ ...options },
|
||||
);
|
||||
return chunkIndices.size;
|
||||
// Every message in a run carries the same denominator; more than one value
|
||||
// would mean the loop changed its chunk count mid-run.
|
||||
expect(totals.size).toBeLessThanOrEqual(1);
|
||||
return totals.values().next().value ?? 0;
|
||||
}
|
||||
|
||||
describe('parse-impl chunkByteBudget resolution (U14 / F7)', () => {
|
||||
|
|
|
|||
|
|
@ -840,6 +840,156 @@ describe('worker pool integration', () => {
|
|||
},
|
||||
);
|
||||
|
||||
it('splits packs into one round, keeping results and chunk hashes per group', async () => {
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-worker-dispatch-groups-',
|
||||
`
|
||||
const { parentPort, threadId } = require('node:worker_threads');
|
||||
let paths = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
for (const file of msg.files) paths.push(file.path);
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: msg.files.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
} else if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({
|
||||
type: 'result',
|
||||
data: { paths, threadId, chunkHash: msg.chunkHash },
|
||||
});
|
||||
paths = [];
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
pool = createWorkerPool(pathToFileURL(workerPath), 4);
|
||||
|
||||
try {
|
||||
// Shaped like real cache packs: mostly tiny, one larger. A single round
|
||||
// must still keep every result attributable to the pack that owns it.
|
||||
const groups = [
|
||||
{ chunkHash: 'hash-a', items: [{ path: 'a0.ts', content: 'export const a0 = 1;' }] },
|
||||
{ chunkHash: 'hash-b', items: [{ path: 'b0.ts', content: 'export const b0 = 1;' }] },
|
||||
{
|
||||
chunkHash: 'hash-c',
|
||||
items: Array.from({ length: 12 }, (_, i) => ({
|
||||
path: `c${i}.ts`,
|
||||
content: 'export const c = 1;',
|
||||
})),
|
||||
},
|
||||
{ chunkHash: 'hash-d', items: [{ path: 'd0.ts', content: 'export const d0 = 1;' }] },
|
||||
];
|
||||
const progress: number[] = [];
|
||||
const perGroup = await pool.dispatchGroups<
|
||||
(typeof groups)[number]['items'][number],
|
||||
{ paths: string[]; threadId: number; chunkHash?: string }
|
||||
>(groups, (completed) => progress.push(completed));
|
||||
|
||||
expect(perGroup).toHaveLength(groups.length);
|
||||
// No job straddles a pack: every path comes back under its own group,
|
||||
// and every result carries that group's chunk hash (the cache key).
|
||||
for (const [index, group] of groups.entries()) {
|
||||
expect(perGroup[index].flatMap((result) => result.paths).sort()).toEqual(
|
||||
group.items.map((file) => file.path).sort(),
|
||||
);
|
||||
for (const result of perGroup[index]) expect(result.chunkHash).toBe(group.chunkHash);
|
||||
}
|
||||
// The whole round shares the pool rather than one pack per barrier.
|
||||
const threads = new Set(perGroup.flat().map((result) => result.threadId));
|
||||
expect(threads.size).toBeGreaterThan(1);
|
||||
expect(progress).toEqual([...progress].sort((a, b) => a - b));
|
||||
expect(progress.at(-1)).toBe(15);
|
||||
} finally {
|
||||
await pool.terminate();
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('returns an empty result array for a group whose items were all quarantined', async () => {
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-worker-groups-quarantine-',
|
||||
`
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
let paths = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
for (const file of msg.files) {
|
||||
if (file.path === 'poison.ts') process.exit(134);
|
||||
paths.push(file.path);
|
||||
}
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: msg.files.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
} else if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({ type: 'result', data: { paths } });
|
||||
paths = [];
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
pool = createWorkerPool(pathToFileURL(workerPath), 2);
|
||||
|
||||
try {
|
||||
await pool.dispatch([{ path: 'poison.ts', content: '' }]);
|
||||
expect(pool.getQuarantinedPaths?.()).toContain('poison.ts');
|
||||
|
||||
// Group alignment must survive quarantine filtering — an all-quarantined
|
||||
// group still occupies its slot so results line up with the input packs.
|
||||
const perGroup = await pool.dispatchGroups<
|
||||
{ path: string; content: string },
|
||||
{ paths: string[] }
|
||||
>([
|
||||
{ chunkHash: 'poisoned', items: [{ path: 'poison.ts', content: '' }] },
|
||||
{ chunkHash: 'healthy', items: [{ path: 'ok.ts', content: 'export const ok = 1;' }] },
|
||||
]);
|
||||
|
||||
expect(perGroup).toHaveLength(2);
|
||||
expect(perGroup[0]).toEqual([]);
|
||||
expect(perGroup[1].flatMap((result) => result.paths)).toEqual(['ok.ts']);
|
||||
} finally {
|
||||
await pool.terminate();
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('rejects a second dispatch while one is still in flight', async () => {
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-worker-reentrant-dispatch-',
|
||||
`
|
||||
const { parentPort } = 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') {
|
||||
setTimeout(() => parentPort.postMessage({ type: 'result', data: { paths } }), 150);
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
pool = createWorkerPool(pathToFileURL(workerPath), 2);
|
||||
|
||||
try {
|
||||
// Concurrent dispatches hand the same slots out twice; both then stall
|
||||
// until every worker idle-times out. Fail at the call, not 10s later.
|
||||
const first = pool.dispatch<{ path: string; content: string }, { paths: string[] }>([
|
||||
{ path: 'first.ts', content: 'export const first = 1;' },
|
||||
]);
|
||||
await expect(
|
||||
pool.dispatch([{ path: 'second.ts', content: 'export const second = 1;' }]),
|
||||
).rejects.toThrow(/not reentrant/);
|
||||
await expect(first).resolves.toHaveLength(1);
|
||||
|
||||
// The guard clears once the in-flight dispatch settles.
|
||||
await expect(
|
||||
pool.dispatch([{ path: 'third.ts', content: 'export const third = 1;' }]),
|
||||
).resolves.toHaveLength(1);
|
||||
} 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-',
|
||||
|
|
|
|||
|
|
@ -37,7 +37,8 @@ describe('processParsing — worker-pool error propagation (U20)', () => {
|
|||
const graph = createKnowledgeGraph();
|
||||
const workerPool: WorkerPool = {
|
||||
size: 1,
|
||||
dispatch: vi.fn(async () => {
|
||||
dispatch: vi.fn(async () => []),
|
||||
dispatchGroups: vi.fn(async () => {
|
||||
throw new Error('replacement worker failed');
|
||||
}),
|
||||
terminate: vi.fn(async () => undefined),
|
||||
|
|
@ -63,7 +64,8 @@ describe('processParsing — worker-pool error propagation (U20)', () => {
|
|||
const graph = createKnowledgeGraph();
|
||||
const workerPool: WorkerPool = {
|
||||
size: 1,
|
||||
dispatch: vi.fn(async () => {
|
||||
dispatch: vi.fn(async () => []),
|
||||
dispatchGroups: vi.fn(async () => {
|
||||
throw new WorkerPoolDispatchError(
|
||||
'Worker pool circuit breaker tripped: 2 consecutive failures on slot 0',
|
||||
['src/poison.ts'],
|
||||
|
|
@ -114,6 +116,7 @@ describe('processParsing — worker-pool error propagation (U20)', () => {
|
|||
const workerPool: WorkerPool = {
|
||||
size: 1,
|
||||
dispatch: vi.fn(async () => []),
|
||||
dispatchGroups: vi.fn(async () => []),
|
||||
terminate: vi.fn(async () => undefined),
|
||||
getQuarantinedPaths: () => ['src/poison.ts'],
|
||||
};
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue