From 3cfe5c8ad3a7ad2c80aabb2b31ae825e0dd874f2 Mon Sep 17 00:00:00 2001 From: Konstantin Wohlwend Date: Fri, 3 Jul 2026 18:11:36 -0700 Subject: [PATCH] Batch Bulldozer transaction operations every 10ms --- .../low-level/implementations/lmdb.test.ts | 40 ++++ .../low-level/implementations/lmdb.ts | 178 +++++++++++------- .../src/databases/piledriver/index.ts | 65 +------ 3 files changed, 156 insertions(+), 127 deletions(-) diff --git a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.test.ts b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.test.ts index 0d48257cb..5472d6a4c 100644 --- a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.test.ts +++ b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.test.ts @@ -99,4 +99,44 @@ describe("LMDB low-level database", () => { await rm(path, { recursive: true, force: true }); } }); + + it("coalesces independent writes into one delayed transaction", async () => { + const path = await tempLmdbPath(); + try { + const db = declareLmdbLowLevelDatabase({ path, dbId: "coalesce" }); + const store = db.declareKvStore("store"); + const beforeVersion = db.getDebugInfo().currentVersion; + + const first = await store.setAll([{ key: buffer("a"), value: buffer("one") }]); + const second = await store.setAll([{ key: buffer("b"), value: buffer("two") }]); + + expect(db.getDebugInfo().currentVersion).toBe(beforeVersion); + await db.waitUntilAvailable(db.combineSeqs(first.seq, second.seq)); + + expect(db.getDebugInfo().currentVersion).toBe(beforeVersion + 1); + expect(text((await store.get(buffer("a"))).buffer)).toBe("one"); + expect(text((await store.get(buffer("b"))).buffer)).toBe("two"); + } finally { + await rm(path, { recursive: true, force: true }); + } + }); + + it("does not deadlock when one queued write requires another queued write", async () => { + const path = await tempLmdbPath(); + try { + const db = declareLmdbLowLevelDatabase({ path, dbId: "same-batch-dependency" }); + const store = db.declareKvStore("store"); + const beforeVersion = db.getDebugInfo().currentVersion; + + const first = await store.setAll([{ key: buffer("parent"), value: buffer("first") }]); + const second = await store.setAll([{ key: buffer("child"), value: buffer("second") }], { requiresSeq: db.combineSeqs(first.seq) }); + + await db.waitUntilAvailable(second.seq); + expect(db.getDebugInfo().currentVersion).toBe(beforeVersion + 1); + expect(text((await store.get(buffer("parent"))).buffer)).toBe("first"); + expect(text((await store.get(buffer("child"))).buffer)).toBe("second"); + } finally { + await rm(path, { recursive: true, force: true }); + } + }); }); diff --git a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts index 8c9b9b31e..8e249e9c6 100644 --- a/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts +++ b/apps/bulldozer-js/src/databases/low-level/implementations/lmdb.ts @@ -10,6 +10,13 @@ type BinaryDatabase = lmdb.Database; type VersionedBinaryDatabase = BinaryDatabase & { getEntry(key: Buffer): { value: Buffer, version?: number } | undefined, }; +type PendingCommitOperation = { + seqId: string, + requiresSeq: DatabaseSeq, + action: (version: number) => Promise, + resolve: () => void, + reject: (error: unknown) => void, +}; function arrayBuffersAreEqual(a: ArrayBuffer, b: ArrayBuffer): boolean { if (a.byteLength !== b.byteLength) return false; @@ -51,6 +58,20 @@ function validateValue(name: string, value: ArrayBuffer) { if (value.byteLength > 2_000_000_000) throw new Error(`KV store ${name} must be <= 2GB`); } +function createVoidDeferred() { + let resolveOperation: () => void = () => { + throw new Error("Deferred promise resolved before initialization"); + }; + let rejectOperation: (error: unknown) => void = (_error) => { + throw new Error("Deferred promise rejected before initialization"); + }; + const promise = new Promise((resolve, reject) => { + resolveOperation = () => resolve(); + rejectOperation = error => reject(error); + }); + return { promise, resolve: resolveOperation, reject: rejectOperation }; +} + type LmdbActivityStats = { puts: number, putBytes: number, @@ -119,6 +140,10 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri const seqToDurability = new Map>(); const combinedSeqToAvailability = new Map>(); const combinedSeqToDurability = new Map>(); + const combinedSeqDependencies = new Map(); + let pendingCommitOperations: PendingCommitOperation[] = []; + let pendingCommitFlushTimer: ReturnType | null = null; + let pendingCommitFlushPromise: Promise | null = null; let activityStats = emptyActivityStats(); let activityWindowStartedAt = performance.now(); const activityInterval = setInterval(() => { @@ -153,6 +178,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri seqToDurability: seqToDurability.size, combinedSeqToAvailability: combinedSeqToAvailability.size, combinedSeqToDurability: combinedSeqToDurability.size, + combinedSeqDependencies: combinedSeqDependencies.size, debugEntriesByStoreId: debugEntriesByStoreId.size, }, currentVersion, @@ -203,6 +229,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri activityStats.combinedSeqAvailabilityResolveTotalMs += performance.now() - insertedAt; activityStats.combinedSeqAvailabilityResolves++; combinedSeqToAvailability.delete(seqId); + combinedSeqDependencies.delete(seqId); })); availability.catch(() => {}); combinedSeqToAvailability.set(seqId, availability); @@ -222,36 +249,68 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri rememberDurability(seqId, promise); return toSeq(seqId); }; - const commit = (requiresSeq: DatabaseSeq, action: (version: number) => Promise) => { - const version = nextVersion(); - const seqId = nextSeqId(); - const promise = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.commit", attributes: { "bulldozer.low_level.backend": "lmdb" } }, async () => { - const requiredSeqWaitStartedAt = performance.now(); - await waitUntilAvailable(requiresSeq); - activityStats.requiredSeqWaits++; - activityStats.requiredSeqWaitTotalMs += performance.now() - requiredSeqWaitStartedAt; - const transactionStartedAt = performance.now(); - let transactionCallbackFinishedAt: number | null = null; - return await root.transaction(() => { - activityStats.transactionQueueWaitTotalMs += performance.now() - transactionStartedAt; - activityStats.transactions++; - return (async () => { - const actionStartedAt = performance.now(); - await action(version); - activityStats.transactionActionTotalMs += performance.now() - actionStartedAt; - const metaPutStartedAt = performance.now(); - await meta.put("seq", version); - activityStats.metaPutTotalMs += performance.now() - metaPutStartedAt; - })().finally(() => { - transactionCallbackFinishedAt = performance.now(); + const commitBatch = async (operations: PendingCommitOperation[]) => { + if (operations.length === 0) return; + try { + const version = nextVersion(); + await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.commit", attributes: { "bulldozer.low_level.backend": "lmdb", "bulldozer.low_level.operation_count": operations.length } }, async () => { + const requiredSeqWaitStartedAt = performance.now(); + const batchSeqIds = new Set(operations.map(operation => operation.seqId)); + await Promise.all(operations.map(async operation => await waitUntilAvailableOutsideBatch(operation.requiresSeq, batchSeqIds))); + activityStats.requiredSeqWaits++; + activityStats.requiredSeqWaitTotalMs += performance.now() - requiredSeqWaitStartedAt; + const transactionStartedAt = performance.now(); + let transactionCallbackFinishedAt: number | null = null; + await root.transaction(() => { + activityStats.transactionQueueWaitTotalMs += performance.now() - transactionStartedAt; + activityStats.transactions++; + return (async () => { + const actionStartedAt = performance.now(); + for (const operation of operations) await operation.action(version); + activityStats.transactionActionTotalMs += performance.now() - actionStartedAt; + const metaPutStartedAt = performance.now(); + await meta.put("seq", version); + activityStats.metaPutTotalMs += performance.now() - metaPutStartedAt; + })().finally(() => { + transactionCallbackFinishedAt = performance.now(); + }); + }).finally(() => { + const transactionFinishedAt = performance.now(); + activityStats.transactionTotalMs += transactionFinishedAt - transactionStartedAt; + if (transactionCallbackFinishedAt !== null) activityStats.transactionCommitTailTotalMs += transactionFinishedAt - transactionCallbackFinishedAt; }); - }).finally(() => { - const transactionFinishedAt = performance.now(); - activityStats.transactionTotalMs += transactionFinishedAt - transactionStartedAt; - if (transactionCallbackFinishedAt !== null) activityStats.transactionCommitTailTotalMs += transactionFinishedAt - transactionCallbackFinishedAt; }); - }); - return trackCommit(seqId, promise); + for (const operation of operations) operation.resolve(); + } catch (error) { + for (const operation of operations) operation.reject(error); + throw error; + } + }; + const flushPendingCommits = async () => { + if (pendingCommitFlushTimer !== null) { + clearTimeout(pendingCommitFlushTimer); + pendingCommitFlushTimer = null; + } + const batch = pendingCommitOperations; + pendingCommitOperations = []; + await commitBatch(batch); + }; + const schedulePendingCommitFlush = () => { + if (pendingCommitFlushTimer !== null) return; + pendingCommitFlushTimer = setTimeout(() => { + pendingCommitFlushTimer = null; + pendingCommitFlushPromise = flushPendingCommits().finally(() => { + pendingCommitFlushPromise = null; + }); + pendingCommitFlushPromise.catch(() => {}); + }, 10); + }; + const commit = (requiresSeq: DatabaseSeq, action: (version: number) => Promise) => { + const seqId = nextSeqId(); + const deferred = createVoidDeferred(); + pendingCommitOperations.push({ seqId, requiresSeq, action, resolve: deferred.resolve, reject: deferred.reject }); + schedulePendingCommitFlush(); + return trackCommit(seqId, deferred.promise); }; const commitIfVersion = async (db: BinaryDatabase, key: Buffer, version: number, action: (version: number) => Promise) => { const nextVersionRef: { value: number | null } = { value: null }; @@ -269,12 +328,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri rememberDurability(seqId, Promise.resolve()); return toSeq(seqId); }; - const dumpKeyForVersion = (version: number, index: number) => { - const key = crypto.getRandomValues(new Uint8Array(48)); - new DataView(key.buffer).setBigUint64(0, BigInt(version)); - new DataView(key.buffer).setUint32(8, index); - return key.buffer; - }; + const dumpKey = () => crypto.getRandomValues(new Uint8Array(48)).buffer; const putWithVersion = async (db: BinaryDatabase, key: Buffer, value: Buffer, version: number) => { activityStats.puts++; activityStats.putBytes += value.byteLength; @@ -291,6 +345,20 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri const waitUntilDurable = async (seq: DatabaseSeq) => { await getDurabilityPromise(getSeqId(seq)); }; + const waitUntilAvailableOutsideBatch = async (seq: DatabaseSeq, batchSeqIds: Set): Promise => { + const seqId = getSeqId(seq); + if (seqId === initialSeqId || batchSeqIds.has(seqId)) return; + const dependencies = combinedSeqDependencies.get(seqId); + if (dependencies === undefined) { + await getAvailabilityPromise(seqId); + return; + } + await Promise.all(dependencies.map(async dependencySeqId => { + if (dependencySeqId === initialSeqId || batchSeqIds.has(dependencySeqId)) return; + const dependencySeq = toSeq(dependencySeqId); + await waitUntilAvailableOutsideBatch(dependencySeq, batchSeqIds); + })); + }; const waitUntilAllAvailable = async () => { await Promise.all(seqToAvailability.values()); }; @@ -348,36 +416,13 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri return await traceSpanHot({ description: "bulldozer-js.low-level.lmdb.insertAll", attributes: { ...attributes, "bulldozer.low_level.value_count": values.length } }, async () => { for (const value of values) validateValue("value", value); if (values.length === 0) return { keys: [], seq: insertOptions?.requiresSeq ?? initialSeq }; - const version = nextVersion(); - const seqId = nextSeqId(); - const keys = values.map((_, index) => dumpKeyForVersion(version, index)); - const promise = traceSpanHot({ description: "bulldozer-js.low-level.lmdb.insertAll.commit", attributes }, async () => { - const requiredSeqWaitStartedAt = performance.now(); - await waitUntilAvailable(insertOptions?.requiresSeq ?? initialSeq); - activityStats.requiredSeqWaits++; - activityStats.requiredSeqWaitTotalMs += performance.now() - requiredSeqWaitStartedAt; - const transactionStartedAt = performance.now(); - let transactionCallbackFinishedAt: number | null = null; - return await root.transaction(() => { - activityStats.transactionQueueWaitTotalMs += performance.now() - transactionStartedAt; - activityStats.transactions++; - return (async () => { - const actionStartedAt = performance.now(); - await Promise.all(values.map(async (value, index) => await putWithVersion(db, bufferFromArrayBuffer(keys[index]), bufferFromArrayBuffer(value), version))); - activityStats.transactionActionTotalMs += performance.now() - actionStartedAt; - const metaPutStartedAt = performance.now(); - await meta.put("seq", version); - activityStats.metaPutTotalMs += performance.now() - metaPutStartedAt; - })().finally(() => { - transactionCallbackFinishedAt = performance.now(); - }); - }).finally(() => { - const transactionFinishedAt = performance.now(); - activityStats.transactionTotalMs += transactionFinishedAt - transactionStartedAt; - if (transactionCallbackFinishedAt !== null) activityStats.transactionCommitTailTotalMs += transactionFinishedAt - transactionCallbackFinishedAt; - }); - }); - return { keys, seq: trackCommit(seqId, promise) }; + const keys = values.map(() => dumpKey()); + return { + keys, + seq: commit(insertOptions?.requiresSeq ?? initialSeq, async version => { + await Promise.all(values.map(async (value, index) => await putWithVersion(db, bufferFromArrayBuffer(keys[index]), bufferFromArrayBuffer(value), version))); + }), + }; }); }, async compareAndSet(key, compare, value, casOptions) { @@ -430,6 +475,10 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri seqToDurability, combinedSeqToAvailability, combinedSeqToDurability, + combinedSeqDependencies, + pendingCommitOperations, + pendingCommitFlushTimer, + pendingCommitFlushPromise, initialSeq, }; }, @@ -455,6 +504,7 @@ export function declareLmdbLowLevelDatabase(options: { path: string, dbId?: stri if (seqs.length === 0) return initialSeq; if (seqs.length === 1) return seqs[0]; const seqId = nextSeqId(); + combinedSeqDependencies.set(seqId, seqs.map(seq => getSeqId(seq))); rememberCombinedAvailability(seqId, Promise.all(seqs.map(seq => getAvailabilityPromise(getSeqId(seq))))); rememberCombinedDurability(seqId, Promise.all(seqs.map(seq => getDurabilityPromise(getSeqId(seq))))); return toSeq(seqId); diff --git a/apps/bulldozer-js/src/databases/piledriver/index.ts b/apps/bulldozer-js/src/databases/piledriver/index.ts index 4ff7deb1d..52e4670eb 100644 --- a/apps/bulldozer-js/src/databases/piledriver/index.ts +++ b/apps/bulldozer-js/src/databases/piledriver/index.ts @@ -1,5 +1,4 @@ import { decodeBase64, encodeBase64 } from "@hexclave/shared/dist/utils/bytes"; -import { runAsynchronously } from "@hexclave/shared/dist/utils/promises"; import { traceSpan, traceSpanHot } from "../../otel.js"; import { Database, DatabaseSeq } from "../index.js"; import { LowLevelDatabase, LowLevelDatabaseDebugSnapshot } from "../low-level/index.js"; @@ -75,13 +74,6 @@ export type PiledriverDatabaseOptions = { // fine: only true ancestors are in the path. Plain-object cycles are detected separately with a // push/pop set during the synchronous payload walk. type HeapSerializationPath = ReadonlySet; -type PendingHeapInsert = { - buffer: ArrayBuffer, - requiresSeq: DatabaseSeq, - serializationTimingStats: PiledriverSerializationTimingStats | undefined, - resolve: (value: { key: ArrayBuffer, seq: DatabaseSeq }) => void, - reject: (error: unknown) => void, -}; type PiledriverSerializationTimingStats = { primitiveNodes: number, arrayNodes: number, @@ -99,9 +91,6 @@ type PiledriverSerializationTimingStats = { heapObjectGetTotalMs: number, heapObjectSerializeTotalMs: number, heapObjectInsertAwaitTotalMs: number, - heapInsertFlushes: number, - heapInsertValues: number, - heapInsertAllTotalMs: number, branchStats: Map, heapObjectCacheMissesByShape: Map, heapObjectCacheMissInlineNodeCountsByShape: Map, @@ -156,9 +145,6 @@ function emptyPiledriverSerializationTimingStats(): PiledriverSerializationTimin heapObjectGetTotalMs: 0, heapObjectSerializeTotalMs: 0, heapObjectInsertAwaitTotalMs: 0, - heapInsertFlushes: 0, - heapInsertValues: 0, - heapInsertAllTotalMs: 0, branchStats: new Map(), heapObjectCacheMissesByShape: new Map(), heapObjectCacheMissInlineNodeCountsByShape: new Map(), @@ -264,8 +250,6 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: const heapObjectsByHeapKeyBase64 = new Map, seq: DatabaseSeq }>(); const heapObjectsByHeapKeyFinalizer = new FinalizationRegistry(([keyBase64, refIdentity]: [string, string]) => heapObjectsByHeapKeyBase64.get(keyBase64)?.refIdentity === refIdentity && heapObjectsByHeapKeyBase64.delete(keyBase64)); const heapKeysAndSeqByHeapObjects = new WeakMap>(); - let pendingHeapInserts: PendingHeapInsert[] = []; - let heapInsertFlushScheduled = false; const cacheHeapObjectByKey = (key: ArrayBuffer, heapObj: PiledriverHeapObject, seq: DatabaseSeq) => { const keyBase64 = encodeBase64(new Uint8Array(key)); @@ -273,48 +257,6 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: heapObjectsByHeapKeyBase64.set(keyBase64, { refIdentity, object: new WeakRef(heapObj), seq }); heapObjectsByHeapKeyFinalizer.register(heapObj, [keyBase64, refIdentity]); }; - const flushPendingHeapInserts = async () => { - const batch = pendingHeapInserts; - pendingHeapInserts = []; - heapInsertFlushScheduled = false; - if (batch.length === 0) return; - - try { - const entriesBySerializationTimingStats = new Map(); - for (const entry of batch) { - if (entry.serializationTimingStats !== undefined) { - entriesBySerializationTimingStats.set(entry.serializationTimingStats, (entriesBySerializationTimingStats.get(entry.serializationTimingStats) ?? 0) + 1); - } - } - const heapInsertAllStartedAt = performance.now(); - const inserted = await heapDump.insertAll(batch.map(entry => entry.buffer)); - const heapInsertAllMs = performance.now() - heapInsertAllStartedAt; - for (const [stats, entryCount] of entriesBySerializationTimingStats) { - stats.heapInsertFlushes++; - stats.heapInsertValues += entryCount; - stats.heapInsertAllTotalMs += heapInsertAllMs * entryCount / batch.length; - } - for (let i = 0; i < batch.length; i++) { - // Most heap payloads (e.g. leaf row data) have no heap children, so requiresSeq is the - // singleton initial seq — skip the combined-seq allocation entirely for those. - batch[i].resolve({ key: inserted.keys[i], seq: batch[i].requiresSeq === lowLevelDb.initialSeq ? inserted.seq : lowLevelDb.combineSeqs(batch[i].requiresSeq, inserted.seq) }); - } - } catch (error) { - for (const entry of batch) entry.reject(error); - } - }; - const insertHeapObjectBatched = async (buffer: ArrayBuffer, requiresSeq: DatabaseSeq, serializationTimingStats: PiledriverSerializationTimingStats | undefined): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { - return await new Promise((resolve, reject) => { - pendingHeapInserts.push({ buffer, requiresSeq, serializationTimingStats, resolve, reject }); - if (!heapInsertFlushScheduled) { - heapInsertFlushScheduled = true; - const timeout = setTimeout(() => { - runAsynchronously(async () => await flushPendingHeapInserts()); - }, 0); - timeout.unref(); - } - }); - }; // Deduplicates and drops initial seqs before delegating to the low-level combineSeqs. This is // important for performance: each low-level combined seq allocates promises/map entries, so we @@ -373,9 +315,9 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: const serialized = await serializePiledriverObject(heapObject, childHeapPath, serializationTimingStats, branchKey); if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectSerializeTotalMs += performance.now() - heapObjectSerializeStartedAt; const heapObjectInsertStartedAt = performance.now(); - const result = await insertHeapObjectBatched(serialized.buffer, serialized.seq, serializationTimingStats); + const inserted = await heapDump.insertAll([serialized.buffer], { requiresSeq: serialized.seq }); if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectInsertAwaitTotalMs += performance.now() - heapObjectInsertStartedAt; - return result; + return { key: inserted.keys[0], seq: inserted.seq }; }); })(); heapKeysAndSeqByHeapObjects.set(heapObj, promise); @@ -740,9 +682,6 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: heapObjectGetTotalMs: timingStats.heapObjectGetTotalMs, heapObjectSerializeTotalMs: timingStats.heapObjectSerializeTotalMs, heapObjectInsertAwaitTotalMs: timingStats.heapObjectInsertAwaitTotalMs, - heapInsertFlushes: timingStats.heapInsertFlushes, - heapInsertValues: timingStats.heapInsertValues, - heapInsertAllTotalMs: timingStats.heapInsertAllTotalMs, topHeapObjectCacheMissShapes, topSerializationBranches, });