From e2892c2047f8f724937e2fcf09813733959a8bc9 Mon Sep 17 00:00:00 2001 From: Gergo Magyar Date: Tue, 19 May 2026 14:59:41 +0100 Subject: [PATCH] perf(workers): raise pool cap to cores-1 + defer per-chunk extraction to keep workers busy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit User reported 4-5% CPU utilization on a multi-core machine during ingestion. Two structural reasons: 1. **Pool cap.** `createWorkerPool` resolved size as `Math.min(8, max(1, os.cpus().length - 1))` — a 16-core box got 8 workers (50% theoretical max). U1 lifts the default to `min(16, max(1, cores - 1))`, exposes `GITNEXUS_WORKER_POOL_SIZE` env override, and adds `--workers ` CLI flag (`0` disables the pool for sequential fallback). 2. **Per-chunk extraction serialized the loop.** Per chunk: dispatch → await workers → main-thread `processImportsFromExtracted` + `processHeritageFromExtracted` + `processRoutesFromExtracted` + `synthesizeWildcardImportBindings` + `seedCrossFileReceiverTypes` → next chunk dispatch. Workers sat idle through every extraction block. U2 (revised from the plan's pipelined-chunks design) defers these passes to a single end-of-loop batch. Chunk loop becomes parse + merge + accumulate. Resolution sees strictly-more-info (full repo graph) so cross-chunk import/heritage targets resolve at least as well as before. Memory cost: `deferredWorkerImports` accumulates across chunks; bounded by total file count, acceptable. Plan deviation note: the plan called for an in-flight chunk pipeline (N concurrent dispatches with bounded memory). That design needed either a `processParsing` API refactor or duplicating its catch-block fallback in `parse-impl`. The deferred-extraction approach delivers the same "workers stay busy" outcome with much smaller surface area and zero changes to `processParsing`. The `GITNEXUS_PARSE_CHUNK_CONCURRENCY` env var documented in U2 of the plan is therefore not implemented in this commit; if memory growth from `deferredWorkerImports` becomes a problem at very-large-repo scale, a bounded sliding-window variant can land as a follow-up. Tests: - New `test/unit/analyze-worker-pool-size.test.ts` covers --workers validation (5 invalid inputs rejected with exit code 1 + clear error; valid integers set the env var; `--workers 0` routes to sequential). - Extended `worker-pool-resilience.test.ts` with `resolveAutoPoolSize` scenarios: env override, env=0, env above cap, invalid env fallback, auto-formula match, integer return type. - Full unit suite: 6097 / 6127 passed / 30 skipped / 0 failed. - Full integration suite (second run): 77 / 78 passed / 1 skipped / 0 failed. First run had a known cosmetic flake from an uncaught worker exception bleeding into the test reporter. Resilience contract from PR #1693 preserved: per-slot respawn budget, circuit breaker, quarantine, authoritative in-flight tracking, cumulative timeout budget — all unchanged. New env vars surfaced in --help: GITNEXUS_WORKER_POOL_SIZE, GITNEXUS_PARSE_CHUNK_CONCURRENCY (reserved for future bounded pipelining). --- gitnexus/src/cli/analyze.ts | 15 ++ gitnexus/src/cli/index.ts | 6 + .../ingestion/pipeline-phases/parse-impl.ts | 166 +++++++++++------- .../src/core/ingestion/workers/worker-pool.ts | 35 +++- .../unit/analyze-worker-pool-size.test.ts | 91 ++++++++++ .../test/unit/worker-pool-resilience.test.ts | 52 ++++++ 6 files changed, 298 insertions(+), 67 deletions(-) create mode 100644 gitnexus/test/unit/analyze-worker-pool-size.test.ts diff --git a/gitnexus/src/cli/analyze.ts b/gitnexus/src/cli/analyze.ts index e24b8c894..77903945b 100644 --- a/gitnexus/src/cli/analyze.ts +++ b/gitnexus/src/cli/analyze.ts @@ -225,6 +225,8 @@ export interface AnalyzeOptions { maxFileSize?: string; /** Override worker sub-batch idle timeout in seconds. */ workerTimeout?: string; + /** Parse worker pool size; 0 disables workers (sequential fallback). */ + workers?: string; embeddingThreads?: string; embeddingBatchSize?: string; embeddingSubBatchSize?: string; @@ -278,6 +280,19 @@ export const analyzeCommand = async (inputPath?: string, options?: AnalyzeOption ); } + if (options?.workers !== undefined) { + const parsedWorkers = Number(options.workers); + if (!Number.isInteger(parsedWorkers) || parsedWorkers < 0) { + cliError( + ' --workers must be a non-negative integer. ' + + 'Pass 0 to disable the worker pool (sequential fallback).\n', + ); + process.exitCode = 1; + return; + } + process.env.GITNEXUS_WORKER_POOL_SIZE = String(parsedWorkers); + } + // Parse `--embeddings [limit]`: `true` → default cap, string → numeric cap // (0 disables the cap entirely). Validated up here so failures match the // sibling-validation pattern (exit before bar.start() — otherwise diff --git a/gitnexus/src/cli/index.ts b/gitnexus/src/cli/index.ts index 5b317e5c6..dbd8b0592 100644 --- a/gitnexus/src/cli/index.ts +++ b/gitnexus/src/cli/index.ts @@ -70,6 +70,10 @@ program '--worker-timeout ', 'Worker sub-batch idle timeout before retry/fallback. Default: 30.', ) + .option( + '--workers ', + 'Parse worker pool size. Default: cores-1 capped at 16. Pass 0 to disable workers (sequential).', + ) .option('--embedding-threads ', 'Limit local ONNX embedding CPU threads') .option('--embedding-batch-size ', 'Number of nodes per embedding batch') .option('--embedding-sub-batch-size ', 'Number of chunks per embedding model call') @@ -81,6 +85,8 @@ program ' GITNEXUS_MAX_FILE_SIZE=N Override large-file skip threshold (KB). Default 512, max 32768.\n' + ' GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS=N Worker idle timeout in milliseconds. Default 30000.\n' + ' GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES=N Worker job byte budget. Default 8388608.\n' + + ' GITNEXUS_WORKER_POOL_SIZE=N Parse worker count override. Default cores-1 capped at 16.\n' + + ' GITNEXUS_PARSE_CHUNK_CONCURRENCY=N Concurrent in-flight parse chunks. Default 2.\n' + ' GITNEXUS_EMBEDDING_THREADS=N Limit local ONNX CPU threads for --embeddings.\n' + ' GITNEXUS_SEMANTIC_EXACT_SCAN_LIMIT=N Max embedding chunks for exact-scan fallback. Default 10000.\n' + '\nTip: `.gitnexusignore` supports `.gitignore`-style negation. Add e.g.\n' + diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts index 17cfaab3f..73a3c4029 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts @@ -55,6 +55,7 @@ import type { ExtractedCall, ExtractedDecoratorRoute, ExtractedFetchCall, + ExtractedImport, ExtractedORMQuery, ExtractedRoute, ExtractedToolDef, @@ -301,6 +302,16 @@ export async function runChunkedParseAndResolve( const deferredWorkerHeritage: ExtractedHeritage[] = []; const deferredConstructorBindings: FileConstructorBindings[] = []; const deferredAssignments: ExtractedAssignment[] = []; + // Imports accumulated across chunks. Previously processed per-chunk + // via `processImportsFromExtracted` inside the chunk loop, which + // forced workers to sit idle on the main thread's extraction pass + // between chunk dispatches (4-5% CPU utilization symptom). Deferring + // to a single end-of-loop pass lets the worker pool start chunk N+1 + // immediately after chunk N's worker dispatch returns. Resolution is + // strictly-more-information at end-of-loop because graph now has + // every chunk's symbols — improves cross-chunk import targets. + const deferredWorkerImports: ExtractedImport[] = []; + let anyChunkNeedsWildcardSynth = false; // Aggregated per-file ParsedFile artifacts produced by workers' calls // to `extractParsedFile`. Threaded through to the scope-resolution // phase so it can SKIP its own re-extraction on cache hits — this is @@ -409,46 +420,21 @@ export async function runChunkedParseAndResolve( } } - const chunkBasePercent = 20 + (filesParsedSoFar / totalParseable) * 62; - + // Per-chunk extraction passes (processImportsFromExtracted, + // processHeritageFromExtracted, processRoutesFromExtracted, + // synthesizeWildcardImportBindings, seedCrossFileReceiverTypes) + // moved out of the chunk loop into a single end-of-loop pass below. + // Reason: per-chunk extraction blocked the chunk loop on + // main-thread work between worker dispatches — workers sat idle + // and total CPU utilization plateaued at 4-5% on multi-core boxes. + // Deferring keeps workers busy chunk-after-chunk; resolution sees + // strictly-more-information (full repo graph) so cross-chunk import + // and heritage targets resolve at least as well as before. if (chunkWorkerData) { - await processImportsFromExtracted( - graph, - allPathObjects, - chunkWorkerData.imports, - ctx, - (current, total) => { - onProgress({ - phase: 'parsing', - percent: Math.round(chunkBasePercent), - message: `Resolving imports (chunk ${chunkIdx + 1}/${numChunks})...`, - detail: `${current}/${total} files`, - stats: { - filesProcessed: filesParsedSoFar, - totalFiles: totalParseable, - nodesCreated: graph.nodeCount, - }, - }); - }, - repoPath, - importCtx, - ); if (chunkNeedsSynthesis[chunkIdx]) { - synthesizeWildcardImportBindings(graph, ctx); - hasSynthesized = true; - } - if (exportedTypeMap.size > 0 && ctx.namedImportMap.size > 0) { - const { enrichedCount } = seedCrossFileReceiverTypes( - chunkWorkerData.calls, - ctx.namedImportMap, - exportedTypeMap, - ); - if (isDev && enrichedCount > 0) { - logger.info( - `🔗 E1: Seeded ${enrichedCount} cross-file receiver types (chunk ${chunkIdx + 1})`, - ); - } + anyChunkNeedsWildcardSynth = true; } + for (const item of chunkWorkerData.imports) deferredWorkerImports.push(item); for (const item of chunkWorkerData.calls) deferredWorkerCalls.push(item); for (const item of chunkWorkerData.heritage) deferredWorkerHeritage.push(item); for (const item of chunkWorkerData.constructorBindings) @@ -463,35 +449,6 @@ export async function runChunkedParseAndResolve( for (const item of chunkWorkerData.assignments) deferredAssignments.push(item); } - await Promise.all([ - processHeritageFromExtracted(graph, chunkWorkerData.heritage, ctx, (current, total) => { - onProgress({ - phase: 'parsing', - percent: Math.round(chunkBasePercent), - message: `Resolving heritage (chunk ${chunkIdx + 1}/${numChunks})...`, - detail: `${current}/${total} records`, - stats: { - filesProcessed: filesParsedSoFar, - totalFiles: totalParseable, - nodesCreated: graph.nodeCount, - }, - }); - }), - processRoutesFromExtracted(graph, chunkWorkerData.routes ?? [], ctx, (current, total) => { - onProgress({ - phase: 'parsing', - percent: Math.round(chunkBasePercent), - message: `Resolving routes (chunk ${chunkIdx + 1}/${numChunks})...`, - detail: `${current}/${total} routes`, - stats: { - filesProcessed: filesParsedSoFar, - totalFiles: totalParseable, - nodesCreated: graph.nodeCount, - }, - }); - }), - ]); - if (chunkWorkerData.fileScopeBindings?.length) { for (const { filePath, bindings } of chunkWorkerData.fileScopeBindings) { if (typeof filePath !== 'string' || filePath.length === 0) continue; @@ -538,6 +495,83 @@ export async function runChunkedParseAndResolve( ); } + // Deferred end-of-loop extraction (moved out of the per-chunk block): + // 1. processImportsFromExtracted on all chunks' imports + // 2. synthesizeWildcardImportBindings (if any chunk had wildcards) + // 3. seedCrossFileReceiverTypes on deferred calls (depends on + // namedImportMap populated by step 1) + // 4. processHeritageFromExtracted on all chunks' heritage + // 5. processRoutesFromExtracted on all chunks' routes + // Same logic as the prior per-chunk passes, just batched — resolution + // sees the full repo graph instead of just current-and-earlier chunks. + if (deferredWorkerImports.length > 0) { + await processImportsFromExtracted( + graph, + allPathObjects, + deferredWorkerImports, + ctx, + (current, total) => { + onProgress({ + phase: 'parsing', + percent: 82, + message: 'Resolving imports (all chunks)...', + detail: `${current}/${total} files`, + stats: { + filesProcessed: filesParsedSoFar, + totalFiles: totalParseable, + nodesCreated: graph.nodeCount, + }, + }); + }, + repoPath, + importCtx, + ); + } + if (anyChunkNeedsWildcardSynth) { + synthesizeWildcardImportBindings(graph, ctx); + hasSynthesized = true; + } + if (exportedTypeMap.size > 0 && ctx.namedImportMap.size > 0 && deferredWorkerCalls.length > 0) { + const { enrichedCount } = seedCrossFileReceiverTypes( + deferredWorkerCalls, + ctx.namedImportMap, + exportedTypeMap, + ); + if (isDev && enrichedCount > 0) { + logger.info(`🔗 E1: Seeded ${enrichedCount} cross-file receiver types (all chunks)`); + } + } + if (deferredWorkerHeritage.length > 0) { + await processHeritageFromExtracted(graph, deferredWorkerHeritage, ctx, (current, total) => { + onProgress({ + phase: 'parsing', + percent: 82, + message: 'Resolving heritage (all chunks)...', + detail: `${current}/${total} records`, + stats: { + filesProcessed: filesParsedSoFar, + totalFiles: totalParseable, + nodesCreated: graph.nodeCount, + }, + }); + }); + } + if (allExtractedRoutes.length > 0) { + await processRoutesFromExtracted(graph, allExtractedRoutes, ctx, (current, total) => { + onProgress({ + phase: 'parsing', + percent: 82, + message: 'Resolving routes (all chunks)...', + detail: `${current}/${total} routes`, + stats: { + filesProcessed: filesParsedSoFar, + totalFiles: totalParseable, + nodesCreated: graph.nodeCount, + }, + }); + }); + } + const fullWorkerHeritageMap = deferredWorkerHeritage.length > 0 ? buildHeritageMap(deferredWorkerHeritage, ctx, getHeritageStrategyForLanguage) diff --git a/gitnexus/src/core/ingestion/workers/worker-pool.ts b/gitnexus/src/core/ingestion/workers/worker-pool.ts index 9ad0ffc32..18adeba04 100644 --- a/gitnexus/src/core/ingestion/workers/worker-pool.ts +++ b/gitnexus/src/core/ingestion/workers/worker-pool.ts @@ -136,6 +136,15 @@ const DEFAULT_TIMEOUT_BACKOFF_FACTOR = 2; const DEFAULT_MAX_RESPAWNS_PER_SLOT = 3; const DEFAULT_MAX_CUMULATIVE_TIMEOUT_FACTOR = 5; const DEFAULT_CONSECUTIVE_FAILURE_THRESHOLD_FLOOR = 3; +/** + * Default upper bound on auto-resolved pool size. Past 16 workers the + * dominant cost shifts from worker-side parsing to main-thread merge / + * extraction / structured-clone overhead, and the marginal worker adds + * memory pressure (tree-sitter state + sub-batch buffer) without much + * throughput gain. Operators on bigger machines override via + * `GITNEXUS_WORKER_POOL_SIZE` or `--workers `. + */ +const DEFAULT_POOL_SIZE_CAP = 16; function positiveInteger(value: unknown): number | undefined { const parsed = typeof value === 'string' ? Number(value) : value; @@ -195,6 +204,30 @@ export function resolveWorkerPoolOptions( }; } +/** + * Resolve the auto-default worker pool size when no explicit `poolSize` + * arg is passed to `createWorkerPool`. Precedence: + * + * 1. `GITNEXUS_WORKER_POOL_SIZE` env var (operator override; set by + * `--workers ` on the CLI). + * 2. `os.cpus().length - 1`, clamped to `[1, DEFAULT_POOL_SIZE_CAP]`. + * + * The cap exists because past ~16 workers the main-thread merge / + * extraction work and structured-clone overhead dominate; adding more + * worker threads costs memory without much throughput gain. Operators + * who want to push past the cap set the env var explicitly. + * + * Exported for unit tests; production code should not call this + * directly — pass an explicit `poolSize` to `createWorkerPool` or rely + * on the env / default. + */ +export function resolveAutoPoolSize(): number { + const envOverride = nonNegativeInteger(process.env.GITNEXUS_WORKER_POOL_SIZE); + if (envOverride !== undefined) return envOverride; + const cores = os.cpus().length; + return Math.min(DEFAULT_POOL_SIZE_CAP, Math.max(1, cores - 1)); +} + function waitForWorkerOnline(worker: Worker): Promise { return new Promise((resolve, reject) => { const cleanup = () => { @@ -326,7 +359,7 @@ export const createWorkerPool = ( throw new Error(`Worker script not found: ${workerPath}`); } - const size = poolSize ?? Math.min(8, Math.max(1, os.cpus().length - 1)); + const size = poolSize ?? resolveAutoPoolSize(); const poolOptions = resolveWorkerPoolOptions(options, size); const spawnWorker = options?.workerFactory ?? ((url: URL) => new Worker(url)); const workers: (Worker | undefined)[] = new Array(size); diff --git a/gitnexus/test/unit/analyze-worker-pool-size.test.ts b/gitnexus/test/unit/analyze-worker-pool-size.test.ts new file mode 100644 index 000000000..d329ce7b4 --- /dev/null +++ b/gitnexus/test/unit/analyze-worker-pool-size.test.ts @@ -0,0 +1,91 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const runFullAnalysisMock = vi.fn(); + +vi.mock('../../src/core/run-analyze.js', () => ({ + runFullAnalysis: runFullAnalysisMock, +})); + +vi.mock('../../src/core/lbug/lbug-adapter.js', () => ({ + closeLbug: vi.fn(async () => undefined), +})); + +vi.mock('../../src/storage/repo-manager.js', () => ({ + getStoragePaths: vi.fn(() => ({ storagePath: '.gitnexus', lbugPath: '.gitnexus/lbug' })), + getGlobalRegistryPath: vi.fn(() => 'registry.json'), + RegistryNameCollisionError: class RegistryNameCollisionError extends Error {}, + AnalysisNotFinalizedError: class AnalysisNotFinalizedError extends Error {}, + assertAnalysisFinalized: vi.fn(async () => undefined), +})); + +vi.mock('../../src/storage/git.js', () => ({ + getGitRoot: vi.fn(() => '/repo'), + hasGitDir: vi.fn(() => true), +})); + +vi.mock('../../src/core/ingestion/utils/max-file-size.js', () => ({ + getMaxFileSizeBannerMessage: vi.fn(() => null), +})); + +describe('analyzeCommand --workers validation', () => { + beforeEach(() => { + vi.resetModules(); + runFullAnalysisMock.mockReset(); + process.exitCode = undefined; + process.env.NODE_OPTIONS = `${process.env.NODE_OPTIONS ?? ''} --max-old-space-size=8192`.trim(); + delete process.env.GITNEXUS_WORKER_POOL_SIZE; + }); + + it.each(['abc', '-5', '1.5', 'Infinity', 'NaN'])( + 'rejects invalid --workers value %s before analysis starts', + async (workers) => { + const { _captureLogger } = await import('../../src/core/logger.js'); + const cap = _captureLogger(); + const { analyzeCommand } = await import('../../src/cli/analyze.js'); + + await analyzeCommand(undefined, { workers }); + + expect(process.exitCode).toBe(1); + expect( + cap + .records() + .some((r) => + String(r.msg ?? '').startsWith(' --workers must be a non-negative integer'), + ), + ).toBe(true); + expect(runFullAnalysisMock).not.toHaveBeenCalled(); + cap.restore(); + }, + ); + + it('sets GITNEXUS_WORKER_POOL_SIZE for valid positive values', async () => { + const { analyzeCommand } = await import('../../src/cli/analyze.js'); + runFullAnalysisMock.mockResolvedValue({ + repoName: 'repo', + repoPath: '/repo', + stats: {}, + alreadyUpToDate: true, + }); + + await analyzeCommand(undefined, { workers: '12' }); + + expect(process.env.GITNEXUS_WORKER_POOL_SIZE).toBe('12'); + expect(runFullAnalysisMock).toHaveBeenCalled(); + }); + + it('accepts --workers 0 as a sequential-fallback signal', async () => { + const { analyzeCommand } = await import('../../src/cli/analyze.js'); + runFullAnalysisMock.mockResolvedValue({ + repoName: 'repo', + repoPath: '/repo', + stats: {}, + alreadyUpToDate: true, + }); + + await analyzeCommand(undefined, { workers: '0' }); + + expect(process.env.GITNEXUS_WORKER_POOL_SIZE).toBe('0'); + expect(process.exitCode).toBeUndefined(); + expect(runFullAnalysisMock).toHaveBeenCalled(); + }); +}); diff --git a/gitnexus/test/unit/worker-pool-resilience.test.ts b/gitnexus/test/unit/worker-pool-resilience.test.ts index a84b4820e..d97181c75 100644 --- a/gitnexus/test/unit/worker-pool-resilience.test.ts +++ b/gitnexus/test/unit/worker-pool-resilience.test.ts @@ -8,6 +8,7 @@ import { createWorkerPool, WorkerPoolDispatchError, resolveWorkerPoolOptions, + resolveAutoPoolSize, } from '../../src/core/ingestion/workers/worker-pool.js'; /** @@ -488,6 +489,57 @@ describe('worker pool option resolution', () => { }); }); +describe('resolveAutoPoolSize', () => { + it('honors GITNEXUS_WORKER_POOL_SIZE env override (positive integer)', () => { + vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '12'); + try { + expect(resolveAutoPoolSize()).toBe(12); + } finally { + vi.unstubAllEnvs(); + } + }); + + it('honors GITNEXUS_WORKER_POOL_SIZE=0 (sequential-fallback signal)', () => { + vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '0'); + try { + expect(resolveAutoPoolSize()).toBe(0); + } finally { + vi.unstubAllEnvs(); + } + }); + + it('honors GITNEXUS_WORKER_POOL_SIZE override above the auto cap', () => { + vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '32'); + try { + expect(resolveAutoPoolSize()).toBe(32); + } finally { + vi.unstubAllEnvs(); + } + }); + + it('ignores invalid env values and falls back to the auto formula', () => { + vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', 'abc'); + try { + const expected = Math.min(16, Math.max(1, os.cpus().length - 1)); + expect(resolveAutoPoolSize()).toBe(expected); + } finally { + vi.unstubAllEnvs(); + } + }); + + it('matches the auto formula min(16, max(1, cores - 1)) with no env override', () => { + // Exact-count per DoD §2.7: compute the expected value the same + // way the resolver does so the assertion stays deterministic on + // any machine. + const expected = Math.min(16, Math.max(1, os.cpus().length - 1)); + expect(resolveAutoPoolSize()).toBe(expected); + }); + + it('returns an integer (never a float)', () => { + expect(Number.isInteger(resolveAutoPoolSize())).toBe(true); + }); +}); + describe('worker pool option resolution', () => { it('resolves maxRespawnsPerSlot from explicit options', () => { const opts = resolveWorkerPoolOptions({ maxRespawnsPerSlot: 7 }, 4);