mirror of
https://github.com/stack-auth/stack.git
synced 2026-07-20 21:29:36 +08:00
Batch Bulldozer transaction operations every 10ms
This commit is contained in:
Konstantin Wohlwend
parent
b6a5fb2ac6
commit
3cfe5c8ad3
@ -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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@ -10,6 +10,13 @@ type BinaryDatabase = lmdb.Database<Buffer, Uint8Array>;
|
||||
type VersionedBinaryDatabase = BinaryDatabase & {
|
||||
getEntry(key: Buffer): { value: Buffer, version?: number } | undefined,
|
||||
};
|
||||
type PendingCommitOperation = {
|
||||
seqId: string,
|
||||
requiresSeq: DatabaseSeq,
|
||||
action: (version: number) => Promise<void>,
|
||||
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<void>((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<string, Promise<void>>();
|
||||
const combinedSeqToAvailability = new Map<string, Promise<void>>();
|
||||
const combinedSeqToDurability = new Map<string, Promise<void>>();
|
||||
const combinedSeqDependencies = new Map<string, string[]>();
|
||||
let pendingCommitOperations: PendingCommitOperation[] = [];
|
||||
let pendingCommitFlushTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let pendingCommitFlushPromise: Promise<void> | 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<void>) => {
|
||||
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<void>) => {
|
||||
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<void>) => {
|
||||
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<string>): Promise<void> => {
|
||||
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);
|
||||
|
||||
@ -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<PiledriverHeapObject>;
|
||||
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<string, PiledriverSerializationBranchStats>,
|
||||
heapObjectCacheMissesByShape: Map<string, number>,
|
||||
heapObjectCacheMissInlineNodeCountsByShape: Map<string, number>,
|
||||
@ -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<string, { refIdentity: string, object: WeakRef<PiledriverHeapObject>, seq: DatabaseSeq }>();
|
||||
const heapObjectsByHeapKeyFinalizer = new FinalizationRegistry(([keyBase64, refIdentity]: [string, string]) => heapObjectsByHeapKeyBase64.get(keyBase64)?.refIdentity === refIdentity && heapObjectsByHeapKeyBase64.delete(keyBase64));
|
||||
const heapKeysAndSeqByHeapObjects = new WeakMap<PiledriverHeapObject, Promise<{ key: ArrayBuffer, seq: DatabaseSeq }>>();
|
||||
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<PiledriverSerializationTimingStats, number>();
|
||||
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,
|
||||
});
|
||||
|
||||
Loading…
Reference in New Issue
Block a user