From bbbd0bb3c66cfe6c1691746aea5e13ab1989109f Mon Sep 17 00:00:00 2001 From: digua Date: Fri, 31 Jul 2026 12:59:39 +0800 Subject: [PATCH] perf(import): establish end-to-end import baseline --- apps/desktop/preload/index.d.ts | 24 ++ .../src/import/auto-importer.test.ts | 23 ++ packages/node-runtime/src/import/index.ts | 2 + .../src/import/streaming-importer.test.ts | 19 + .../src/import/streaming-importer.ts | 382 ++++++++++++------ packages/node-runtime/src/index.ts | 2 + scripts/bench-streaming-import.mts | 327 +++++++++++++++ src/services/import/types.ts | 25 ++ src/stores/session.ts | 18 +- 9 files changed, 672 insertions(+), 150 deletions(-) create mode 100644 scripts/bench-streaming-import.mts diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index b76862d3..2a6e0995 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -38,6 +38,30 @@ interface ImportDiagnostics { invalidTimestamp: number noType: number } + /** End-to-end import phase timings and memory samples */ + performance: { + timings: { + detectionMs: number + preprocessingMs: number + databaseSetupMs: number + parserMs: number + metaWriteMs: number + memberWriteMs: number + messageWriteMs: number + nicknameHistoryMs: number + indexCreationMs: number + ftsMs: number + checkpointMs: number + sessionIndexMs: number + postImportHookMs: number + totalMs: number + } + messageBatchCount: number + messageTransactionCount: number + rssStartMb: number + rssPeakMb: number + rssDeltaMb: number + } } interface ChatImportResult { diff --git a/packages/node-runtime/src/import/auto-importer.test.ts b/packages/node-runtime/src/import/auto-importer.test.ts index 59a21b6b..2f675d79 100644 --- a/packages/node-runtime/src/import/auto-importer.test.ts +++ b/packages/node-runtime/src/import/auto-importer.test.ts @@ -58,6 +58,29 @@ function createDeps(options?: { duplicateCount: 0, messagesSkipped: 2, skipReasons: { noSenderId: 0, noAccountName: 0, invalidTimestamp: 2, noType: 0 }, + performance: { + timings: { + detectionMs: 0, + preprocessingMs: 0, + databaseSetupMs: 0, + parserMs: 0, + metaWriteMs: 0, + memberWriteMs: 0, + messageWriteMs: 0, + nicknameHistoryMs: 0, + indexCreationMs: 0, + ftsMs: 0, + checkpointMs: 0, + sessionIndexMs: 0, + postImportHookMs: 0, + totalMs: 0, + }, + messageBatchCount: 0, + messageTransactionCount: 0, + rssStartMb: 0, + rssPeakMb: 0, + rssDeltaMb: 0, + }, }, } }, diff --git a/packages/node-runtime/src/import/index.ts b/packages/node-runtime/src/import/index.ts index 84f61bd0..c8feb5b4 100644 --- a/packages/node-runtime/src/import/index.ts +++ b/packages/node-runtime/src/import/index.ts @@ -17,6 +17,8 @@ export { streamingImport, analyzeNewImport, streamParseFileInfo } from './stream export type { SkipReasons, ImportDiagnostics, + ImportStageTimings, + ImportPerformanceDiagnostics, StreamImportResult, ImportProgressCallback, ImportLogger, diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index ac04d9fe..b0222e00 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/packages/node-runtime/src/import/streaming-importer.test.ts @@ -14,6 +14,7 @@ import { streamingImport, streamParseFileInfo, type ImportLogger, + type StreamImportResult, type StreamImportDeps, } from './streaming-importer' @@ -75,6 +76,23 @@ function createCollectingLogger(messages: string[]): ImportLogger { } } +function assertPerformanceDiagnostics(result: StreamImportResult): void { + const diagnostics = result.diagnostics?.performance + assert.ok(diagnostics) + + for (const [stage, durationMs] of Object.entries(diagnostics.timings)) { + assert.equal(Number.isFinite(durationMs), true, `${stage} should be finite`) + assert.ok(durationMs >= 0, `${stage} should not be negative`) + } + + assert.ok(diagnostics.timings.totalMs >= diagnostics.timings.detectionMs) + assert.ok(diagnostics.timings.totalMs >= diagnostics.timings.messageWriteMs) + assert.ok(diagnostics.messageBatchCount > 0) + assert.ok(diagnostics.messageTransactionCount > 0) + assert.ok(diagnostics.rssPeakMb >= diagnostics.rssStartMb) + assert.ok(diagnostics.rssDeltaMb >= 0) +} + function writeChunkedShuakamiQqExport(root: string): string { const chunksDir = path.join(root, 'chunks') fs.mkdirSync(chunksDir, { recursive: true }) @@ -544,6 +562,7 @@ test('streamingImport applies incremental-equivalent deduplication on first impo assert.equal(result.diagnostics?.messagesReceived, 5) assert.equal(result.diagnostics?.messagesWritten, 3) assert.equal(result.diagnostics?.duplicateCount, 2) + assertPerformanceDiagnostics(result) const db = new Database(dbPath, { readonly: true, nativeBinding }) const row = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index 1ac15a74..e4e5e22f 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -25,6 +25,7 @@ import { type ParseProgress, } from '@openchatlab/parser' import * as fs from 'fs' +import { performance } from 'node:perf_hooks' import { buildFtsIndex } from '../fts' import { createMessageDedupState, registerMessageAndCheckDuplicate, type DedupMessage } from './message-deduplicator' @@ -45,6 +46,33 @@ export interface ImportDiagnostics { duplicateCount: number messagesSkipped: number skipReasons: SkipReasons + performance: ImportPerformanceDiagnostics +} + +export interface ImportStageTimings { + detectionMs: number + preprocessingMs: number + databaseSetupMs: number + parserMs: number + metaWriteMs: number + memberWriteMs: number + messageWriteMs: number + nicknameHistoryMs: number + indexCreationMs: number + ftsMs: number + checkpointMs: number + sessionIndexMs: number + postImportHookMs: number + totalMs: number +} + +export interface ImportPerformanceDiagnostics { + timings: ImportStageTimings + messageBatchCount: number + messageTransactionCount: number + rssStartMb: number + rssPeakMb: number + rssDeltaMb: number } export interface StreamImportResult { @@ -91,6 +119,15 @@ const SYSTEM_SENDER_ID = 'SYSTEM' export const SYSTEM_MEMBER_NAME = '系统消息' const RESERVED_SYSTEM_SENDER_FORMATS = new Set(['chatlab', 'chatlab-jsonl']) +interface ImportTimingContext { + totalStartedAt: number + detectionMs: number +} + +function elapsedMs(startedAt: number): number { + return performance.now() - startedAt +} + /** * Let the event loop process pending I/O before a long synchronous step. * Without this, progress events written to an SSE socket stay buffered @@ -158,13 +195,19 @@ export async function streamingImport( formatOptions?: Record, externalSessionId?: string ): Promise { + const totalStartedAt = performance.now() + const detectionStartedAt = performance.now() + if (formatOptions?.formatId) { const formatId = formatOptions.formatId as string const feature = getFormatFeatureById(formatId) if (!feature) { return { success: false, error: 'error.unknown_format_id' } } - return streamImportSingle(filePath, deps, feature, formatOptions, externalSessionId) + return streamImportSingle(filePath, deps, feature, formatOptions, externalSessionId, { + totalStartedAt, + detectionMs: elapsedMs(detectionStartedAt), + }) } const candidates = detectAllFormats(filePath) @@ -172,11 +215,16 @@ export async function streamingImport( return { success: false, error: 'error.unrecognized_format' } } - if (candidates.length > 1) { - return streamImportWithFallback(filePath, deps, candidates, formatOptions, externalSessionId) + const timingContext = { + totalStartedAt, + detectionMs: elapsedMs(detectionStartedAt), } - return streamImportSingle(filePath, deps, candidates[0], formatOptions, externalSessionId) + if (candidates.length > 1) { + return streamImportWithFallback(filePath, deps, candidates, formatOptions, externalSessionId, timingContext) + } + + return streamImportSingle(filePath, deps, candidates[0], formatOptions, externalSessionId, timingContext) } async function streamImportWithFallback( @@ -184,13 +232,14 @@ async function streamImportWithFallback( deps: StreamImportDeps, candidates: FormatFeature[], formatOptions?: Record, - externalSessionId?: string + externalSessionId?: string, + timingContext?: ImportTimingContext ): Promise { for (let i = 0; i < candidates.length; i++) { const candidate = candidates[i] deps.logger?.info(`[StreamImport] Trying format ${i + 1}/${candidates.length}: ${candidate.name} (${candidate.id})`) - const result = await streamImportSingle(filePath, deps, candidate, formatOptions, externalSessionId) + const result = await streamImportSingle(filePath, deps, candidate, formatOptions, externalSessionId, timingContext) if (result.success) { if (i > 0) { @@ -214,10 +263,33 @@ async function streamImportSingle( deps: StreamImportDeps, formatFeature: FormatFeature, formatOptions?: Record, - externalSessionId?: string + externalSessionId?: string, + timingContext?: ImportTimingContext ): Promise { const { onProgress, logger } = deps const genId = deps.generateSessionId ?? defaultGenerateSessionId + const totalStartedAt = timingContext?.totalStartedAt ?? performance.now() + const timings: ImportStageTimings = { + detectionMs: timingContext?.detectionMs ?? 0, + preprocessingMs: 0, + databaseSetupMs: 0, + parserMs: 0, + metaWriteMs: 0, + memberWriteMs: 0, + messageWriteMs: 0, + nicknameHistoryMs: 0, + indexCreationMs: 0, + ftsMs: 0, + checkpointMs: 0, + sessionIndexMs: 0, + postImportHookMs: 0, + totalMs: 0, + } + const rssStartBytes = process.memoryUsage().rss + let rssPeakBytes = rssStartBytes + const sampleRss = () => { + rssPeakBytes = Math.max(rssPeakBytes, process.memoryUsage().rss) + } logger?.reset() const sessionId = externalSessionId || genId() @@ -235,7 +307,6 @@ async function streamImportSingle( const needsLargeFilePreprocess = preprocessor && needsPreprocess(filePath) const nativeCanParseOriginal = needsLargeFilePreprocess && isNativeFormatAvailable(formatFeature.id) - if (nativeCanParseOriginal) { logger?.info( `[NativeParser] Kernel ${formatFeature.id} is available; skipping large-file preprocessing and parsing the original export` @@ -251,6 +322,7 @@ async function streamImportSingle( message: '', }) + const preprocessingStartedAt = performance.now() try { tempFilePath = await preprocessor.preprocess(filePath, (progress: ParseProgress) => { onProgress({ ...progress, message: '' }) @@ -261,9 +333,13 @@ async function streamImportSingle( const errorMsg = `Preprocessing failed: ${err instanceof Error ? err.message : String(err)}` logger?.error(errorMsg, err instanceof Error ? err : undefined) return { success: false, error: errorMsg } + } finally { + timings.preprocessingMs = elapsedMs(preprocessingStartedAt) + sampleRss() } } + const databaseSetupStartedAt = performance.now() const databaseSetup = (() => { let db: DatabaseAdapter | undefined try { @@ -304,6 +380,8 @@ async function streamImportSingle( return { ok: false as const, error, databaseOpened: db !== undefined } } })() + timings.databaseSetupMs = elapsedMs(databaseSetupStartedAt) + sampleRss() if (!databaseSetup.ok) { if (tempFilePath && preprocessor) preprocessor.cleanup(tempFilePath) @@ -341,11 +419,13 @@ async function streamImportSingle( let duplicateCount = 0 let lastCheckpointCount = 0 let inTransaction = false + let messageTransactionCount = 0 const beginTransaction = () => { if (!inTransaction) { db.exec('BEGIN TRANSACTION') inTransaction = true + messageTransactionCount++ } } @@ -400,6 +480,16 @@ async function streamImportSingle( const dedupState = createMessageDedupState() logger?.info('Starting streamParseFile...') + const parsePipelineStartedAt = performance.now() + let parserTimingFinished = false + const finishParserTiming = () => { + if (parserTimingFinished) return + const parsePipelineMs = elapsedMs(parsePipelineStartedAt) + const writeCallbackMs = timings.metaWriteMs + timings.memberWriteMs + timings.messageWriteMs + timings.parserMs = Math.max(0, parsePipelineMs - writeCallbackMs) + parserTimingFinished = true + sampleRss() + } try { beginTransaction() @@ -424,148 +514,134 @@ async function streamImportSingle( }, onMeta: (meta: ParsedMeta) => { - callbackStats.onMetaCalls++ - importedPlatform = meta.platform || importedPlatform - if (!metaInserted) { - logger?.info(`Writing meta: name=${meta.name}, type=${meta.type}, platform=${meta.platform}`) - insertMeta.run( - meta.name, - meta.platform, - meta.type, - Math.floor(Date.now() / 1000), - meta.groupId || null, - meta.groupAvatar || null, - meta.ownerId || null - ) - metaInserted = true + const startedAt = performance.now() + try { + callbackStats.onMetaCalls++ + importedPlatform = meta.platform || importedPlatform + if (!metaInserted) { + logger?.info(`Writing meta: name=${meta.name}, type=${meta.type}, platform=${meta.platform}`) + insertMeta.run( + meta.name, + meta.platform, + meta.type, + Math.floor(Date.now() / 1000), + meta.groupId || null, + meta.groupAvatar || null, + meta.ownerId || null + ) + metaInserted = true + } + } finally { + timings.metaWriteMs += elapsedMs(startedAt) + sampleRss() } }, onMembers: (members: ParsedMember[]) => { - callbackStats.onMembersCalls++ - callbackStats.totalMembersReceived += members.length - logger?.info(`Received member batch: ${members.length} members`) - for (const member of members) { - const accountName = normalizeSystemMemberName(formatFeature.id, member.platformId, member.accountName) - const groupNickname = normalizeSystemMemberName(formatFeature.id, member.platformId, member.groupNickname) - insertMember.run( - member.platformId, - accountName || null, - groupNickname || null, - member.aliases ? JSON.stringify(member.aliases) : '[]', - member.avatar || null, - member.roles ? JSON.stringify(member.roles) : '[]' - ) - const row = getMemberId.get(member.platformId) as { id: number } | undefined - if (row) memberIdMap.set(member.platformId, row.id) + const startedAt = performance.now() + try { + callbackStats.onMembersCalls++ + callbackStats.totalMembersReceived += members.length + logger?.info(`Received member batch: ${members.length} members`) + for (const member of members) { + const accountName = normalizeSystemMemberName(formatFeature.id, member.platformId, member.accountName) + const groupNickname = normalizeSystemMemberName(formatFeature.id, member.platformId, member.groupNickname) + insertMember.run( + member.platformId, + accountName || null, + groupNickname || null, + member.aliases ? JSON.stringify(member.aliases) : '[]', + member.avatar || null, + member.roles ? JSON.stringify(member.roles) : '[]' + ) + const row = getMemberId.get(member.platformId) as { id: number } | undefined + if (row) memberIdMap.set(member.platformId, row.id) + } + } finally { + timings.memberWriteMs += elapsedMs(startedAt) + sampleRss() } }, onMessageBatch: (messages: ParsedMessage[]) => { - callbackStats.onMessageBatchCalls++ - callbackStats.totalMessagesReceived += messages.length - if (callbackStats.onMessageBatchCalls <= 3 || callbackStats.onMessageBatchCalls % 10 === 0) { - logger?.info(`Received message batch #${callbackStats.onMessageBatchCalls}: ${messages.length} messages`) - } - - let memberLookupTime = 0 - let memberInsertTime = 0 - let messageInsertTime = 0 - let nicknameTrackTime = 0 - let memberLookupCount = 0 - let memberInsertCount = 0 - let nicknameChangeCount = 0 - - for (const msg of messages) { - const senderAccountName = normalizeSystemMemberName( - formatFeature.id, - msg.senderPlatformId, - msg.senderAccountName - ) - const senderGroupNickname = normalizeSystemMemberName( - formatFeature.id, - msg.senderPlatformId, - msg.senderGroupNickname - ) - const prepared = prepareMessageForCreate(msg, senderAccountName) - if ('skipCounter' in prepared) { - callbackStats[prepared.skipCounter]++ - continue + const startedAt = performance.now() + try { + callbackStats.onMessageBatchCalls++ + callbackStats.totalMessagesReceived += messages.length + if (callbackStats.onMessageBatchCalls <= 3 || callbackStats.onMessageBatchCalls % 10 === 0) { + logger?.info(`Received message batch #${callbackStats.onMessageBatchCalls}: ${messages.length} messages`) } - const dedupMessage = prepared.message - - if (registerMessageAndCheckDuplicate(dedupMessage, dedupState)) { - duplicateCount++ - continue - } - - let t0 = Date.now() - if (!memberIdMap.has(msg.senderPlatformId)) { - insertMember.run( + for (const msg of messages) { + const senderAccountName = normalizeSystemMemberName( + formatFeature.id, msg.senderPlatformId, + msg.senderAccountName + ) + const senderGroupNickname = normalizeSystemMemberName( + formatFeature.id, + msg.senderPlatformId, + msg.senderGroupNickname + ) + const prepared = prepareMessageForCreate(msg, senderAccountName) + if ('skipCounter' in prepared) { + callbackStats[prepared.skipCounter]++ + continue + } + + const dedupMessage = prepared.message + + if (registerMessageAndCheckDuplicate(dedupMessage, dedupState)) { + duplicateCount++ + continue + } + + if (!memberIdMap.has(msg.senderPlatformId)) { + insertMember.run( + msg.senderPlatformId, + senderAccountName || null, + senderGroupNickname || null, + '[]', + null, + '[]' + ) + const row = getMemberId.get(msg.senderPlatformId) as { id: number } | undefined + if (row) memberIdMap.set(msg.senderPlatformId, row.id) + } + + const senderId = memberIdMap.get(msg.senderPlatformId) + if (senderId === undefined) continue + + insertMessage.run( + senderId, senderAccountName || null, senderGroupNickname || null, - '[]', - null, - '[]' + dedupMessage.timestamp, + dedupMessage.type, + dedupMessage.content, + msg.replyToMessageId || null, + msg.platformMessageId || null ) - const row = getMemberId.get(msg.senderPlatformId) as { id: number } | undefined - if (row) memberIdMap.set(msg.senderPlatformId, row.id) - memberInsertCount++ - memberInsertTime += Date.now() - t0 - } else { - memberLookupCount++ - memberLookupTime += Date.now() - t0 - } + messageCountInBatch++ + totalMessageCount++ - const senderId = memberIdMap.get(msg.senderPlatformId) - if (senderId === undefined) continue + trackNickname(accountNameTracker, msg.senderPlatformId, senderAccountName, msg.timestamp) + trackNickname(groupNicknameTracker, msg.senderPlatformId, senderGroupNickname, msg.timestamp) - t0 = Date.now() - insertMessage.run( - senderId, - senderAccountName || null, - senderGroupNickname || null, - dedupMessage.timestamp, - dedupMessage.type, - dedupMessage.content, - msg.replyToMessageId || null, - msg.platformMessageId || null - ) - messageInsertTime += Date.now() - t0 - messageCountInBatch++ - totalMessageCount++ - - t0 = Date.now() - trackNickname(accountNameTracker, msg.senderPlatformId, senderAccountName, msg.timestamp) - trackNickname(groupNicknameTracker, msg.senderPlatformId, senderGroupNickname, msg.timestamp) - nicknameTrackTime += Date.now() - t0 - // nicknameChangeCount is approximate but sufficient for logging - nicknameChangeCount += accountNameTracker.get(msg.senderPlatformId)?.history.length === 1 ? 0 : 0 - - if (messageCountInBatch >= BATCH_COMMIT_SIZE) { - const detail = - `[Detail] Member lookup: ${memberLookupTime}ms (${memberLookupCount} times) | ` + - `Member insert: ${memberInsertTime}ms (${memberInsertCount} times) | ` + - `Message insert: ${messageInsertTime}ms | ` + - `Nickname tracking: ${nicknameTrackTime}ms (${nicknameChangeCount} changes)` - logger?.perfDetail(detail) - commitAndBeginNew() - messageCountInBatch = 0 - memberLookupTime = 0 - memberInsertTime = 0 - messageInsertTime = 0 - nicknameTrackTime = 0 - memberLookupCount = 0 - memberInsertCount = 0 - nicknameChangeCount = 0 + if (messageCountInBatch >= BATCH_COMMIT_SIZE) { + commitAndBeginNew() + messageCountInBatch = 0 + } } + } finally { + timings.messageWriteMs += elapsedMs(startedAt) + sampleRss() } }, }, formatFeature.id ) + finishParserTiming() if (inTransaction) { db.exec('COMMIT') @@ -584,6 +660,7 @@ async function streamImportSingle( await yieldToEventLoop() logger?.perf('Writing nickname history', totalMessageCount) + const nicknameHistoryStartedAt = performance.now() db.exec('BEGIN TRANSACTION') let historyCount = 0 @@ -598,6 +675,8 @@ async function streamImportSingle( historyCount = countHistory(accountNameTracker) + countHistory(groupNicknameTracker) db.exec('COMMIT') + timings.nicknameHistoryMs = elapsedMs(nicknameHistoryStartedAt) + sampleRss() logger?.perf(`Nickname history written (${historyCount} entries)`, totalMessageCount) // Create indexes (deferred for performance) @@ -611,15 +690,22 @@ async function streamImportSingle( }) await yieldToEventLoop() logger?.perf('Creating indexes', totalMessageCount) + const indexCreationStartedAt = performance.now() db.exec(CHAT_DB_INDEXES) + timings.indexCreationMs = elapsedMs(indexCreationStartedAt) + sampleRss() logger?.perf('Indexes created', totalMessageCount) // Build FTS index + const ftsStartedAt = performance.now() try { buildFtsIndex(db) logger?.perf('FTS index built', totalMessageCount) } catch (ftsError) { logger?.error('FTS index build failed (non-fatal)', ftsError instanceof Error ? ftsError : undefined) + } finally { + timings.ftsMs = elapsedMs(ftsStartedAt) + sampleRss() } // Final WAL checkpoint + session index + post-import hook @@ -632,25 +718,43 @@ async function streamImportSingle( message: '', }) await yieldToEventLoop() + const checkpointStartedAt = performance.now() doCheckpoint() + timings.checkpointMs = elapsedMs(checkpointStartedAt) + sampleRss() logger?.perf('WAL checkpoint done', totalMessageCount) // Build session index (segment / message_context tables) + const sessionIndexStartedAt = performance.now() try { generateSessionIndex(db) logger?.perf('Session index built', totalMessageCount) } catch { /* non-fatal */ + } finally { + timings.sessionIndexMs = elapsedMs(sessionIndexStartedAt) + sampleRss() } // Post-import hook (e.g. overview cache) - try { - await deps.postImportHook?.(db, sessionId) - if (deps.postImportHook) logger?.perf('Post-import hook done', totalMessageCount) - } catch { - /* non-fatal */ + if (deps.postImportHook) { + const postImportHookStartedAt = performance.now() + try { + await deps.postImportHook(db, sessionId) + logger?.perf('Post-import hook done', totalMessageCount) + } catch { + /* non-fatal */ + } finally { + timings.postImportHookMs = elapsedMs(postImportHookStartedAt) + sampleRss() + } } + logger?.perfDetail( + `[Stages] parser=${timings.parserMs.toFixed(1)}ms | message-write=${timings.messageWriteMs.toFixed(1)}ms | ` + + `indexes=${timings.indexCreationMs.toFixed(1)}ms | fts=${timings.ftsMs.toFixed(1)}ms | ` + + `session-index=${timings.sessionIndexMs.toFixed(1)}ms | hook=${timings.postImportHookMs.toFixed(1)}ms` + ) logger?.perf('Import completed', totalMessageCount) // Diagnostic logging @@ -691,6 +795,7 @@ async function streamImportSingle( importError = 'error.no_messages' } } catch (error) { + finishParserTiming() logger?.error('Import failed', error instanceof Error ? error : undefined) if (inTransaction) { try { @@ -711,8 +816,11 @@ async function streamImportSingle( if (shouldDeleteDb) { deps.deleteDatabase(sessionId) } + sampleRss() } + timings.totalMs = elapsedMs(totalStartedAt) + const bytesPerMegabyte = 1024 * 1024 const diagnostics: ImportDiagnostics = { logFile: logger?.getCurrentLogFile() ?? null, detectedFormat: formatFeature ? `${formatFeature.name} (${formatFeature.id})` : null, @@ -730,6 +838,14 @@ async function streamImportSingle( invalidTimestamp: callbackStats.skippedInvalidTimestamp, noType: callbackStats.skippedNoType, }, + performance: { + timings, + messageBatchCount: callbackStats.onMessageBatchCalls, + messageTransactionCount, + rssStartMb: rssStartBytes / bytesPerMegabyte, + rssPeakMb: rssPeakBytes / bytesPerMegabyte, + rssDeltaMb: (rssPeakBytes - rssStartBytes) / bytesPerMegabyte, + }, } if (importError) { diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 3e0988a6..98211ce4 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -44,6 +44,8 @@ export type { WriteParseResultStats, SkipReasons, ImportDiagnostics, + ImportStageTimings, + ImportPerformanceDiagnostics, StreamImportResult, ImportProgressCallback, ImportLogger, diff --git a/scripts/bench-streaming-import.mts b/scripts/bench-streaming-import.mts new file mode 100644 index 00000000..f1e2b4b7 --- /dev/null +++ b/scripts/bench-streaming-import.mts @@ -0,0 +1,327 @@ +/** + * End-to-end streaming import benchmark. + * + * Each measured run executes in a fresh process and writes fresh databases. + * Fixture generation and result verification are excluded from the timed range. + * + * Usage: + * pnpm exec tsx scripts/bench-streaming-import.mts single 100000 [runs=3] + * pnpm exec tsx scripts/bench-streaming-import.mts batch 100 10000 [runs=3] + */ + +import { createHash } from 'node:crypto' +import { once } from 'node:events' +import { copyFileSync, createWriteStream, linkSync, mkdirSync, rmSync, statSync, unlinkSync } from 'node:fs' +import path from 'node:path' +import { fileURLToPath } from 'node:url' +import { spawnSync } from 'node:child_process' +import Database from 'better-sqlite3' +import { CHAT_DB_TABLES } from '@openchatlab/core' +import { BetterSqliteAdapter } from '../packages/node-runtime/src/better-sqlite3-adapter' +import { computeAndSetOverviewCache } from '../packages/node-runtime/src/cache/session-cache' +import { + streamingImport, + type ImportStageTimings, + type StreamImportDeps, +} from '../packages/node-runtime/src/import/streaming-importer' +import { createChatLabTempDir } from './chatlab-temp.mjs' + +const scriptPath = fileURLToPath(import.meta.url) +const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') +const memberCount = 200 +const sampleTexts = [ + '今天天气不错，我们出去玩吧！', + '哈哈哈哈哈哈这也太好笑了', + '[图片]', + '好的，收到，明天见。', + 'This is a mixed language message with some English words 和中文混排。', + '周末有人一起打球吗？地点老地方，时间下午三点，人齐就开打。', +] + +interface BenchmarkResult { + scenario: 'single' | 'batch' + fileCount: number + inputMessages: number + inputBytes: number + durationMs: number + peakRssMb: number + rssDeltaMb: number + databaseBytes: number + messagesWritten: number + membersWritten: number + outputSignature: string + timings: ImportStageTimings +} + +const zeroTimings = (): ImportStageTimings => ({ + detectionMs: 0, + preprocessingMs: 0, + databaseSetupMs: 0, + parserMs: 0, + metaWriteMs: 0, + memberWriteMs: 0, + messageWriteMs: 0, + nicknameHistoryMs: 0, + indexCreationMs: 0, + ftsMs: 0, + checkpointMs: 0, + sessionIndexMs: 0, + postImportHookMs: 0, + totalMs: 0, +}) + +function addTimings(target: ImportStageTimings, source: ImportStageTimings): void { + for (const key of Object.keys(target) as Array) target[key] += source[key] +} + +async function generateWeflowFixture(filePath: string, count: number): Promise { + const stream = createWriteStream(filePath, { encoding: 'utf-8' }) + const write = async (chunk: string) => { + if (!stream.write(chunk)) await once(stream, 'drain') + } + + await write( + `{"weflow":{"version":"1.0.0","exportedAt":1704164645},` + + `"session":{"wxid":"bench@chatroom","nickname":"性能测试群","displayName":"性能测试群","type":"群聊"},` + + '"avatars":{},"messages":[' + ) + + const batch: string[] = [] + for (let index = 0; index < count; index++) { + const member = index % memberCount + batch.push( + JSON.stringify({ + localId: index + 1, + createTime: 1_704_164_645 + index * 3, + formattedTime: '2024-01-02 03:04:05', + type: '文本消息', + localType: 1, + content: `${sampleTexts[index % sampleTexts.length]} #${index}`, + isSend: member === 0 ? 1 : 0, + senderUsername: `wxid_member_${member}`, + senderDisplayName: `成员${member}号`, + senderAvatarKey: `wxid_member_${member}`, + source: '', + }) + ) + if (batch.length >= 5000) { + await write((index + 1 > batch.length ? ',' : '') + batch.join(',')) + batch.length = 0 + } + } + if (batch.length > 0) await write((count > batch.length ? ',' : '') + batch.join(',')) + await write(']}') + stream.end() + await once(stream, 'finish') +} + +function createImportDeps(dbPath: string, cacheDir: string): StreamImportDeps { + return { + openDatabase() { + const raw = new Database(dbPath, { nativeBinding }) + raw.pragma('journal_mode = WAL') + raw.pragma('synchronous = NORMAL') + raw.exec(CHAT_DB_TABLES) + return new BetterSqliteAdapter(raw) + }, + deleteDatabase() { + for (const suffix of ['', '-wal', '-shm']) { + try { + unlinkSync(dbPath + suffix) + } catch { + /* ignore cleanup failures in a disposable benchmark workspace */ + } + } + }, + onProgress() { + /* benchmark excludes UI transport */ + }, + postImportHook(db, sessionId) { + computeAndSetOverviewCache(db, sessionId, cacheDir) + }, + } +} + +function inspectDatabase(dbPath: string): { + messages: number + members: number + signature: string +} { + const db = new Database(dbPath, { readonly: true, nativeBinding }) + const row = db + .prepare( + `SELECT + (SELECT COUNT(*) FROM message) AS messages, + (SELECT COUNT(*) FROM member) AS members, + (SELECT COALESCE(SUM(ts), 0) FROM message) AS timestampSum, + (SELECT COALESCE(SUM(sender_id), 0) FROM message) AS senderSum, + (SELECT COALESCE(SUM(type), 0) FROM message) AS typeSum, + (SELECT COALESCE(SUM(LENGTH(content)), 0) FROM message) AS contentLengthSum` + ) + .get() as Record + db.close() + const signature = createHash('sha256').update(JSON.stringify(row)).digest('hex').slice(0, 16) + return { messages: row.messages, members: row.members, signature } +} + +async function runWorker(fileCount: number, messagesPerFile: number): Promise { + const root = createChatLabTempDir('bench', 'streaming-import-') + const fixturePath = path.join(root, 'fixture.json') + const cacheDir = path.join(root, 'cache') + mkdirSync(cacheDir, { recursive: true }) + + try { + await generateWeflowFixture(fixturePath, messagesPerFile) + const inputPaths = [fixturePath] + for (let index = 1; index < fileCount; index++) { + const linkedPath = path.join(root, `fixture-${index}.json`) + try { + linkSync(fixturePath, linkedPath) + } catch { + copyFileSync(fixturePath, linkedPath) + } + inputPaths.push(linkedPath) + } + + global.gc?.() + const rssStartMb = process.memoryUsage().rss / 1024 / 1024 + const timings = zeroTimings() + let peakRssMb = rssStartMb + const dbPaths: string[] = [] + const startedAt = performance.now() + + for (let index = 0; index < inputPaths.length; index++) { + const dbPath = path.join(root, `session-${index}.db`) + dbPaths.push(dbPath) + const result = await streamingImport( + inputPaths[index], + createImportDeps(dbPath, cacheDir), + { formatId: 'weflow' }, + `bench-${index}` + ) + if (!result.success || !result.diagnostics?.performance) { + throw new Error(`Import ${index} failed: ${result.error ?? 'missing performance diagnostics'}`) + } + addTimings(timings, result.diagnostics.performance.timings) + peakRssMb = Math.max(peakRssMb, result.diagnostics.performance.rssPeakMb) + } + const durationMs = performance.now() - startedAt + + let messagesWritten = 0 + let membersWritten = 0 + let databaseBytes = 0 + const signatures: string[] = [] + for (const dbPath of dbPaths) { + const inspected = inspectDatabase(dbPath) + messagesWritten += inspected.messages + membersWritten += inspected.members + signatures.push(inspected.signature) + databaseBytes += statSync(dbPath).size + } + + return { + scenario: fileCount === 1 ? 'single' : 'batch', + fileCount, + inputMessages: fileCount * messagesPerFile, + inputBytes: statSync(fixturePath).size * fileCount, + durationMs, + peakRssMb, + rssDeltaMb: Math.max(0, peakRssMb - rssStartMb), + databaseBytes, + messagesWritten, + membersWritten, + outputSignature: createHash('sha256').update(signatures.sort().join(':')).digest('hex').slice(0, 16), + timings, + } + } finally { + rmSync(root, { recursive: true, force: true }) + } +} + +function runIsolated(fileCount: number, messagesPerFile: number): BenchmarkResult { + const child = spawnSync( + process.execPath, + [...process.execArgv, scriptPath, '--worker', String(fileCount), String(messagesPerFile)], + { + cwd: process.cwd(), + encoding: 'utf-8', + maxBuffer: 10 * 1024 * 1024, + env: { ...process.env }, + } + ) + if (child.status !== 0) { + throw new Error(`Benchmark worker failed (${child.status}):\n${child.stdout}\n${child.stderr}`) + } + const resultLine = child.stdout.split('\n').find((line) => line.startsWith('BENCH_RESULT ')) + if (!resultLine) throw new Error(`Benchmark worker returned no result:\n${child.stdout}\n${child.stderr}`) + return JSON.parse(resultLine.slice('BENCH_RESULT '.length)) as BenchmarkResult +} + +function formatBytes(bytes: number): string { + return `${(bytes / 1024 / 1024).toFixed(1)} MB` +} + +function printResult(result: BenchmarkResult, label: string): void { + console.log( + `${label}: ${result.durationMs.toFixed(0)} ms | peak RSS ${result.peakRssMb.toFixed(0)} MB ` + + `(+${result.rssDeltaMb.toFixed(0)} MB) | DB ${formatBytes(result.databaseBytes)} | ` + + `signature ${result.outputSignature}` + ) +} + +async function main(): Promise { + const [mode, first, second, third] = process.argv.slice(2) + if (mode === '--worker') { + const result = await runWorker(Number(first), Number(second)) + console.log(`BENCH_RESULT ${JSON.stringify(result)}`) + return + } + + let fileCount: number + let messagesPerFile: number + let runs: number + if (mode === 'single') { + fileCount = 1 + messagesPerFile = Number(first ?? 100_000) + runs = Number(second ?? 3) + } else if (mode === 'batch') { + fileCount = Number(first ?? 100) + messagesPerFile = Number(second ?? 10_000) + runs = Number(third ?? 3) + } else { + throw new Error('Usage: single [runs=3] | batch [runs=3]') + } + if (![fileCount, messagesPerFile, runs].every((value) => Number.isInteger(value) && value > 0)) { + throw new Error('All benchmark counts must be positive integers') + } + + console.log( + `Benchmarking ${fileCount} file(s), ${messagesPerFile.toLocaleString()} messages/file, ${runs} isolated run(s)` + ) + const results: BenchmarkResult[] = [] + for (let index = 0; index < runs; index++) { + const result = runIsolated(fileCount, messagesPerFile) + results.push(result) + printResult(result, `run ${index + 1}`) + } + + const sorted = [...results].sort((left, right) => left.durationMs - right.durationMs) + const median = sorted[Math.floor(sorted.length / 2)] + if (results.some((result) => result.outputSignature !== median.outputSignature)) { + throw new Error('Output signatures differ between isolated runs') + } + if (median.messagesWritten !== fileCount * messagesPerFile) { + throw new Error(`Expected ${fileCount * messagesPerFile} messages, got ${median.messagesWritten}`) + } + + printResult(median, 'median') + const stageRows = Object.entries(median.timings) + .filter(([stage]) => stage !== 'totalMs') + .sort(([, left], [, right]) => right - left) + console.log('Median phase totals:') + for (const [stage, durationMs] of stageRows) { + console.log(` ${stage.padEnd(22)} ${durationMs.toFixed(1).padStart(10)} ms`) + } +} + +await main() diff --git a/src/services/import/types.ts b/src/services/import/types.ts index 194cd881..4f31f48a 100644 --- a/src/services/import/types.ts +++ b/src/services/import/types.ts @@ -65,6 +65,31 @@ export interface ImportDiagnosticsInfo { invalidTimestamp: number noType: number } + performance: ImportPerformanceDiagnosticsInfo +} + +export interface ImportPerformanceDiagnosticsInfo { + timings: { + detectionMs: number + preprocessingMs: number + databaseSetupMs: number + parserMs: number + metaWriteMs: number + memberWriteMs: number + messageWriteMs: number + nicknameHistoryMs: number + indexCreationMs: number + ftsMs: number + checkpointMs: number + sessionIndexMs: number + postImportHookMs: number + totalMs: number + } + messageBatchCount: number + messageTransactionCount: number + rssStartMb: number + rssPeakMb: number + rssDeltaMb: number } // ==================== 格式信息 ==================== diff --git a/src/stores/session.ts b/src/stores/session.ts index 68b856fd..87d226f9 100644 --- a/src/stores/session.ts +++ b/src/stores/session.ts @@ -2,7 +2,7 @@ import { defineStore } from 'pinia' import { ref, computed } from 'vue' import type { AnalysisSession, ImportProgress, ChatType } from '@/types/base' import { useDataService, useImportService, useSessionIndexService, usePlatformService } from '@/services' -import type { AutoImportCreateReason, AutoImportMatchMethod, AutoImportMode } from '@/services' +import type { AutoImportCreateReason, AutoImportMatchMethod, AutoImportMode, ImportDiagnosticsInfo } from '@/services' import { IS_ELECTRON } from '@/utils/platform' /** 侧边栏筛选类型 */ @@ -217,22 +217,6 @@ export const useSessionStore = defineStore( } } - /** 导入诊断信息类型 */ - interface ImportDiagnosticsInfo { - logFile: string | null - detectedFormat: string | null - messagesReceived: number - messagesWritten: number - duplicateCount: number - messagesSkipped: number - skipReasons: { - noSenderId: number - noAccountName: number - invalidTimestamp: number - noType: number - } - } - /** * 从指定路径执行导入（支持拖拽） */