From 12caa6706bf6f442a9e6055cfb7f8d6301acfed1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9E=97=E5=87=AF90331?= <90331@sangfor.com> Date: Thu, 14 May 2026 11:07:02 +0800 Subject: [PATCH] perf(rawDump): reduce CPU usage and fix worker spawn reliability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add session messages and repo info caching to avoid repeated JSONL parsing and git subprocess spawning across tasks. Share state across all tasks in a single batch to minimize file lock contention and JSON parse overhead. Convert queue operations from sync to async to prevent event loop blocking in the main process. Restore inline batch worker fallback when external spawn fails, and increase worker interval from 30s to 2min to reduce CPU impact. Add 5s debounce per session+messageID to prevent duplicate enqueues. Improve binary mode spawn reliability by falling back to node runtime when bun is not available. Signed-off-by: 林凯90331 <90331@sangfor.com> Co-authored-by: CoStrict --- src/services/rawDump/batchWorker.ts | 71 +++++++++++++++++++++++------ src/services/rawDump/index.ts | 23 +++++++++- src/services/rawDump/queue.ts | 29 ++++++------ src/services/rawDump/spawn.ts | 2 + src/services/rawDump/worker.ts | 11 ++++- 5 files changed, 105 insertions(+), 31 deletions(-) diff --git a/src/services/rawDump/batchWorker.ts b/src/services/rawDump/batchWorker.ts index 561089cae..7086929e3 100644 --- a/src/services/rawDump/batchWorker.ts +++ b/src/services/rawDump/batchWorker.ts @@ -13,7 +13,24 @@ import { createLogger } from './logger.js' const log = createLogger('raw-dump-batch') -const BATCH_INTERVAL_MS = 30_000 // 每轮间隔 +type RepoInfo = Awaited> + +const BATCH_INTERVAL_MS = 120_000 // 每轮间隔（2 分钟），降低内联运行时的 CPU 影响 + +// Git repo 信息缓存，同一 directory 的多个 task 短时间内不需要重复 spawn git +const repoInfoCache = new Map() +const REPO_CACHE_TTL_MS = 60_000 + +async function getCachedRepoInfo(directory: string): Promise { + const cached = repoInfoCache.get(directory) + if (cached && Date.now() - cached.ts < REPO_CACHE_TTL_MS) { + log.debug('repo info cache hit', { directory }) + return cached.repoInfo + } + const repoInfo = await getRepoInfo(directory) + repoInfoCache.set(directory, { repoInfo, ts: Date.now() }) + return repoInfo +} // 进程内重入保护：文件锁不防同进程重入，必须用内存 flag 兜底 let isRunning = false @@ -30,21 +47,41 @@ function isParentAlive(): boolean { } } -async function processTask(task: QueueTask) { +// Session messages 缓存：同一 session 的多个 task 短时间内不需要重复读取 JSONL +const sessionMessagesCache = new Map[]; ts: number }>() +const SESSION_CACHE_TTL_MS = 60_000 + +async function getCachedSessionMessages(sessionDir: string, sessionID: string, messageID?: string) { + const cacheKey = `${sessionDir}:${sessionID}` + const cached = sessionMessagesCache.get(cacheKey) + if (cached && Date.now() - cached.ts < SESSION_CACHE_TTL_MS) { + log.debug('session messages cache hit', { sessionID, messageID }) + return cached.messages + } + const start = Date.now() + const messages = await loadSessionMessages(sessionDir, sessionID, messageID) + const elapsed = Date.now() - start + if (elapsed > 100) { + log.info('loadSessionMessages slow', { sessionID, elapsedMs: elapsed, messageCount: messages.length }) + } + sessionMessagesCache.set(cacheKey, { messages, ts: Date.now() }) + return messages +} + +async function processTask(task: QueueTask, state: Awaited>) { log.info('processing task', { sessionID: task.sessionID, messageID: task.messageID }) const sessionDir = getSessionDirectory(task.directory, task.sessionID) - const messages = await loadSessionMessages(sessionDir, task.sessionID, task.messageID) + const messages = await getCachedSessionMessages(sessionDir, task.sessionID, task.messageID) if (messages.length === 0) { log.warn('no messages found', { sessionDir, sessionID: task.sessionID }) } const authData = await authWithFallback() - const state = await readState() // 预加载 git 信息，commits 和 repo 字段共享，避免每个 task 重复 spawn git 进程 - const repoInfo = await getRepoInfo(task.directory) + const repoInfo = await getCachedRepoInfo(task.directory) try { // conversation @@ -66,9 +103,12 @@ async function processTask(task: QueueTask) { await uploadCommits({ directory: task.directory }, authData, state, { repoInfo }) log.info('task completed', { sessionID: task.sessionID, conversationUploaded }) - } finally { - // 无论成功或失败，都写入 state（commits 已逐条更新） - await writeState(state) + } catch (err) { + log.error('task failed', { + error: err instanceof Error ? err.message : String(err), + sessionID: task.sessionID, + }) + throw err } } @@ -82,13 +122,13 @@ async function runBatch() { try { // 第二道防线：跨进程文件锁 - if (!acquireLock()) { + if (!(await acquireLock())) { log.debug('another worker process holds the lock, skip') return } try { - const tasks = readQueue() + const tasks = await readQueue() if (tasks.length === 0) { log.debug('queue empty') return @@ -97,7 +137,7 @@ async function runBatch() { // 第三道防线：读完立刻清空队列 // - 处理期间新进来的任务会在下一轮处理 // - 即使有意外的并发 runBatch 拿到锁，也只会看到空队列直接返回 - clearQueue() + await clearQueue() log.info(`processing ${tasks.length} tasks`) @@ -114,9 +154,12 @@ async function runBatch() { const uniqueTasks = Array.from(deduped.values()).sort((a, b) => a.enqueuedAt - b.enqueuedAt) log.info(`deduped to ${uniqueTasks.length} unique tasks`) + // 一次性读取 state，所有 task 共享，减少文件锁竞争和重复 JSON 解析 + const state = await readState() + for (const task of uniqueTasks) { try { - await processTask(task) + await processTask(task, state) } catch (err) { log.error('task failed', { error: err instanceof Error ? err.message : String(err), @@ -125,9 +168,11 @@ async function runBatch() { } } + // 所有 task 处理完后一次性写入 state + await writeState(state) log.info('batch completed') } finally { - releaseLock() + await releaseLock() } } finally { isRunning = false diff --git a/src/services/rawDump/index.ts b/src/services/rawDump/index.ts index 85b55f571..bf559f711 100644 --- a/src/services/rawDump/index.ts +++ b/src/services/rawDump/index.ts @@ -7,9 +7,16 @@ import { isLocalDumpMode } from './localStorage.js' import { enqueue } from './queue.js' import { spawnBatchWorker } from './spawn.js' import { startBatchWorker } from './batchWorker.js' +import { createLogger } from './logger.js' + +const log = createLogger('raw-dump') let batchWorkerSpawned = false +// 调用频率限制：同一 session + messageID 5s 内不重复 enqueue +const lastEnqueueMap = new Map() +const ENQUEUE_DEBOUNCE_MS = 5_000 + function isEnabled(): boolean { // 本地调试模式自动启用 if (isLocalDumpMode()) return true @@ -24,23 +31,37 @@ function ensureBatchWorker() { batchWorkerSpawned = true const spawned = spawnBatchWorker() if (!spawned) { - // Fallback: compiled binary mode where external worker file is not available + log.warn('batch worker spawn failed, falling back to inline worker') startBatchWorker() } } +function shouldEnqueue(sessionID: string, messageID: string): boolean { + const key = `${sessionID}:${messageID}` + const now = Date.now() + const last = lastEnqueueMap.get(key) + if (last && now - last < ENQUEUE_DEBOUNCE_MS) { + log.debug('reportTurn debounced', { sessionID, messageID, lastMs: now - last }) + return false + } + lastEnqueueMap.set(key, now) + return true +} + /** * 上报一轮对话 * 只写入队列，由 batch worker 顺序消费 */ export function reportTurn(sessionID: string, messageID: string, directory: string): void { if (!isEnabled()) return + if (!shouldEnqueue(sessionID, messageID)) return enqueue({ sessionID, messageID, directory }) ensureBatchWorker() } export function reportSession(sessionID: string, directory: string): void { if (!isEnabled()) return + if (!shouldEnqueue(sessionID, '__summary__')) return enqueue({ sessionID, messageID: '__summary__', directory }) ensureBatchWorker() } diff --git a/src/services/rawDump/queue.ts b/src/services/rawDump/queue.ts index 5334cb948..31b21ef46 100644 --- a/src/services/rawDump/queue.ts +++ b/src/services/rawDump/queue.ts @@ -3,7 +3,8 @@ * 主进程只写队列，独立 batch worker 顺序消费 */ -import { appendFileSync, readFileSync, writeFileSync } from 'node:fs' +import { appendFileSync } from 'node:fs' +import { promises as fs } from 'node:fs' import os from 'node:os' import path from 'node:path' @@ -19,16 +20,13 @@ export interface QueueTask { export function enqueue(task: Omit): void { const item: QueueTask = { ...task, enqueuedAt: Date.now() } - try { - appendFileSync(QUEUE_FILE, JSON.stringify(item) + '\n', 'utf-8') - } catch { - // ignore - } + // 使用 fire-and-forget 异步写入，避免阻塞主进程 event loop + fs.appendFile(QUEUE_FILE, JSON.stringify(item) + '\n', 'utf-8').catch(() => {}) } -export function readQueue(): QueueTask[] { +export async function readQueue(): Promise { try { - const text = readFileSync(QUEUE_FILE, 'utf-8') + const text = await fs.readFile(QUEUE_FILE, 'utf-8') return text .split('\n') .filter(Boolean) @@ -45,19 +43,18 @@ export function readQueue(): QueueTask[] { } } -export function clearQueue(): void { +export async function clearQueue(): Promise { try { - writeFileSync(QUEUE_FILE, '', 'utf-8') + await fs.writeFile(QUEUE_FILE, '', 'utf-8') } catch { // ignore } } -export function acquireLock(): boolean { +export async function acquireLock(): Promise { try { - // 简单文件锁：如果 lock 文件存在且 60 秒内，认为已有 worker try { - const stat = readFileSync(LOCK_FILE, 'utf-8') + const stat = await fs.readFile(LOCK_FILE, 'utf-8') const pid = parseInt(stat, 10) if (!isNaN(pid) && pid !== process.pid) { // 检查进程是否还在运行 @@ -71,16 +68,16 @@ export function acquireLock(): boolean { } catch { // lock 文件不存在 } - writeFileSync(LOCK_FILE, String(process.pid), 'utf-8') + await fs.writeFile(LOCK_FILE, String(process.pid), 'utf-8') return true } catch { return false } } -export function releaseLock(): void { +export async function releaseLock(): Promise { try { - writeFileSync(LOCK_FILE, '', 'utf-8') + await fs.writeFile(LOCK_FILE, '', 'utf-8') } catch { // ignore } diff --git a/src/services/rawDump/spawn.ts b/src/services/rawDump/spawn.ts index 095180e53..8b2f88454 100644 --- a/src/services/rawDump/spawn.ts +++ b/src/services/rawDump/spawn.ts @@ -44,6 +44,8 @@ function resolveRuntime(): { entry: string; isBun: boolean } | null { if (typeof Bun !== 'undefined' && Bun.which) { const bun = Bun.which('bun') if (bun) return { entry: bun, isBun: true } + const node = Bun.which('node') + if (node) return { entry: node, isBun: false } } return null diff --git a/src/services/rawDump/worker.ts b/src/services/rawDump/worker.ts index aec15f59e..2bf691780 100644 --- a/src/services/rawDump/worker.ts +++ b/src/services/rawDump/worker.ts @@ -248,7 +248,16 @@ export async function loadSessionMessages(sessionDir: string, sessionId: string, const jsonlFiles = entries.filter((f) => f.endsWith('.jsonl')) log.debug('found jsonl files', { sessionDir, count: jsonlFiles.length, files: jsonlFiles.slice(0, 5) }) - for (const file of jsonlFiles) { + // 优先读取文件名包含 sessionId 的文件，减少无意义解析 + const prioritized = jsonlFiles.sort((a, b) => { + const aHas = a.includes(sessionId) + const bHas = b.includes(sessionId) + if (aHas && !bHas) return -1 + if (!aHas && bHas) return 1 + return 0 + }) + + for (const file of prioritized) { const filePath = path.join(sessionDir, file) try { const text = await fs.readFile(filePath, 'utf-8')