From 28068fa0f13724ef974e238423aa04e664fdeb16 Mon Sep 17 00:00:00 2001 From: Colby McHenry Date: Thu, 16 Jul 2026 23:38:32 -0500 Subject: [PATCH] =?UTF-8?q?perf(kernel):=20direct-to-store=20decode=20?= =?UTF-8?q?=E2=80=94=20buffers=20flow=20to=20the=20store=20worker,=20main?= =?UTF-8?q?=20thread=20never=20materializes=20nodes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kernel-routed files ship their flat tables from the parse worker to the store worker as buffers (tryKernelExtractRaw → kernelBuffers on the result → KernelStoreBundle); the store worker decodes and finalizes (finalizeStoreBundle shared with the object path so filter semantics can never drift). Files with applicable framework extract() hooks keep the decoded path; the no-writer fallback materializes via materializeKernelResult. Byte-identical dumps re-verified on dubbo, excalidraw, express, gson; full suite green (2,467). Measurement (plan §4d): dubbo's parse-loop wall is 94% store-writer busy time — the many-core fresh-index wall is single-writer SQLite ingest, not extraction or main-thread work. d2s improves the writer lane ~11% (structured-clone deserialization avoided on the writer) and frees the main thread; the remaining many-core gap is a store-architecture arc (deferred index builds, multi-file transactions, buffer→bind), out of the kernel project's scope. Co-Authored-By: Claude Fable 5 --- docs/design/rust-kernel-migration-plan.md | 25 +++++++ src/extraction/index.ts | 85 ++++++++++++++--------- src/extraction/kernel/index.ts | 83 ++++++++++++++++++++++ src/extraction/parse-worker.ts | 33 ++++++++- src/extraction/store-worker.ts | 25 ++++++- src/extraction/store-writer.ts | 48 ++++++++++++- src/types.ts | 17 +++++ 7 files changed, 277 insertions(+), 39 deletions(-) diff --git a/docs/design/rust-kernel-migration-plan.md b/docs/design/rust-kernel-migration-plan.md index 593b66e..68806ed 100644 --- a/docs/design/rust-kernel-migration-plan.md +++ b/docs/design/rust-kernel-migration-plan.md @@ -264,6 +264,31 @@ Default routing: `DEFAULT_ROUTED = {typescript, tsx, javascript, jsx}` in Windows VM still deferred (same fallback rationale as §4b). - Default routing now includes `java`. +### 4d. Direct-to-store decode (2026-07-16) — and where the wall ACTUALLY is + +Kernel-routed files now ship their flat buffers from the parse worker all the way +to the STORE WORKER, which decodes + finalizes them there (`tryKernelExtractRaw` → +`ExtractionResult.kernelBuffers` → `KernelStoreBundle` → `decodeKernelBundle`; +filter semantics shared via `finalizeStoreBundle`). The main thread's per-file work +drops to O(1) + the content hash — it never materializes per-node objects, and both +postMessage hops move flat bytes instead of object graphs. Files whose applicable +frameworks carry an `extract()` hook keep the decoded path (hooks merge into decoded +results); non-writer paths (main-thread store, tests) materialize via +`materializeKernelResult`. Byte-identical dumps re-verified on dubbo, excalidraw, +express, gson. + +**Measurement that closes the §4c question:** with the store worker instrumented, +dubbo's parse-loop wall is **94% store-writer busy time** (4,202ms of 4,493ms on the +kernel arm). The many-core fresh-index wall is the single-writer SQLite ingest — +not extraction, not main-thread work. d2s still improves the writer lane ~11% +(4,726→4,202ms: buffers skip structured-clone deserialization ON the writer) and +frees the main thread, but the remaining cbm gap on many-core medium repos is a +STORE-ARCHITECTURE question (their RAM-first design defers all durability). Next +levers there (a separate perf arc, not this project): deferred/bulk index builds +during the parse phase, multi-file write transactions, buffer→bind without object +materialization. Note the #1320-arc post-mortem already measured statement batching +and sorted inserts as ~zero on this path — B-tree maintenance is the floor. + ## 4. Per-language tracker Tiers: **T1** = mostly `.scm` + mapping config. **T2** = needs bespoke pre/post passes kept diff --git a/src/extraction/index.ts b/src/extraction/index.ts index 4f45b2e..185e45c 100644 --- a/src/extraction/index.ts +++ b/src/extraction/index.ts @@ -23,7 +23,8 @@ import { import { QueryBuilder } from '../db/queries'; import { extractFromSource } from './tree-sitter'; import { ParseWorkerPool, resolveParsePoolSize, resolveParseTimeoutMs } from './parse-pool'; -import { StoreWriter, StoreBundle } from './store-writer'; +import { StoreWriter, StoreBundle, finalizeStoreBundle } from './store-writer'; +import { materializeKernelResult } from './kernel'; import { detectLanguage, isSourceFile, isLanguageSupported, isFileLevelOnlyLanguage, initGrammars, loadGrammarsForLanguages, readGrammarWasmBytes } from './grammars'; import { loadExtensionOverrides, loadIncludeIgnoredPatterns, loadExcludePatterns, loadIncludePatterns } from '../project-config'; import { isCodeGraphDataDir } from '../directory'; @@ -1706,16 +1707,34 @@ export class ExtractionOrchestrator { const bp = walBackpressure?.(); if (bp) await bp; + // Kernel deferred-decode results carry table sizes in kernelCounts + // (their object arrays are empty — decode happens at the store). + const nodeCount = result.kernelCounts?.nodes ?? result.nodes.length; + const edgeCount = result.kernelCounts?.edges ?? result.edges.length; + // Store: on the writer thread when active (fresh DB — bundles applied // in the same file order this chain dispatches them), else on the main // thread (SQLite connections are per-thread). - if (result.nodes.length > 0 || result.errors.length === 0) { + if (nodeCount > 0 || result.errors.length === 0) { const language = detectLanguage(filePath, content, overrides); if (storeWriter) { - storeWriter.send(this.buildFreshStoreBundle(filePath, content, language, stats, result)); + if (result.kernelBuffers) { + // Buffers go to the writer as-is; the worker decodes + finalizes. + // The main thread's only per-file work stays O(1) + the content hash. + storeWriter.send({ + kernel: true, + filePath, + language, + buffers: result.kernelBuffers, + file: this.buildFileRecord(filePath, content, language, stats, nodeCount, result.errors), + }); + } else { + storeWriter.send(this.buildFreshStoreBundle(filePath, content, language, stats, result)); + } await storeWriter.waitBelow(STORE_WRITER_WINDOW); } else { - await this.storeExtractionResult(filePath, content, language, stats, result, commitYield); + const materialized = materializeKernelResult(result, filePath, language); + await this.storeExtractionResult(filePath, content, language, stats, materialized, commitYield); } } @@ -1726,10 +1745,10 @@ export class ExtractionOrchestrator { errors.push(...result.errors); } - if (result.nodes.length > 0) { + if (nodeCount > 0) { filesIndexed++; - totalNodes += result.nodes.length; - totalEdges += result.edges.length; + totalNodes += nodeCount; + totalEdges += edgeCount; } else if (result.errors.some((e) => e.severity === 'error')) { filesErrored++; } else { @@ -2383,6 +2402,27 @@ export class ExtractionOrchestrator { * check, no cross-file edge snapshot (both are re-index concerns — a fresh * database has neither). Filters mirror storeExtractionResult exactly. */ + /** The FileRecord for a fresh-index store (nodeCount is the PRE-filter count). */ + private buildFileRecord( + filePath: string, + content: string, + language: Language, + stats: fs.Stats, + nodeCount: number, + resultErrors: ExtractionResult['errors'] + ): FileRecord { + return { + path: filePath, + contentHash: hashContent(content), + language, + size: stats.size, + modifiedAt: stats.mtimeMs, + indexedAt: Date.now(), + nodeCount, + errors: resultErrors.length > 0 ? resultErrors : undefined, + }; + } + private buildFreshStoreBundle( filePath: string, content: string, @@ -2390,33 +2430,12 @@ export class ExtractionOrchestrator { stats: fs.Stats, result: ExtractionResult ): StoreBundle { - const validNodes = result.nodes.filter((n) => n.id && n.kind && n.name && n.filePath && n.language); - const insertedIds = new Set(validNodes.map((n) => n.id)); - const validEdges = result.edges.filter( - (e) => insertedIds.has(e.source) && insertedIds.has(e.target) + return finalizeStoreBundle( + result, + filePath, + language, + this.buildFileRecord(filePath, content, language, stats, result.nodes.length, result.errors) ); - const validRefs = result.unresolvedReferences - .filter((ref) => insertedIds.has(ref.fromNodeId)) - .map((ref) => ({ - ...ref, - filePath: ref.filePath ?? filePath, - language: ref.language ?? language, - })); - return { - nodes: validNodes, - edges: validEdges, - refs: validRefs, - file: { - path: filePath, - contentHash: hashContent(content), - language, - size: stats.size, - modifiedAt: stats.mtimeMs, - indexedAt: Date.now(), - nodeCount: result.nodes.length, - errors: result.errors.length > 0 ? result.errors : undefined, - }, - }; } /** diff --git a/src/extraction/kernel/index.ts b/src/extraction/kernel/index.ts index 3024ea8..96df323 100644 --- a/src/extraction/kernel/index.ts +++ b/src/extraction/kernel/index.ts @@ -17,6 +17,11 @@ import type { ExtractionResult, Language } from '../../types'; import { getKernel, kernelSupports } from './loader'; import { decodeExtractBuffers } from './decode'; +import { + KERNEL_ABI_VERSION as LAYOUT_ABI, + META as LAYOUT_META, + NONE as LAYOUT_NONE, +} from './layout'; export { getKernel, kernelSupports, resetKernelForTests } from './loader'; export { decodeExtractBuffers } from './decode'; @@ -66,6 +71,84 @@ export function kernelRoutes(language: Language): boolean { /** Warned-once registry so a broken language logs a single line, not one per file. */ const warned = new Set(); +/** The raw table buffers + the cheap facts the orchestrator needs pre-decode. */ +export interface KernelRawResult { + buffers: NonNullable; + counts: { nodes: number; edges: number; refs: number }; + errors: ExtractionResult['errors']; +} + +/** + * Extract via the kernel WITHOUT decoding — the bulk-index fast path. The + * tables ride to the store boundary as buffers (decoded on the store worker), + * so the main thread never materializes per-node objects. Returns null under + * exactly the conditions tryKernelExtract does, PLUS when the language has a + * registered post() pass (post passes operate on decoded results, so those + * languages keep the decoded path). + */ +export function tryKernelExtractRaw( + filePath: string, + source: string, + language: Language +): KernelRawResult | null { + if (!kernelRoutes(language) || POST_PASSES[language]) return null; + const kernel = getKernel(); + if (!kernel) return null; + try { + const buffers = kernel.extractFile(filePath, source, language); + const meta = buffers.meta; + if (meta.readUInt8(LAYOUT_META.version) !== LAYOUT_ABI) { + throw new Error(`kernel buffer ABI ${meta.readUInt8(0)} != expected ${LAYOUT_ABI}`); + } + const counts = { + nodes: meta.readUInt32LE(LAYOUT_META.nodeCount), + edges: meta.readUInt32LE(LAYOUT_META.edgeCount), + refs: meta.readUInt32LE(LAYOUT_META.refCount), + }; + let errors: ExtractionResult['errors'] = []; + const errorsOff = meta.readUInt32LE(LAYOUT_META.errorsOff); + if (errorsOff !== LAYOUT_NONE) { + const errorsLen = meta.readUInt32LE(LAYOUT_META.errorsLen); + errors = JSON.parse( + buffers.arena.toString('utf8', errorsOff, errorsOff + errorsLen) + ) as ExtractionResult['errors']; + } + return { buffers, counts, errors }; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + if (message.includes('defer:')) return null; + if (!warned.has(language)) { + warned.add(language); + process.stderr.write( + `[codegraph-kernel] ${language} extraction failed (${message}) — falling back to the wasm path\n` + ); + } + return null; + } +} + +/** + * Decode a buffer-carrying result (see ExtractionResult.kernelBuffers) into a + * plain, fully-materialized ExtractionResult — the fallback for store paths + * that need objects (main-thread store, tests). + */ +export function materializeKernelResult( + result: ExtractionResult, + filePath: string, + language: Language +): ExtractionResult { + if (!result.kernelBuffers) return result; + const b = result.kernelBuffers; + const asBuf = (u: Uint8Array) => Buffer.from(u.buffer, u.byteOffset, u.byteLength); + const decoded = decodeExtractBuffers( + { meta: asBuf(b.meta), nodes: asBuf(b.nodes), edges: asBuf(b.edges), refs: asBuf(b.refs), arena: asBuf(b.arena) }, + filePath, + language + ); + decoded.durationMs = result.durationMs; + return decoded; +} + /** * Extract via the native kernel. Returns null when the kernel doesn't apply * (not routed / not available / kill switch) — the caller falls back to the diff --git a/src/extraction/parse-worker.ts b/src/extraction/parse-worker.ts index d086e32..b5be212 100644 --- a/src/extraction/parse-worker.ts +++ b/src/extraction/parse-worker.ts @@ -16,6 +16,8 @@ try { import { parentPort } from 'worker_threads'; import { extractFromSource } from './tree-sitter'; import { detectLanguage, loadGrammarsForLanguages, resetParser } from './grammars'; +import { tryKernelExtractRaw } from './kernel'; +import { getAllFrameworkResolvers, getApplicableFrameworks } from '../resolution/frameworks'; import type { Language, ExtractionResult } from '../types'; // Emscripten prints `Aborted()` (and a follow-up RuntimeError diag @@ -80,7 +82,36 @@ parentPort!.on('message', async (msg: { type: string; id?: number; filePath?: st // codegraph.json extension overrides) and sends it; fall back to detection // for older callers / safety. const language = msg.language ?? detectLanguage(filePath!, content); - const result: ExtractionResult = extractFromSource(filePath!, content!, language, frameworkNames); + + // Kernel deferred-decode fast path: ship the file's tables as flat + // buffers and decode at the STORE boundary, so the main thread never + // materializes per-node objects (nor pays their structured-clone cost — + // buffer clone is a flat memcpy). Only when no applicable framework has + // an extract() hook: those merge extra nodes/refs into the DECODED + // result inside extractFromSource, so such files keep the decoded path. + let result: ExtractionResult | undefined; + const frameworksNeedDecode = + frameworkNames && frameworkNames.length > 0 + ? getApplicableFrameworks( + getAllFrameworkResolvers().filter((r) => frameworkNames.includes(r.name)), + language + ).some((fw) => !!fw.extract) + : false; + if (!frameworksNeedDecode) { + const raw = tryKernelExtractRaw(filePath!, content!, language); + if (raw) { + result = { + nodes: [], + edges: [], + unresolvedReferences: [], + errors: raw.errors, + durationMs: 0, + kernelBuffers: raw.buffers, + kernelCounts: raw.counts, + }; + } + } + result ??= extractFromSource(filePath!, content!, language, frameworkNames); // Periodic parser reset to reclaim WASM heap memory const count = (parseCounts.get(language) ?? 0) + 1; diff --git a/src/extraction/store-worker.ts b/src/extraction/store-worker.ts index 74b5a1d..8413ac7 100644 --- a/src/extraction/store-worker.ts +++ b/src/extraction/store-worker.ts @@ -30,7 +30,8 @@ try { import { parentPort } from 'worker_threads'; import { QueryBuilder } from '../db/queries'; import { createDatabase, SqliteDatabase } from '../db/sqlite-adapter'; -import type { StoreBundle } from './store-writer'; +import { finalizeStoreBundle, type KernelStoreBundle, type StoreBundle } from './store-writer'; +import { decodeExtractBuffers } from './kernel/decode'; if (!parentPort) { throw new Error('store-worker must be run as a worker thread'); @@ -42,10 +43,27 @@ let queries: QueryBuilder | null = null; type InMessage = | { type: 'open'; dbPath: string; fastInit: boolean } - | { type: 'bundle'; bundle: StoreBundle } + | { type: 'bundle'; bundle: StoreBundle | KernelStoreBundle } | { type: 'drain'; id: number } | { type: 'close' }; +/** Decode a kernel bundle's buffers into the standard pre-filtered StoreBundle. */ +function decodeKernelBundle(bundle: KernelStoreBundle): StoreBundle { + const asBuf = (u: Uint8Array) => Buffer.from(u.buffer, u.byteOffset, u.byteLength); + const decoded = decodeExtractBuffers( + { + meta: asBuf(bundle.buffers.meta), + nodes: asBuf(bundle.buffers.nodes), + edges: asBuf(bundle.buffers.edges), + refs: asBuf(bundle.buffers.refs), + arena: asBuf(bundle.buffers.arena), + }, + bundle.filePath, + bundle.language + ); + return finalizeStoreBundle(decoded, bundle.filePath, bundle.language, bundle.file); +} + port.on('message', (msg: InMessage) => { try { switch (msg.type) { @@ -70,7 +88,8 @@ port.on('message', (msg: InMessage) => { } case 'bundle': { if (!queries) throw new Error('store-worker: bundle before open'); - queries.storeFileBundle(msg.bundle); + const bundle = 'kernel' in msg.bundle ? decodeKernelBundle(msg.bundle) : msg.bundle; + queries.storeFileBundle(bundle); port.postMessage({ type: 'ack' }); break; } diff --git a/src/extraction/store-writer.ts b/src/extraction/store-writer.ts index f0372d8..a766bd8 100644 --- a/src/extraction/store-writer.ts +++ b/src/extraction/store-writer.ts @@ -8,7 +8,7 @@ */ import { Worker } from 'worker_threads'; -import { Node, Edge, UnresolvedReference, FileRecord } from '../types'; +import { ExtractionResult, Language, Node, Edge, UnresolvedReference, FileRecord } from '../types'; /** One file's complete store payload (pre-filtered — see storeFileBundle). */ export interface StoreBundle { @@ -18,6 +18,50 @@ export interface StoreBundle { file: FileRecord; } +/** + * A kernel deferred-decode payload: the file's raw table buffers plus the + * FileRecord the main thread built from meta counts. The store WORKER decodes + * and finalizes (same filters as the object path), so per-node objects never + * exist on the main thread. + */ +export interface KernelStoreBundle { + kernel: true; + filePath: string; + language: Language; + buffers: NonNullable; + file: FileRecord; +} + +/** + * The validation/denormalization every bundle gets before storeFileBundle — + * shared by the orchestrator's object path and the store worker's kernel + * decode path so the two can never drift: + * - nodes missing identity fields are dropped (#42-class safety), + * - edges must connect inserted nodes (FK integrity), + * - refs must originate from inserted nodes and carry the denormalized + * filePath/language the resolver reads. + */ +export function finalizeStoreBundle( + result: Pick, + filePath: string, + language: Language, + file: FileRecord +): StoreBundle { + const validNodes = result.nodes.filter((n) => n.id && n.kind && n.name && n.filePath && n.language); + const insertedIds = new Set(validNodes.map((n) => n.id)); + const validEdges = result.edges.filter( + (e) => insertedIds.has(e.source) && insertedIds.has(e.target) + ); + const validRefs = result.unresolvedReferences + .filter((ref) => insertedIds.has(ref.fromNodeId)) + .map((ref) => ({ + ...ref, + filePath: ref.filePath ?? filePath, + language: ref.language ?? language, + })); + return { nodes: validNodes, edges: validEdges, refs: validRefs, file }; +} + export class StoreWriter { private worker: Worker; private readyPromise: Promise; @@ -102,7 +146,7 @@ export class StoreWriter { } /** Post one file's bundle. Throws immediately if the writer already failed. */ - send(bundle: StoreBundle): void { + send(bundle: StoreBundle | KernelStoreBundle): void { if (this.firstError) throw this.firstError; if (this.exited) throw new Error('store worker already exited'); this.outstanding++; diff --git a/src/types.ts b/src/types.ts index 443f608..5b0e407 100644 --- a/src/types.ts +++ b/src/types.ts @@ -276,6 +276,23 @@ export interface ExtractionResult { /** Extraction duration in milliseconds */ durationMs: number; + + /** + * Deferred-decode transport (native kernel, bulk-index path): when present, + * `nodes`/`edges`/`unresolvedReferences` are EMPTY and the file's tables + * ride as flat buffers to be decoded at the store boundary (the store + * worker), so the MAIN thread never materializes per-node objects. + * `kernelCounts` carries the table sizes for bookkeeping. Decode into a + * plain result with `materializeKernelResult` (src/extraction/kernel). + */ + kernelBuffers?: { + meta: Uint8Array; + nodes: Uint8Array; + edges: Uint8Array; + refs: Uint8Array; + arena: Uint8Array; + }; + kernelCounts?: { nodes: number; edges: number; refs: number }; } /**