diff --git a/apps/bulldozer-js/src/databases/bulldozer/index.ts b/apps/bulldozer-js/src/databases/bulldozer/index.ts index 022e75152..2deebc806 100644 --- a/apps/bulldozer-js/src/databases/bulldozer/index.ts +++ b/apps/bulldozer-js/src/databases/bulldozer/index.ts @@ -887,15 +887,59 @@ export function declareBulldozerDatabase(piledriverDatabase: PiledriverDatabase, options: { replicated: boolean }, ) => { return await traceSpan({ description: "bulldozer-js.bulldozer.withSnapshot", attributes: { "bulldozer.replicated": options.replicated } }, async () => { + const startedAt = performance.now(); + let writeLockWaitMs = 0; + let getSnapshotMs = 0; + let updateSnapshotMs = 0; + let toPiledriverObjectMs = 0; + let setRootMs = 0; + let waitUntilAvailableMs = 0; + let waitUntilReplicatedMs = 0; + let mutationDebugInfo: BulldozerSnapshotMutationDebugInfo | undefined; + const writeLockWaitStartedAt = performance.now(); const result = await withWriteLock(async () => { + writeLockWaitMs = performance.now() - writeLockWaitStartedAt; + const getSnapshotStartedAt = performance.now(); const { snapshot } = await getSnapshot(); + getSnapshotMs = performance.now() - getSnapshotStartedAt; + const updateSnapshotStartedAt = performance.now(); const updateResult = await updateSnapshot(snapshot); + updateSnapshotMs = performance.now() - updateSnapshotStartedAt; + mutationDebugInfo = updateResult instanceof BulldozerDatabaseSnapshot ? undefined : updateResult.debugInfo; const newSnapshot = updateResult instanceof BulldozerDatabaseSnapshot ? updateResult : updateResult.newSnapshot; - const { seq } = await setRoot({ snapshot: newSnapshot.toPiledriverObject() }); + const toPiledriverObjectStartedAt = performance.now(); + const newSnapshotPiledriverObject = newSnapshot.toPiledriverObject(); + toPiledriverObjectMs = performance.now() - toPiledriverObjectStartedAt; + const setRootStartedAt = performance.now(); + const { seq } = await setRoot({ snapshot: newSnapshotPiledriverObject }); + setRootMs = performance.now() - setRootStartedAt; + const waitUntilAvailableStartedAt = performance.now(); await piledriverDatabase.waitUntilAvailable(seq); + waitUntilAvailableMs = performance.now() - waitUntilAvailableStartedAt; return { snapshot: newSnapshot, seq }; }); - if (options.replicated) await piledriverDatabase.waitUntilReplicated(result.seq); + if (options.replicated) { + const waitUntilReplicatedStartedAt = performance.now(); + await piledriverDatabase.waitUntilReplicated(result.seq); + waitUntilReplicatedMs = performance.now() - waitUntilReplicatedStartedAt; + } + console.debug("bulldozer-js withSnapshot timing", inspect({ + replicated: options.replicated, + elapsedMs: performance.now() - startedAt, + writeLockWaitMs, + getSnapshotMs, + updateSnapshotMs, + toPiledriverObjectMs, + setRootMs, + waitUntilAvailableMs, + waitUntilReplicatedMs, + mutation: mutationDebugInfo === undefined ? undefined : { + operation: mutationDebugInfo.operation, + sourceTableId: mutationDebugInfo.sourceTableId, + rowsSetOrDeleted: mutationDebugInfo.rowsSetOrDeleted, + durationMs: mutationDebugInfo.durationMs, + }, + }, { depth: null, maxArrayLength: null })); return result; }); }; diff --git a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts index 0fa07cfb9..dd37a2659 100644 --- a/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts +++ b/apps/bulldozer-js/src/databases/piledriver/data-structures/augmented-tree-map.ts @@ -26,10 +26,11 @@ const noAugmentation = Symbol("no-augmentation"); type NoAugmentation = typeof noAugmentation; type Entry = [K, V]; type Child = { ref: PiledriverHeapObject, augmentation: A, size: number, entryCount: number, minKey: K, maxKey: K }; -type Split = { entry: Entry, right: Child }; +type StoredValue = V | PiledriverHeapObject; +type Split = { entry: Entry>, right: Child }; // Persisted B-tree node. Children are heap objects, so unchanged subtrees are shared across versions. -type Node = { - entries: Entry[], +type Node = { + entries: Entry>[], children: Child[], augmentation: A, size: number, @@ -176,7 +177,7 @@ export class AugmentedTreeMultiMap): Promise { for (let node = await this.node(this.root?.ref ?? null); node;) { const { index, found } = this.search(node.entries, key); - if (found) return node.entries[index][1]; + if (found) return await this.loadValue(node.entries[index][1]); node = await this.node(node.children[index]?.ref ?? null); } } @@ -206,8 +207,8 @@ export class AugmentedTreeMultiMap= options.limit) || !tree.overlaps(child, options)) return; const node = (await tree.node(child.ref))!; - const visitEntry = function* (entry: Entry, Value>): Iterable, Value>> { - if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield entry; + const visitEntry = async function* (entry: Entry, StoredValue>): AsyncIterable, Value>> { + if (isInRange(entry[0]) && (options.limit === undefined || yielded++ < options.limit)) yield [entry[0], await tree.loadValue(entry[1])]; }; if (options.reverse) { @@ -246,6 +247,14 @@ export class AugmentedTreeMultiMap, Value, Augmentation> : null; } + private storeValue(value: Value): StoredValue { + return value !== null && typeof value === "object" ? asHeapObject(value) : value; + } + + private async loadValue(value: StoredValue): Promise { + return value !== null && typeof value === "object" && isPiledriverHeapObjectSymbol in value ? await value.get() as Value : value; + } + private async empty() { return this.options.initialAugmentation !== undefined ? this.options.initialAugmentation : await this.options.mergeAugmentations(); } @@ -268,7 +277,7 @@ export class AugmentedTreeMultiMap, Value>[], children: Child, Augmentation>[] = []) { + private async make(entries: Entry, StoredValue>[], children: Child, Augmentation>[] = []) { if (!entries.length && children.length !== 1) throw new Error("Invalid empty B-tree node"); if (!entries.length) { const [onlyChild] = children; @@ -292,7 +301,7 @@ export class AugmentedTreeMultiMap, Value>[], key: MultiKey) { + private search(entries: Entry, StoredValue>[], key: MultiKey) { let low = 0, high = entries.length; while (low < high) { const mid = (low + high) >> 1; @@ -337,21 +346,21 @@ export class AugmentedTreeMultiMap, Value>[], children: Child, Augmentation>[]) { + private async split(entries: Entry, StoredValue>[], children: Child, Augmentation>[]) { const middle = entries.length >> 1; const left = await this.make(entries.slice(0, middle), children.length ? children.slice(0, middle + 1) : []); const right = await this.make(entries.slice(middle + 1), children.length ? children.slice(middle + 1) : []); return { root: left, split: { entry: entries[middle], right } }; } - private async done(entries: Entry, Value>[], children: Child, Augmentation>[], added: boolean) { + private async done(entries: Entry, StoredValue>[], children: Child, Augmentation>[], added: boolean) { const result = entries.length > this.maxEntries() ? await this.split(entries, children) : { root: await this.make(entries, children) }; return { ...result, added }; } private async upsert(heapObject: PiledriverHeapObject | null, key: MultiKey, value: Value, replace: boolean): Promise<{ root: Child, Augmentation>, split?: Split, Value, Augmentation>, added: boolean }> { const node = await this.node(heapObject); - if (!node) return { root: await this.make([[key, value]]), added: true }; + if (!node) return { root: await this.make([[key, this.storeValue(value)]]), added: true }; const { index, found } = this.search(node.entries, key); const entries = [...node.entries]; @@ -359,12 +368,12 @@ export class AugmentedTreeMultiMap, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, Value> }> { + private async deleteMin(child: Child, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, StoredValue> }> { const node = (await this.node(child.ref))!; const entries = [...node.entries]; const children = [...node.children] as (Child, Augmentation> | null)[]; @@ -432,7 +441,7 @@ export class AugmentedTreeMultiMap, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, Value> }> { + private async deleteMax(child: Child, Augmentation>): Promise<{ root: Child, Augmentation> | null, entry: Entry, StoredValue> }> { const node = (await this.node(child.ref))!; const entries = [...node.entries]; const children = [...node.children] as (Child, Augmentation> | null)[]; @@ -448,13 +457,13 @@ export class AugmentedTreeMultiMap, Value>[], children: (Child, Augmentation> | null)[], isRoot: boolean, deleted: boolean) { + private async afterDelete(entries: Entry, StoredValue>[], children: (Child, Augmentation> | null)[], isRoot: boolean, deleted: boolean) { const liveChildren = children.filter(child => child !== null); if (!entries.length) return { root: isRoot ? liveChildren[0] ?? null : liveChildren[0] ? await this.make([], [liveChildren[0]]) : null, deleted }; return { root: await this.make(entries, liveChildren), deleted }; } - private async fixChild(entries: Entry, Value>[], children: (Child, Augmentation> | null)[], index: number, isRoot: boolean): Promise<{ root: Child, Augmentation> | null, deleted: boolean }> { + private async fixChild(entries: Entry, StoredValue>[], children: (Child, Augmentation> | null)[], index: number, isRoot: boolean): Promise<{ root: Child, Augmentation> | null, deleted: boolean }> { const child = children[index]; if (child && child.entryCount >= this.minEntries()) return await this.afterDelete(entries, children, isRoot, true); @@ -491,7 +500,7 @@ export class AugmentedTreeMultiMap, Augmentation> | null, entry: Entry, Value>, right: Child, Augmentation> | null) { + private async mergeChildren(left: Child, Augmentation> | null, entry: Entry, StoredValue>, right: Child, Augmentation> | null) { const leftNode = left ? (await this.node(left.ref))! : { entries: [], children: [] }; const rightNode = right ? (await this.node(right.ref))! : { entries: [], children: [] }; return await this.make( @@ -518,7 +527,7 @@ export class AugmentedTreeMultiMap, - heapObjects: ReadonlySet, + objects: Set, + heapObjects: Set, }; -const emptySerializationPath: SerializationPath = { objects: new Set(), heapObjects: new Set() }; +const emptySerializationPath = (): SerializationPath => ({ objects: new Set(), heapObjects: new Set() }); +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, + arrayItems: number, + objectNodes: number, + objectEntries: number, + heapReferenceNodes: number, + serializeToJsonableTotalMs: number, + jsonStringifyTotalMs: number, + textEncodeTotalMs: number, + heapObjectCacheHits: number, + heapObjectCacheMisses: number, + heapObjectCacheHitAwaitTotalMs: number, + heapObjectCacheMissAwaitTotalMs: number, + heapObjectGetTotalMs: number, + heapObjectSerializeTotalMs: number, + heapObjectInsertAwaitTotalMs: number, + heapInsertFlushes: number, + heapInsertValues: number, + heapInsertAllTotalMs: number, + branchStats: Map, + heapObjectCacheMissesByShape: Map, + heapObjectCacheMissInlineNodeCountsByShape: Map, +}; +type PiledriverSerializationBranchStats = { + primitiveNodes: number, + arrayNodes: number, + objectNodes: number, + objectEntries: number, + heapReferenceNodes: number, + heapObjectCacheHits: number, + heapObjectCacheMisses: number, + heapObjectCacheMissesByShape: Map, + heapObjectCacheMissInlineNodeCountsByShape: Map, +}; +type PiledriverInlineNodeCounts = { + primitiveNodes: number, + arrayNodes: number, + objectNodes: number, + heapReferenceNodes: number, +}; + +function emptyPiledriverSerializationBranchStats(): PiledriverSerializationBranchStats { + return { + primitiveNodes: 0, + arrayNodes: 0, + objectNodes: 0, + objectEntries: 0, + heapReferenceNodes: 0, + heapObjectCacheHits: 0, + heapObjectCacheMisses: 0, + heapObjectCacheMissesByShape: new Map(), + heapObjectCacheMissInlineNodeCountsByShape: new Map(), + }; +} + +function emptyPiledriverSerializationTimingStats(): PiledriverSerializationTimingStats { + return { + primitiveNodes: 0, + arrayNodes: 0, + arrayItems: 0, + objectNodes: 0, + objectEntries: 0, + heapReferenceNodes: 0, + serializeToJsonableTotalMs: 0, + jsonStringifyTotalMs: 0, + textEncodeTotalMs: 0, + heapObjectCacheHits: 0, + heapObjectCacheMisses: 0, + heapObjectCacheHitAwaitTotalMs: 0, + heapObjectCacheMissAwaitTotalMs: 0, + heapObjectGetTotalMs: 0, + heapObjectSerializeTotalMs: 0, + heapObjectInsertAwaitTotalMs: 0, + heapInsertFlushes: 0, + heapInsertValues: 0, + heapInsertAllTotalMs: 0, + branchStats: new Map(), + heapObjectCacheMissesByShape: new Map(), + heapObjectCacheMissInlineNodeCountsByShape: new Map(), + }; +} + +function incrementMapCount(map: Map, key: string) { + map.set(key, (map.get(key) ?? 0) + 1); +} + +function addMapCount(map: Map, key: string, value: number) { + map.set(key, (map.get(key) ?? 0) + value); +} + +function emptyPiledriverInlineNodeCounts(): PiledriverInlineNodeCounts { + return { + primitiveNodes: 0, + arrayNodes: 0, + objectNodes: 0, + heapReferenceNodes: 0, + }; +} + +function addPiledriverInlineNodeCounts(target: PiledriverInlineNodeCounts, source: PiledriverInlineNodeCounts) { + target.primitiveNodes += source.primitiveNodes; + target.arrayNodes += source.arrayNodes; + target.objectNodes += source.objectNodes; + target.heapReferenceNodes += source.heapReferenceNodes; +} + +function totalInlineNodes(counts: PiledriverInlineNodeCounts) { + return counts.primitiveNodes + counts.arrayNodes + counts.objectNodes + counts.heapReferenceNodes; +} + +function countPiledriverInlineNodes(obj: PiledriverObject, path: Set = new Set()): PiledriverInlineNodeCounts { + switch (typeof obj) { + case "number": + case "string": + case "boolean": { + return { ...emptyPiledriverInlineNodeCounts(), primitiveNodes: 1 }; + } + case "object": { + if (obj === null) return { ...emptyPiledriverInlineNodeCounts(), primitiveNodes: 1 }; + if (isPiledriverHeapObjectSymbol in obj) return { ...emptyPiledriverInlineNodeCounts(), heapReferenceNodes: 1 }; + if (path.has(obj)) throw new Error("Piledriver objects must not contain cycles"); + const childPath = new Set(path).add(obj); + const counts = emptyPiledriverInlineNodeCounts(); + if (Array.isArray(obj)) { + counts.arrayNodes++; + for (const item of obj) addPiledriverInlineNodeCounts(counts, countPiledriverInlineNodes(item, childPath)); + } else { + counts.objectNodes++; + for (const value of Object.values(obj)) addPiledriverInlineNodeCounts(counts, countPiledriverInlineNodes(value, childPath)); + } + return counts; + } + default: { + throw new Error("Assertion error: Unknown type of Piledriver object " + typeof obj); + } + } +} + +function classifyHeapObjectPayload(obj: PiledriverObject): string { + if (obj === null) return "null"; + if (Array.isArray(obj)) return `array:${obj.length}`; + if (typeof obj !== "object") return typeof obj; + const keys = Object.keys(obj).sort(); + if (keys.includes("entries") && keys.includes("children") && keys.includes("augmentation") && keys.includes("size") && keys.includes("minKey") && keys.includes("maxKey")) { + const entries = Reflect.get(obj, "entries"); + const children = Reflect.get(obj, "children"); + return `btree-node:entries=${Array.isArray(entries) ? entries.length : "?"}:children=${Array.isArray(children) ? children.length : "?"}`; + } + if (keys.includes("key") && keys.includes("id") && keys.length === 2) return "multi-key"; + if (keys.includes("groupKey") && keys.includes("rows") && keys.length === 2) return "group-with-rows"; + if (keys.includes("inputRowData") && keys.includes("outputRowData") && keys.includes("stateAfter")) return "left-fold-row"; + if (keys.includes("state") && keys.includes("nextTriggerTimeMs") && keys.includes("emittedRows")) return "time-fold-row"; + if (keys.includes("outputRows") && keys.length === 1) return "time-fold-group"; + if (keys.includes("rowData") && keys.includes("rowIdentifier") && keys.includes("rowSortKey") && keys.includes("groupKey")) return "row-object"; + return `object:${keys.slice(0, 6).join(",")}${keys.length > 6 ? ",..." : ""}`; +} + +function serializationBranchKey(path: readonly string[]): string { + if (path[0] === "snapshot" && path[1] === "serializedTables") return `table:${path[2]}`; + return path[0] ?? ""; +} + +function serializationBranchStats(stats: PiledriverSerializationTimingStats | undefined, path: readonly string[]) { + if (stats === undefined) return undefined; + const key = serializationBranchKey(path); + let result = stats.branchStats.get(key); + if (result === undefined) { + result = emptyPiledriverSerializationBranchStats(); + stats.branchStats.set(key, result); + } + return result; +} export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: PiledriverDatabaseOptions = {}): PiledriverDatabase { // TODO actually support cycles both for heap and non-heap objects (right now they are detected and rejected) @@ -85,6 +266,8 @@ 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)); @@ -92,36 +275,112 @@ 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; - const getHeapKeyAndSeq = async (heapObj: PiledriverHeapObject, path: SerializationPath = emptySerializationPath): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { + 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++) { + batch[i].resolve({ key: inserted.keys[i], 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(); + } + }); + }; + + const getHeapKeyAndSeq = async (heapObj: PiledriverHeapObject, path: SerializationPath = emptySerializationPath(), serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ key: ArrayBuffer, seq: DatabaseSeq }> => { // Must be checked before the memo lookup: awaiting the memoized promise of an ancestor // that is still being serialized would deadlock. if (path.heapObjects.has(heapObj)) throw new Error("Piledriver objects must not contain cycles (found a cycle of heap objects)"); const existing = heapKeysAndSeqByHeapObjects.get(heapObj); - if (existing) return await existing; + if (existing) { + if (serializationTimingStats !== undefined) { + serializationTimingStats.heapObjectCacheHits++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.heapObjectCacheHits++; + } + const cacheHitAwaitStartedAt = performance.now(); + try { + return await existing; + } finally { + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectCacheHitAwaitTotalMs += performance.now() - cacheHitAwaitStartedAt; + } + } + if (serializationTimingStats !== undefined) { + serializationTimingStats.heapObjectCacheMisses++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.heapObjectCacheMisses++; + } const promise = (async () => { // A heap object starts a fresh plain-object path; plain object cycles can't span heap // boundaries without also forming a heap cycle, which is tracked separately. const childPath: SerializationPath = { objects: new Set(), heapObjects: new Set(path.heapObjects).add(heapObj) }; return await traceSpan("bulldozer-js.piledriver.heap.serializeAndInsert", async () => { - const serialized = await serializePiledriverObject(await heapObj.get(), childPath); - const inserted = await heapDump.insertAll([serialized.buffer]); - return { - key: inserted.keys[0], - seq: lowLevelDb.combineSeqs(serialized.seq, inserted.seq), - }; + const heapObjectGetStartedAt = performance.now(); + const heapObject = await heapObj.get(); + let shape: string | undefined; + if (serializationTimingStats !== undefined) { + shape = classifyHeapObjectPayload(heapObject); + const inlineNodeCount = totalInlineNodes(countPiledriverInlineNodes(heapObject)); + incrementMapCount(serializationTimingStats.heapObjectCacheMissesByShape, shape); + addMapCount(serializationTimingStats.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount); + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) { + incrementMapCount(branch.heapObjectCacheMissesByShape, shape); + addMapCount(branch.heapObjectCacheMissInlineNodeCountsByShape, shape, inlineNodeCount); + } + } + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectGetTotalMs += performance.now() - heapObjectGetStartedAt; + const heapObjectSerializeStartedAt = performance.now(); + const serialized = await serializePiledriverObject(heapObject, childPath, serializationTimingStats, logicalPath); + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectSerializeTotalMs += performance.now() - heapObjectSerializeStartedAt; + const heapObjectInsertStartedAt = performance.now(); + const result = await insertHeapObjectBatched(serialized.buffer, serialized.seq, serializationTimingStats); + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectInsertAwaitTotalMs += performance.now() - heapObjectInsertStartedAt; + return result; }); })(); heapKeysAndSeqByHeapObjects.set(heapObj, promise); let result; + const cacheMissAwaitStartedAt = performance.now(); try { result = await promise; } catch (error) { // Don't leave a poisoned rejected promise in the cache; a later retry may succeed. if (heapKeysAndSeqByHeapObjects.get(heapObj) === promise) heapKeysAndSeqByHeapObjects.delete(heapObj); throw error; + } finally { + if (serializationTimingStats !== undefined) serializationTimingStats.heapObjectCacheMissAwaitTotalMs += performance.now() - cacheMissAwaitStartedAt; } cacheHeapObjectByKey(result.key, heapObj, result.seq); return result; @@ -167,30 +426,56 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: return { object: heapObj, seq }; }; - const serializePiledriverObjectToJsonableObject = async (obj: PiledriverObject, path: SerializationPath): Promise<{ jsonableObject: unknown, seq: DatabaseSeq }> => { + const serializePiledriverObjectToJsonableObject = async (obj: PiledriverObject, path: SerializationPath, serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ jsonableObject: unknown, seq: DatabaseSeq }> => { switch (typeof obj) { case "number": { + if (serializationTimingStats !== undefined) { + serializationTimingStats.primitiveNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.primitiveNodes++; + } if (!Number.isFinite(obj)) return { jsonableObject: [obj.toString()], seq: lowLevelDb.initialSeq }; if (Object.is(obj, -0)) return { jsonableObject: ["-0"], seq: lowLevelDb.initialSeq }; // intentionally fall through to the primitive case below } case "string": case "boolean": { + if (serializationTimingStats !== undefined && typeof obj !== "number") { + serializationTimingStats.primitiveNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.primitiveNodes++; + } return { jsonableObject: obj, seq: lowLevelDb.initialSeq }; } case "object": { if (obj === null) { + if (serializationTimingStats !== undefined) { + serializationTimingStats.primitiveNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.primitiveNodes++; + } return { jsonableObject: obj, seq: lowLevelDb.initialSeq }; } else if (Array.isArray(obj)) { + if (serializationTimingStats !== undefined) { + serializationTimingStats.arrayNodes++; + serializationTimingStats.arrayItems += obj.length; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.arrayNodes++; + } if (path.objects.has(obj)) throw new Error("Piledriver objects must not contain cycles"); const childPath: SerializationPath = { ...path, objects: new Set(path.objects).add(obj) }; - const itemsSerializeResults = await Promise.all(obj.map(async o => await serializePiledriverObjectToJsonableObject(o, childPath))); + const itemsSerializeResults = await Promise.all(obj.map(async o => await serializePiledriverObjectToJsonableObject(o, childPath, serializationTimingStats, logicalPath))); return { jsonableObject: ["array", itemsSerializeResults.map(r => r.jsonableObject)], seq: lowLevelDb.combineSeqs(...itemsSerializeResults.map(r => r.seq)), }; } else if (isPiledriverHeapObjectSymbol in obj) { - const heapKeyAndSeq = await getHeapKeyAndSeq(obj, path); + if (serializationTimingStats !== undefined) { + serializationTimingStats.heapReferenceNodes++; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) branch.heapReferenceNodes++; + } + const heapKeyAndSeq = await getHeapKeyAndSeq(obj, path, serializationTimingStats, logicalPath); return { jsonableObject: ["heap-reference", encodeBase64(new Uint8Array(heapKeyAndSeq.key))], seq: heapKeyAndSeq.seq, @@ -201,7 +486,17 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: if (path.objects.has(obj)) throw new Error("Piledriver objects must not contain cycles"); const childPath: SerializationPath = { ...path, objects: new Set(path.objects).add(obj) }; - const entriesSerializeResults = await Promise.all(Object.entries(obj).map(async ([k, v]) => [k, await serializePiledriverObjectToJsonableObject(v, childPath)] as const)); + const entries = Object.entries(obj); + if (serializationTimingStats !== undefined) { + serializationTimingStats.objectNodes++; + serializationTimingStats.objectEntries += entries.length; + const branch = serializationBranchStats(serializationTimingStats, logicalPath); + if (branch !== undefined) { + branch.objectNodes++; + branch.objectEntries += entries.length; + } + } + const entriesSerializeResults = await Promise.all(entries.map(async ([k, v]) => [k, await serializePiledriverObjectToJsonableObject(v, childPath, serializationTimingStats, [...logicalPath, k])] as const)); return { jsonableObject: Object.fromEntries(entriesSerializeResults.map(([k, v]) => [k, v.jsonableObject] as const)), seq: lowLevelDb.combineSeqs(...entriesSerializeResults.map(([_, v]) => v.seq)), @@ -214,10 +509,18 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: } }; - const serializePiledriverObject = async (obj: PiledriverObject, path: SerializationPath = emptySerializationPath): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => { - const toJsonableResponse = await serializePiledriverObjectToJsonableObject(obj, path); + const serializePiledriverObject = async (obj: PiledriverObject, path: SerializationPath = emptySerializationPath(), serializationTimingStats?: PiledriverSerializationTimingStats, logicalPath: readonly string[] = []): Promise<{ buffer: ArrayBuffer, seq: DatabaseSeq }> => { + const toJsonableStartedAt = performance.now(); + const toJsonableResponse = await serializePiledriverObjectToJsonableObject(obj, path, serializationTimingStats, logicalPath); + if (serializationTimingStats !== undefined) serializationTimingStats.serializeToJsonableTotalMs += performance.now() - toJsonableStartedAt; + const jsonStringifyStartedAt = performance.now(); + const json = JSON.stringify(toJsonableResponse.jsonableObject); + if (serializationTimingStats !== undefined) serializationTimingStats.jsonStringifyTotalMs += performance.now() - jsonStringifyStartedAt; + const textEncodeStartedAt = performance.now(); + const buffer = new TextEncoder().encode(json).buffer; + if (serializationTimingStats !== undefined) serializationTimingStats.textEncodeTotalMs += performance.now() - textEncodeStartedAt; return { - buffer: new TextEncoder().encode(JSON.stringify(toJsonableResponse.jsonableObject)).buffer, + buffer, seq: toJsonableResponse.seq, }; }; @@ -311,8 +614,74 @@ export function declarePiledriverDatabase(lowLevelDb: LowLevelDatabase, options: }, async setRootObject(key, value): Promise<{ seq: DatabaseSeq }> { return await traceSpan("bulldozer-js.piledriver.setRootObject", async () => { - const { buffer, seq } = await serializePiledriverObject(value); + const timingStats = emptyPiledriverSerializationTimingStats(); + const startedAt = performance.now(); + const serializeStartedAt = performance.now(); + const serializeCpuStartedAt = process.cpuUsage(); + const { buffer, seq } = await serializePiledriverObject(value, emptySerializationPath(), timingStats); + const serializeCpuUsage = process.cpuUsage(serializeCpuStartedAt); + const serializePiledriverObjectMs = performance.now() - serializeStartedAt; + const serializeCpuMs = (serializeCpuUsage.user + serializeCpuUsage.system) / 1000; + const rootStoreSetAllStartedAt = performance.now(); const { seq: rootSeq } = await rootStore.setAll([{ key, value: buffer }], { requiresSeq: seq }); + const rootStoreSetAllMs = performance.now() - rootStoreSetAllStartedAt; + const topSerializationBranches = [...timingStats.branchStats.entries()] + .map(([branch, stats]) => ({ + branch, + totalNodes: stats.primitiveNodes + stats.arrayNodes + stats.objectNodes + stats.heapReferenceNodes, + primitiveNodes: stats.primitiveNodes, + arrayNodes: stats.arrayNodes, + objectNodes: stats.objectNodes, + objectEntries: stats.objectEntries, + heapReferenceNodes: stats.heapReferenceNodes, + heapObjectCacheHits: stats.heapObjectCacheHits, + heapObjectCacheMisses: stats.heapObjectCacheMisses, + topHeapObjectCacheMissShapes: [...stats.heapObjectCacheMissesByShape.entries()] + .map(([shape, count]) => { + const totalInlineNodeCount = stats.heapObjectCacheMissInlineNodeCountsByShape.get(shape) ?? 0; + return { shape, count, totalInlineNodeCount, averageInlineNodeCount: count === 0 ? 0 : totalInlineNodeCount / count }; + }) + .sort((a, b) => b.count - a.count) + .slice(0, 5), + })) + .sort((a, b) => b.totalNodes - a.totalNodes) + .slice(0, 5); + const topHeapObjectCacheMissShapes = [...timingStats.heapObjectCacheMissesByShape.entries()] + .map(([shape, count]) => { + const totalInlineNodeCount = timingStats.heapObjectCacheMissInlineNodeCountsByShape.get(shape) ?? 0; + return { shape, count, totalInlineNodeCount, averageInlineNodeCount: count === 0 ? 0 : totalInlineNodeCount / count }; + }) + .sort((a, b) => b.count - a.count) + .slice(0, 10); + console.debug("bulldozer-js piledriver setRootObject timing", { + elapsedMs: performance.now() - startedAt, + serializePiledriverObjectMs, + serializeCpuMs, + serializeCpuToWallRatio: serializePiledriverObjectMs === 0 ? 0 : serializeCpuMs / serializePiledriverObjectMs, + rootStoreSetAllMs, + rootValueBytes: buffer.byteLength, + primitiveNodes: timingStats.primitiveNodes, + arrayNodes: timingStats.arrayNodes, + arrayItems: timingStats.arrayItems, + objectNodes: timingStats.objectNodes, + objectEntries: timingStats.objectEntries, + heapReferenceNodes: timingStats.heapReferenceNodes, + serializeToJsonableTotalMs: timingStats.serializeToJsonableTotalMs, + jsonStringifyTotalMs: timingStats.jsonStringifyTotalMs, + textEncodeTotalMs: timingStats.textEncodeTotalMs, + heapObjectCacheHits: timingStats.heapObjectCacheHits, + heapObjectCacheMisses: timingStats.heapObjectCacheMisses, + heapObjectCacheHitAwaitTotalMs: timingStats.heapObjectCacheHitAwaitTotalMs, + heapObjectCacheMissAwaitTotalMs: timingStats.heapObjectCacheMissAwaitTotalMs, + heapObjectGetTotalMs: timingStats.heapObjectGetTotalMs, + heapObjectSerializeTotalMs: timingStats.heapObjectSerializeTotalMs, + heapObjectInsertAwaitTotalMs: timingStats.heapObjectInsertAwaitTotalMs, + heapInsertFlushes: timingStats.heapInsertFlushes, + heapInsertValues: timingStats.heapInsertValues, + heapInsertAllTotalMs: timingStats.heapInsertAllTotalMs, + topHeapObjectCacheMissShapes, + topSerializationBranches, + }); return { seq: rootSeq }; }); },