feat(mcp): off-load read-tool dispatch to a worker pool to fix concurrent-call timeouts (#1002)
The shared daemon served every session on one event loop with synchronous node:sqlite. codegraph_explore is CPU-bound work stitched together by microtask awaits, so N concurrent explores keep the microtask queue continuously full and starve the macrotask phases — timers AND socket I/O. The transport freezes: no response can flush until the whole batch drains, so with ~10 subagents on a large repo clients routinely time out (reported via X by @symbolic2020). Move the heavy read-tool dispatch onto a worker-thread pool. Each worker holds its own WAL read connection (verified: a worker reader sees the main writer's committed catch-up/watcher writes); the single watcher/writer, the catch-up gate, codegraph_status, and the staleness/worktree notices stay on the main thread. Concurrent reads now run in true parallel up to core count and the main loop stays free for the MCP transport, so responses flush incrementally instead of all-at-once after the batch drains. Enabled for the shared daemon only; direct (single-stdio-client) mode is unchanged. - crash recovery: respawn + retry-once, with a circuit breaker that falls back to in-process dispatch if workers can't run on this platform - graceful backstop: an overloaded pool returns success-shaped "busy, retry" guidance, never isError (so it can't teach the agent to abandon codegraph) - pending-aware growth + capped concurrent cold-starts avoid a startup thundering herd (N simultaneous module-loads + DB opens could stall the loop) - config: CODEGRAPH_QUERY_POOL_SIZE (default clamp(cores-1, 1, 16); 0 disables → in-process), CODEGRAPH_QUERY_BUSY_TIMEOUT_MS (default 45s) 10 concurrent explores on vscode (10.5k files): 31s → ~9s, staggered flush, 0 timeouts, byte-identical output; scales with cores (≈3.3× on 8, 1.8× on 2). Full suite passes plus 10 new query-pool tests (fake-worker injection so the scheduling logic is covered without spawning threads). Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Colby Mchenry
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
4077ed19b7
commit
dfe13b03c8
@@ -0,0 +1,119 @@
|
||||
#!/usr/bin/env node
|
||||
// Reproduction harness A — does the shared daemon serialize concurrent explore?
|
||||
//
|
||||
// Mirrors the daemon's reality: ONE CodeGraph + ONE ToolHandler (as MCPEngine
|
||||
// shares across all sessions), then fires N concurrent codegraph_explore calls
|
||||
// and measures:
|
||||
// - each call's wall-clock latency + completion order
|
||||
// - an event-loop HEARTBEAT (setInterval 50ms): the max gap between ticks is a
|
||||
// direct measure of how long synchronous compute blocked the loop. In the
|
||||
// real daemon a blocked loop can't flush a finished response or read the
|
||||
// next request, so this gap is what starves the MCP transport.
|
||||
//
|
||||
// Usage: node repro-concurrent-explore.mjs <repo-with-.codegraph> <N> [timeoutMs]
|
||||
import { pathToFileURL } from 'node:url';
|
||||
import { resolve } from 'node:path';
|
||||
import { performance } from 'node:perf_hooks';
|
||||
|
||||
const [, , repo, nRaw, timeoutRaw] = process.argv;
|
||||
if (!repo) {
|
||||
console.error('usage: repro-concurrent-explore.mjs <repo> <N=10> [timeoutMs=60000]');
|
||||
process.exit(1);
|
||||
}
|
||||
const N = Number(nRaw) || 10;
|
||||
const TIMEOUT_MS = Number(timeoutRaw) || 60000; // ~ MCP SDK default request timeout
|
||||
|
||||
const load = async (rel) => import(pathToFileURL(resolve(rel)).href);
|
||||
const idx = await load('dist/index.js');
|
||||
const tools = await load('dist/mcp/tools.js');
|
||||
const CodeGraph = idx.default?.default ?? idx.default ?? idx.CodeGraph;
|
||||
const ToolHandler = tools.ToolHandler ?? tools.default?.ToolHandler;
|
||||
|
||||
// Distinct queries so no two calls are trivially identical. Mix of NL questions
|
||||
// (exercise FTS + RWR over the whole graph) — the expensive explore path.
|
||||
const QUERIES = [
|
||||
'how does the text model handle edits and undo',
|
||||
'how does the file service watch for changes on disk',
|
||||
'how does the keybinding service resolve a chord to a command',
|
||||
'how does the extension host activate an extension',
|
||||
'how does the editor render decorations in the viewport',
|
||||
'how does the search service stream results to the UI',
|
||||
'how does the terminal process manager spawn a shell',
|
||||
'how does the configuration service merge user and workspace settings',
|
||||
'how does the debug adapter forward breakpoints to the runtime',
|
||||
'how does the quick input widget filter its items',
|
||||
'how does the notification service queue and show toasts',
|
||||
'how does the git extension compute the diff for a file',
|
||||
'how does the language features registry dispatch a hover request',
|
||||
'how does the workbench layout restore editor groups on reload',
|
||||
'how does the storage service persist state between sessions',
|
||||
'how does the menu service build a context menu from contributions',
|
||||
];
|
||||
|
||||
const cg = CodeGraph.openSync(repo);
|
||||
let fileCount = 0;
|
||||
try { fileCount = cg.getStats().fileCount; } catch {}
|
||||
const handler = new ToolHandler(cg);
|
||||
|
||||
// --- event-loop heartbeat ---
|
||||
let lastTick = performance.now();
|
||||
let maxGap = 0;
|
||||
const gaps = [];
|
||||
const hb = setInterval(() => {
|
||||
const now = performance.now();
|
||||
const gap = now - lastTick;
|
||||
lastTick = now;
|
||||
if (gap > 60) gaps.push(Math.round(gap)); // expected ~50ms; record stalls
|
||||
if (gap > maxGap) maxGap = gap;
|
||||
}, 50);
|
||||
|
||||
function runOne(i) {
|
||||
const q = QUERIES[i % QUERIES.length];
|
||||
const startedAt = performance.now();
|
||||
let timer;
|
||||
const timeout = new Promise((res) => {
|
||||
timer = setTimeout(() => res({ timedOut: true }), TIMEOUT_MS);
|
||||
});
|
||||
const work = handler
|
||||
.execute('codegraph_explore', { query: q })
|
||||
.then((r) => ({ ok: !r.isError, chars: r.content?.[0]?.text?.length ?? 0 }))
|
||||
.catch((e) => ({ ok: false, err: String(e?.message ?? e) }));
|
||||
return Promise.race([work, timeout]).then((r) => {
|
||||
clearTimeout(timer);
|
||||
return { i, q, ms: Math.round(performance.now() - startedAt), ...r };
|
||||
});
|
||||
}
|
||||
|
||||
// Baseline: one warm single call (so the first-call cold paths don't skew N).
|
||||
const warmStart = performance.now();
|
||||
await runOne(0);
|
||||
const warmMs = Math.round(performance.now() - warmStart);
|
||||
|
||||
// Reset heartbeat stats for the concurrent run.
|
||||
gaps.length = 0; maxGap = 0; lastTick = performance.now();
|
||||
|
||||
const batchStart = performance.now();
|
||||
const results = await Promise.all(Array.from({ length: N }, (_, i) => runOne(i)));
|
||||
const batchMs = Math.round(performance.now() - batchStart);
|
||||
clearInterval(hb);
|
||||
|
||||
const lat = results.map((r) => r.ms).sort((a, b) => a - b);
|
||||
const timeouts = results.filter((r) => r.timedOut).length;
|
||||
const p = (q) => lat[Math.min(lat.length - 1, Math.floor(q * lat.length))];
|
||||
|
||||
console.log('='.repeat(64));
|
||||
console.log(`repo=${repo}`);
|
||||
console.log(`fileCount=${fileCount} N=${N} perCallTimeout=${TIMEOUT_MS}ms`);
|
||||
console.log(`single warm explore: ${warmMs}ms`);
|
||||
console.log('-'.repeat(64));
|
||||
console.log(`concurrent batch wall-clock: ${batchMs}ms`);
|
||||
console.log(`per-call latency min=${lat[0]} p50=${p(0.5)} p90=${p(0.9)} max=${lat[lat.length - 1]} (ms)`);
|
||||
console.log(`TIMEOUTS (>${TIMEOUT_MS}ms): ${timeouts} / ${N}`);
|
||||
console.log(`event-loop max stall: ${Math.round(maxGap)}ms stalls>60ms: ${gaps.length}`);
|
||||
console.log(` sum of stalls: ${gaps.reduce((a, b) => a + b, 0)}ms biggest 5: ${gaps.sort((a,b)=>b-a).slice(0,5).join(', ')}`);
|
||||
console.log('-'.repeat(64));
|
||||
console.log('SERIALIZATION CHECK:');
|
||||
console.log(` if serialized, batch ≈ N×single = ~${N * warmMs}ms; actual=${batchMs}ms (ratio ${(batchMs / (N * warmMs)).toFixed(2)})`);
|
||||
console.log(` max latency / single = ${(lat[lat.length - 1] / warmMs).toFixed(1)}× (≈N means last call waited for all others)`);
|
||||
console.log('='.repeat(64));
|
||||
try { cg.close?.(); } catch {}
|
||||
@@ -0,0 +1,125 @@
|
||||
#!/usr/bin/env node
|
||||
// Reproduction harness B — the FAITHFUL opencode scenario.
|
||||
//
|
||||
// Spawns N real `codegraph serve --mcp --path <repo>` processes (each becomes a
|
||||
// proxy that attaches to ONE shared daemon — exactly what opencode does with N
|
||||
// subagents), drives clean MCP JSON-RPC over each child's stdio, then fires ONE
|
||||
// concurrent wave of codegraph_explore tools/call across all N and measures
|
||||
// end-to-end latency + timeouts. This captures transport-flush starvation: a
|
||||
// daemon event-loop blocked in synchronous explore compute can neither read the
|
||||
// next request nor flush a finished response.
|
||||
//
|
||||
// Usage: node repro-daemon-clients.mjs <repo> <N=10> [perCallTimeoutMs=60000] [warm=1]
|
||||
import { spawn } from 'node:child_process';
|
||||
import { performance } from 'node:perf_hooks';
|
||||
import { resolve } from 'node:path';
|
||||
|
||||
const [, , repoRaw, nRaw, timeoutRaw, warmRaw] = process.argv;
|
||||
const repo = resolve(repoRaw || '.');
|
||||
const N = Number(nRaw) || 10;
|
||||
const TIMEOUT_MS = Number(timeoutRaw) || 60000;
|
||||
const WARM = warmRaw === undefined ? true : warmRaw !== '0';
|
||||
const CLI = resolve('dist/bin/codegraph.js');
|
||||
|
||||
const QUERIES = [
|
||||
'how does the text model handle edits and undo',
|
||||
'how does the file service watch for changes on disk',
|
||||
'how does the keybinding service resolve a chord to a command',
|
||||
'how does the extension host activate an extension',
|
||||
'how does the editor render decorations in the viewport',
|
||||
'how does the search service stream results to the UI',
|
||||
'how does the terminal process manager spawn a shell',
|
||||
'how does the configuration service merge user and workspace settings',
|
||||
'how does the debug adapter forward breakpoints to the runtime',
|
||||
'how does the quick input widget filter its items',
|
||||
'how does the notification service queue and show toasts',
|
||||
'how does the git extension compute the diff for a file',
|
||||
];
|
||||
|
||||
function makeClient(id) {
|
||||
const child = spawn('node', [CLI, 'serve', '--mcp', '--path', repo], {
|
||||
env: { ...process.env, CODEGRAPH_TELEMETRY: '0', DO_NOT_TRACK: '1', CODEGRAPH_MCP_LOG_ATTACH: '0' },
|
||||
stdio: ['pipe', 'pipe', 'inherit'],
|
||||
});
|
||||
let buf = '';
|
||||
const waiters = new Map(); // id -> resolve
|
||||
child.stdout.setEncoding('utf8');
|
||||
child.stdout.on('data', (chunk) => {
|
||||
buf += chunk;
|
||||
let idx;
|
||||
while ((idx = buf.indexOf('\n')) !== -1) {
|
||||
const line = buf.slice(0, idx).trim();
|
||||
buf = buf.slice(idx + 1);
|
||||
if (!line) continue;
|
||||
let msg; try { msg = JSON.parse(line); } catch { continue; }
|
||||
if (msg.id !== undefined && waiters.has(msg.id)) {
|
||||
waiters.get(msg.id)(msg);
|
||||
waiters.delete(msg.id);
|
||||
}
|
||||
}
|
||||
});
|
||||
const send = (obj) => child.stdin.write(JSON.stringify(obj) + '\n');
|
||||
const request = (method, params, rpcId, timeoutMs) =>
|
||||
new Promise((res) => {
|
||||
let timer;
|
||||
if (timeoutMs) timer = setTimeout(() => { waiters.delete(rpcId); res({ __timeout: true }); }, timeoutMs);
|
||||
waiters.set(rpcId, (m) => { if (timer) clearTimeout(timer); res(m); });
|
||||
send({ jsonrpc: '2.0', id: rpcId, method, params });
|
||||
});
|
||||
return { id, child, send, request };
|
||||
}
|
||||
|
||||
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
const clients = Array.from({ length: N }, (_, i) => makeClient(i));
|
||||
|
||||
// Initialize every client (handshake is answered locally by each proxy, instant).
|
||||
await Promise.all(clients.map((c) =>
|
||||
c.request('initialize', { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 'repro', version: '1' } }, `init-${c.id}`, 10000)
|
||||
.then(() => c.send({ jsonrpc: '2.0', method: 'initialized' }))
|
||||
));
|
||||
|
||||
// Warm the daemon: one explore through client 0 forces daemon spawn + project
|
||||
// open + catch-up gate to complete, so the concurrent wave measures the STEADY
|
||||
// state (the user's real scenario after the first call), not cold start.
|
||||
if (WARM) {
|
||||
process.stderr.write('[repro] warming daemon (first explore triggers spawn+open+catchup)...\n');
|
||||
const t0 = performance.now();
|
||||
const r = await clients[0].request('tools/call', { name: 'codegraph_explore', arguments: { query: QUERIES[0] } }, 'warm-0', 120000);
|
||||
process.stderr.write(`[repro] warm explore took ${Math.round(performance.now() - t0)}ms (timeout=${!!r.__timeout})\n`);
|
||||
await sleep(500);
|
||||
}
|
||||
|
||||
// THE WAVE: fire one explore on every client as simultaneously as possible.
|
||||
process.stderr.write(`[repro] firing ${N} concurrent explores...\n`);
|
||||
const waveStart = performance.now();
|
||||
const results = await Promise.all(clients.map((c, i) => {
|
||||
const started = performance.now();
|
||||
return c.request('tools/call', { name: 'codegraph_explore', arguments: { query: QUERIES[i % QUERIES.length] } }, `call-${c.id}`, TIMEOUT_MS)
|
||||
.then((m) => ({
|
||||
id: c.id,
|
||||
ms: Math.round(performance.now() - started),
|
||||
timedOut: !!m.__timeout,
|
||||
ok: !!m.result && !m.result.isError,
|
||||
chars: m.result?.content?.[0]?.text?.length ?? 0,
|
||||
}));
|
||||
}));
|
||||
const waveMs = Math.round(performance.now() - waveStart);
|
||||
|
||||
const lat = results.map((r) => r.ms).sort((a, b) => a - b);
|
||||
const timeouts = results.filter((r) => r.timedOut).length;
|
||||
const p = (q) => lat[Math.min(lat.length - 1, Math.floor(q * lat.length))];
|
||||
|
||||
console.log('='.repeat(64));
|
||||
console.log(`HARNESS B (real daemon + ${N} proxies) repo=${repo}`);
|
||||
console.log(`warm=${WARM} perCallTimeout=${TIMEOUT_MS}ms`);
|
||||
console.log('-'.repeat(64));
|
||||
console.log(`wave wall-clock: ${waveMs}ms`);
|
||||
console.log(`per-call latency min=${lat[0]} p50=${p(0.5)} p90=${p(0.9)} max=${lat[lat.length - 1]} (ms)`);
|
||||
console.log(`TIMEOUTS (>${TIMEOUT_MS}ms): ${timeouts} / ${N}`);
|
||||
console.log(`completion order (id:ms): ${results.slice().sort((a,b)=>a.ms-b.ms).map(r=>`${r.id}:${r.ms}`).join(' ')}`);
|
||||
console.log('='.repeat(64));
|
||||
|
||||
for (const c of clients) { try { c.child.stdin.end(); c.child.kill('SIGTERM'); } catch {} }
|
||||
await sleep(300);
|
||||
process.exit(0);
|
||||
Reference in New Issue
Block a user