diff --git a/.changeset/wa-sqlite-rpc-transfer-efficiency.md b/.changeset/wa-sqlite-rpc-transfer-efficiency.md new file mode 100644 index 00000000..bef0a570 --- /dev/null +++ b/.changeset/wa-sqlite-rpc-transfer-efficiency.md @@ -0,0 +1,5 @@ +--- +'@treecrdt/wa-sqlite': patch +--- + +Reduce wa-sqlite worker payload read copies by returning transferable binary RPC results. diff --git a/packages/treecrdt-wa-sqlite/e2e/src/responsiveness.ts b/packages/treecrdt-wa-sqlite/e2e/src/responsiveness.ts index 442f37f1..0841a976 100644 --- a/packages/treecrdt-wa-sqlite/e2e/src/responsiveness.ts +++ b/packages/treecrdt-wa-sqlite/e2e/src/responsiveness.ts @@ -56,6 +56,36 @@ type ReadSampleResult = { finalChildCount: number; }; +type PayloadSeedOptions = { + nodeInt?: number; + payloadBytes: number; + replicaLabel?: string; +}; + +type PayloadSeedResult = { + ok: true; + node: string; + payloadBytes: number; +}; + +type PayloadReadSampleOptions = { + node: string; + payloadBytes: number; + samples: number; + intervalMs?: number; +}; + +type PayloadReadSampleResult = { + ok: true; + samples: number; + payloadBytes: number; + durationsMs: number[]; + minMs: number; + p50Ms: number; + p95Ms: number; + maxMs: number; +}; + let responsivenessClient: TreecrdtClient | null = null; let writeStatus: WriteBatchStatus = { state: 'idle' }; let writePromise: Promise | null = null; @@ -70,6 +100,15 @@ function orderKeyFromOffset(offset: number): Uint8Array { return bytes; } +function payloadBytesFromSeed(seed: number, size: number): Uint8Array { + if (!Number.isInteger(size) || size <= 0) throw new Error(`invalid payload size: ${size}`); + const payload = new Uint8Array(size); + for (let i = 0; i < payload.length; i += 1) { + payload[i] = (seed + i * 31) % 251; + } + return payload; +} + async function sleep(ms: number): Promise { await new Promise((resolve) => setTimeout(resolve, ms)); } @@ -233,6 +272,54 @@ export async function sampleResponsivenessReads( }; } +export async function seedResponsivenessPayload( + opts: PayloadSeedOptions, +): Promise { + const client = ensureClient(); + const replica = replicaFromLabel(opts.replicaLabel ?? 'responsiveness-payload'); + const nodeInt = opts.nodeInt ?? 900_000; + const node = nodeIdFromInt(nodeInt); + await client.ops.append( + makeOp(replica, 1, 1, { + type: 'insert', + parent: rootNode(), + node, + orderKey: orderKeyFromOffset(nodeInt), + payload: payloadBytesFromSeed(nodeInt, opts.payloadBytes), + }), + ); + return { ok: true, node, payloadBytes: opts.payloadBytes }; +} + +export async function sampleResponsivenessPayloadReads( + opts: PayloadReadSampleOptions, +): Promise { + const client = ensureClient(); + const intervalMs = opts.intervalMs ?? 0; + const durationsMs: number[] = []; + + for (let i = 0; i < opts.samples; i += 1) { + const start = performance.now(); + const payload = await client.tree.getPayload(opts.node); + durationsMs.push(performance.now() - start); + if (!payload) throw new Error(`missing payload for node ${opts.node}`); + if (payload.byteLength !== opts.payloadBytes) { + throw new Error( + `payload length mismatch for node ${opts.node}: expected ${opts.payloadBytes}, got ${payload.byteLength}`, + ); + } + if (intervalMs > 0) await sleep(intervalMs); + } + + return { + ok: true, + samples: opts.samples, + payloadBytes: opts.payloadBytes, + durationsMs, + ...summarizeDurations(durationsMs), + }; +} + export async function closeResponsivenessClient(): Promise { const client = responsivenessClient; responsivenessClient = null; @@ -248,6 +335,8 @@ declare global { __treecrdtResponsivenessWriteStatus?: typeof responsivenessWriteStatus; __waitTreecrdtResponsivenessWrites?: typeof waitResponsivenessWrites; __sampleTreecrdtResponsivenessReads?: typeof sampleResponsivenessReads; + __seedTreecrdtResponsivenessPayload?: typeof seedResponsivenessPayload; + __sampleTreecrdtResponsivenessPayloadReads?: typeof sampleResponsivenessPayloadReads; __closeTreecrdtResponsivenessClient?: typeof closeResponsivenessClient; } } @@ -258,5 +347,7 @@ if (typeof window !== 'undefined') { window.__treecrdtResponsivenessWriteStatus = responsivenessWriteStatus; window.__waitTreecrdtResponsivenessWrites = waitResponsivenessWrites; window.__sampleTreecrdtResponsivenessReads = sampleResponsivenessReads; + window.__seedTreecrdtResponsivenessPayload = seedResponsivenessPayload; + window.__sampleTreecrdtResponsivenessPayloadReads = sampleResponsivenessPayloadReads; window.__closeTreecrdtResponsivenessClient = closeResponsivenessClient; } diff --git a/packages/treecrdt-wa-sqlite/e2e/tests/responsiveness.spec.ts b/packages/treecrdt-wa-sqlite/e2e/tests/responsiveness.spec.ts index 15f2d4d4..24d02eee 100644 --- a/packages/treecrdt-wa-sqlite/e2e/tests/responsiveness.spec.ts +++ b/packages/treecrdt-wa-sqlite/e2e/tests/responsiveness.spec.ts @@ -22,10 +22,28 @@ type WriteMetrics = { batchDurationsMs: number[]; }; +type PayloadSeedMetrics = { + ok: true; + node: string; + payloadBytes: number; +}; + +type PayloadReadMetrics = { + ok: true; + samples: number; + payloadBytes: number; + durationsMs: number[]; + minMs: number; + p50Ms: number; + p95Ms: number; + maxMs: number; +}; + const totalOps = Number(process.env.TREECRDT_RESPONSIVENESS_TOTAL_OPS ?? 5_000); -const batchSize = Number(process.env.TREECRDT_RESPONSIVENESS_BATCH_SIZE ?? 500); +const batchSize = Number(process.env.TREECRDT_RESPONSIVENESS_BATCH_SIZE ?? totalOps); const samples = Number(process.env.TREECRDT_RESPONSIVENESS_READ_SAMPLES ?? 20); const intervalMs = Number(process.env.TREECRDT_RESPONSIVENESS_READ_INTERVAL_MS ?? 0); +const payloadBytes = Number(process.env.TREECRDT_RESPONSIVENESS_PAYLOAD_BYTES ?? 64 * 1024); async function waitForHarness(page: Page) { await page.goto('/'); @@ -34,6 +52,8 @@ async function waitForHarness(page: Page) { typeof window.__openTreecrdtResponsivenessClient === 'function' && typeof window.__startTreecrdtResponsivenessWriteBatches === 'function' && typeof window.__sampleTreecrdtResponsivenessReads === 'function' && + typeof window.__seedTreecrdtResponsivenessPayload === 'function' && + typeof window.__sampleTreecrdtResponsivenessPayloadReads === 'function' && typeof window.__waitTreecrdtResponsivenessWrites === 'function', ); } @@ -71,6 +91,36 @@ async function sampleReads( ); } +async function seedPayload( + page: Page, + opts: { payloadBytes: number; nodeInt?: number }, +): Promise { + return page.evaluate(async (payloadOpts) => { + const seed = window.__seedTreecrdtResponsivenessPayload; + if (!seed) throw new Error('__seedTreecrdtResponsivenessPayload not available'); + return await seed(payloadOpts); + }, opts); +} + +async function samplePayloadReads( + page: Page, + opts: { node: string; payloadBytes: number; samples?: number; intervalMs?: number }, +): Promise { + return page.evaluate( + async ({ node, payloadBytes, samples, intervalMs }) => { + const sample = window.__sampleTreecrdtResponsivenessPayloadReads; + if (!sample) throw new Error('__sampleTreecrdtResponsivenessPayloadReads not available'); + return await sample({ node, payloadBytes, samples, intervalMs }); + }, + { + node: opts.node, + payloadBytes: opts.payloadBytes, + samples: opts.samples ?? samples, + intervalMs: opts.intervalMs ?? intervalMs, + }, + ); +} + async function waitWrites(page: Page): Promise { return page.evaluate(async () => { const wait = window.__waitTreecrdtResponsivenessWrites; @@ -118,9 +168,14 @@ test.describe('worker read responsiveness under write pressure', () => { expect(writerSummary).toEqual({ mode: 'worker', runtime, storage: 'opfs' }); expect(readerSummary).toEqual({ mode: 'worker', runtime, storage: 'opfs' }); + const payloadSeed = await seedPayload(writerPage, { payloadBytes }); await startWrites(writerPage, { batchCount, batchSize }); - const [reads, writes] = await Promise.all([ + const [reads, payloadReads, writes] = await Promise.all([ sampleReads(readerPage), + samplePayloadReads(readerPage, { + node: payloadSeed.node, + payloadBytes: payloadSeed.payloadBytes, + }), waitWrites(writerPage), ]); const afterWrites = await sampleReads(readerPage, { samples: 1, intervalMs: 0 }); @@ -140,6 +195,11 @@ test.describe('worker read responsiveness under write pressure', () => { readP95Ms: reads.p95Ms, readMaxMs: reads.maxMs, readSamples: reads.durationsMs, + payloadBytes: payloadReads.payloadBytes, + payloadReadP50Ms: payloadReads.p50Ms, + payloadReadP95Ms: payloadReads.p95Ms, + payloadReadMaxMs: payloadReads.maxMs, + payloadReadSamples: payloadReads.durationsMs, finalChildCount: reads.finalChildCount, finalChildCountAfterWrites: afterWrites.finalChildCount, }), @@ -147,9 +207,12 @@ test.describe('worker read responsiveness under write pressure', () => { expect(writes.totalOps).toBe(batchCount * batchSize); expect(reads.samples).toBe(samples); + expect(payloadReads.samples).toBe(samples); + expect(payloadReads.payloadBytes).toBe(payloadBytes); // This is a liveness guard, not a benchmark threshold. Stress runs can tighten it locally. expect(reads.maxMs).toBeLessThan(30_000); - expect(afterWrites.finalChildCount).toBe(writes.totalOps); + expect(payloadReads.maxMs).toBeLessThan(30_000); + expect(afterWrites.finalChildCount).toBe(writes.totalOps + 1); } finally { await Promise.allSettled([closeClient(writerPage), closeClient(readerPage)]); await Promise.allSettled([writerPage.close(), readerPage.close()]); diff --git a/packages/treecrdt-wa-sqlite/src/client.ts b/packages/treecrdt-wa-sqlite/src/client.ts index d8446288..4d519389 100644 --- a/packages/treecrdt-wa-sqlite/src/client.ts +++ b/packages/treecrdt-wa-sqlite/src/client.ts @@ -21,6 +21,7 @@ import { import type { LocalWriteOptions, MaterializationEvent, + MaterializationOutcome, WriteOptions, } from '@treecrdt/interface/engine'; import { @@ -28,13 +29,14 @@ import { createTreecrdtEngineLocal, } from '@treecrdt/interface/engine'; import { dbGetText } from './sql.js'; -import type { - RpcMethod, - RpcParams, - RpcPushMessage, - RpcRequest, - RpcResponse, - RpcResult, +import { + rpcBinaryResult, + type RpcMethod, + type RpcParams, + type RpcPushMessage, + type RpcRequest, + type RpcResponse, + type RpcResult, } from './rpc.js'; import { openTreecrdtDb } from './open.js'; import type { @@ -58,6 +60,8 @@ import type { } from './types.js'; export const CLIENT_CLOSED_ERROR = 'TreecrdtClient was closed'; +// Keep long browser appendMany calls from monopolizing the worker queue. +const APPEND_MANY_RPC_CHUNK_SIZE = 2500; function normalizeStorageOptions(opts: ClientOptions): NormalizedStorageOptions { const raw = opts.storage ?? { type: 'auto' }; @@ -648,14 +652,14 @@ async function createDirectClient(opts: { case 'treePayload': { const [node] = params as RpcParams<'treePayload'>; const payload = await adapter.treePayload(nodeIdToBytes16(node)); - return (payload === null ? null : Array.from(payload)) as any; + return rpcBinaryResult(payload) as any; } case 'treeNodeCount': return (await adapter.treeNodeCount()) as any; case 'treeParent': { const [node] = params as RpcParams<'treeParent'>; const result = await adapter.treeParent(nodeIdToBytes16(node)); - return result ? Array.from(result) : null; + return rpcBinaryResult(result) as any; } case 'treeExists': { const [node] = params as RpcParams<'treeExists'>; @@ -774,16 +778,31 @@ function makeTreecrdtClientFromCall(opts: { const treeParentImpl = async (node: string) => { const result = await call('treeParent', [node]); if (result === null) return null; - return nodeIdFromBytes16(Uint8Array.from(result)); + return nodeIdFromBytes16(toRpcBytes(result)); }; const treeExistsImpl = async (node: string) => Boolean(await call('treeExists', [node])); const treeGetPayloadImpl = async (node: string) => { const result = await call('treePayload', [node]); - return result === null ? null : Uint8Array.from(result); + return result === null ? null : toRpcBytes(result); }; const headLamportImpl = async () => Number(await call('headLamport', [])); const replicaMaxCounterImpl = async (replica: Operation['meta']['id']['replica']) => Number(await call('replicaMaxCounter', [Array.from(encodeReplica(replica))])); + const appendManyImpl = async (operations: Operation[], writeOpts?: WriteOptions) => { + if (operations.length <= APPEND_MANY_RPC_CHUNK_SIZE) { + const outcome = await call('appendMany', [operations]); + materialized.emitOutcome(outcome, writeOpts?.writeId); + return; + } + + const outcomes: MaterializationOutcome[] = []; + for (let start = 0; start < operations.length; start += APPEND_MANY_RPC_CHUNK_SIZE) { + outcomes.push( + await call('appendMany', [operations.slice(start, start + APPEND_MANY_RPC_CHUNK_SIZE)]), + ); + } + materialized.emitOutcome(mergeMaterializationOutcomes(outcomes), writeOpts?.writeId); + }; const localInsertImpl = async ( replica: ReplicaId, parent: string, @@ -857,10 +876,7 @@ function makeTreecrdtClientFromCall(opts: { const outcome = await call('append', [op]); materialized.emitOutcome(outcome, writeOpts?.writeId); }, - appendMany: async (ops, writeOpts?: WriteOptions) => { - const outcome = await call('appendMany', [ops]); - materialized.emitOutcome(outcome, writeOpts?.writeId); - }, + appendMany: appendManyImpl, all: () => opsSinceImpl(0), since: opsSinceImpl, children: async (parent) => opsByOpRefsImpl(await opRefsChildrenImpl(parent)), @@ -887,3 +903,15 @@ function makeTreecrdtClientFromCall(opts: { function encodeReplica(replica: ReplicaId): Uint8Array { return replicaIdToBytes(replica); } + +function toRpcBytes(bytes: Uint8Array | number[]): Uint8Array { + return bytes instanceof Uint8Array ? bytes : Uint8Array.from(bytes); +} + +function mergeMaterializationOutcomes(outcomes: MaterializationOutcome[]): MaterializationOutcome { + const last = outcomes[outcomes.length - 1]; + return { + headSeq: last?.headSeq ?? 0, + changes: outcomes.flatMap((outcome) => outcome.changes), + }; +} diff --git a/packages/treecrdt-wa-sqlite/src/common-worker.ts b/packages/treecrdt-wa-sqlite/src/common-worker.ts index fcee30e9..2547be86 100644 --- a/packages/treecrdt-wa-sqlite/src/common-worker.ts +++ b/packages/treecrdt-wa-sqlite/src/common-worker.ts @@ -5,7 +5,7 @@ import type { Operation } from '@treecrdt/interface'; import type { TreecrdtAdapter } from '@treecrdt/interface'; import type { OpenTreecrdtDbResult } from './open.js'; import { clearOpfsStorage } from './opfs.js'; -import type { RpcInitResult, RpcSqlParams } from './rpc.js'; +import { rpcBinaryResult, type RpcInitResult, type RpcSqlParams } from './rpc.js'; export function openedToRpcInitResult(opened: OpenTreecrdtDbResult): RpcInitResult { return opened.opfsError @@ -118,7 +118,7 @@ export class CommonWorkerSession { async treePayload(node: string) { const payload = await this.ensureApi().treePayload(nodeIdToBytes16(node)); - return payload === null ? null : Array.from(payload); + return rpcBinaryResult(payload); } async treeNodeCount() { @@ -127,7 +127,7 @@ export class CommonWorkerSession { async treeParent(node: string) { const result = await this.ensureApi().treeParent(nodeIdToBytes16(node)); - return result === null ? null : Array.from(result); + return rpcBinaryResult(result); } async treeExists(node: string) { diff --git a/packages/treecrdt-wa-sqlite/src/rpc.ts b/packages/treecrdt-wa-sqlite/src/rpc.ts index 1ae2aa5c..0a3754a5 100644 --- a/packages/treecrdt-wa-sqlite/src/rpc.ts +++ b/packages/treecrdt-wa-sqlite/src/rpc.ts @@ -39,9 +39,9 @@ export type RpcSchema = { }; treeDump: { params: []; result: unknown[] }; treeNodeCount: { params: []; result: number }; - treeParent: { params: [node: string]; result: number[] | null }; + treeParent: { params: [node: string]; result: Uint8Array | null }; treeExists: { params: [node: string]; result: boolean }; - treePayload: { params: [node: string]; result: number[] | null }; + treePayload: { params: [node: string]; result: Uint8Array | null }; headLamport: { params: []; result: number }; replicaMaxCounter: { params: [replica: number[]]; result: number }; close: { params: []; result: void }; @@ -66,3 +66,29 @@ export type RpcPushMessage = { type: 'materialized'; event: MaterializationEvent; }; + +export function rpcBinaryResult(bytes: Uint8Array | null): Uint8Array | null { + if (bytes === null) return null; + const buffer = bytes.buffer; + if ( + buffer instanceof ArrayBuffer && + bytes.byteOffset === 0 && + bytes.byteLength === buffer.byteLength + ) { + return bytes; + } + return Uint8Array.from(bytes); +} + +export function transferablesForRpcBinaryResult(result: unknown): Transferable[] { + if (!(result instanceof Uint8Array)) return []; + const buffer = result.buffer; + if ( + buffer instanceof ArrayBuffer && + result.byteOffset === 0 && + result.byteLength === buffer.byteLength + ) { + return [buffer]; + } + return []; +} diff --git a/packages/treecrdt-wa-sqlite/src/shared-worker.ts b/packages/treecrdt-wa-sqlite/src/shared-worker.ts index c9c69c78..05800351 100644 --- a/packages/treecrdt-wa-sqlite/src/shared-worker.ts +++ b/packages/treecrdt-wa-sqlite/src/shared-worker.ts @@ -1,6 +1,13 @@ /// import type { MaterializationEvent } from '@treecrdt/interface/engine'; -import type { RpcInitResult, RpcMethod, RpcParams, RpcRequest, RpcResult } from './rpc.js'; +import { + transferablesForRpcBinaryResult, + type RpcInitResult, + type RpcMethod, + type RpcParams, + type RpcRequest, + type RpcResult, +} from './rpc.js'; import { openTreecrdtDb } from './open.js'; import { CommonWorkerSession, @@ -54,14 +61,21 @@ function broadcastMaterialized(event: MaterializationEvent, exclude?: MessagePor ports.add(port); port.onmessage = (message: MessageEvent) => { const request = message.data; - const respond = (ok: boolean, result?: any, error?: string) => { - port.postMessage({ id: request.id, ok, result, error }); + const respondSuccess = (result?: unknown) => { + const transfer = + request.method === 'treePayload' || request.method === 'treeParent' + ? transferablesForRpcBinaryResult(result) + : []; + port.postMessage({ id: request.id, ok: true, result }, transfer); + }; + const respondError = (error: string) => { + port.postMessage({ id: request.id, ok: false, error }); }; const run = callQueue.then(() => handleRequest(port, request)); callQueue = settleQueue(run); run.then( - (result) => respond(true, result), - (err) => respond(false, null, err instanceof Error ? err.message : String(err)), + (result) => respondSuccess(result), + (err) => respondError(err instanceof Error ? err.message : String(err)), ); }; port.start(); diff --git a/packages/treecrdt-wa-sqlite/src/worker.ts b/packages/treecrdt-wa-sqlite/src/worker.ts index f232dd68..ec5306dd 100644 --- a/packages/treecrdt-wa-sqlite/src/worker.ts +++ b/packages/treecrdt-wa-sqlite/src/worker.ts @@ -1,6 +1,6 @@ /// import type { MaterializationEvent } from '@treecrdt/interface/engine'; -import type { RpcMethod, RpcRequest } from './rpc.js'; +import { transferablesForRpcBinaryResult, type RpcMethod, type RpcRequest } from './rpc.js'; import { openTreecrdtDb } from './open.js'; import { CommonWorkerSession, @@ -48,19 +48,26 @@ const methods = { self.onmessage = async (ev: MessageEvent) => { const { id, method, params } = ev.data; - const respond = (ok: boolean, result?: any, error?: string) => { - (self as unknown as Worker).postMessage({ id, ok, result, error }); + const respondSuccess = (result?: unknown) => { + const transfer = + method === 'treePayload' || method === 'treeParent' + ? transferablesForRpcBinaryResult(result) + : []; + (self as unknown as Worker).postMessage({ id, ok: true, result }, transfer); + }; + const respondError = (error: string) => { + (self as unknown as Worker).postMessage({ id, ok: false, error }); }; try { const methodFn = (methods as Record Promise>)[method]; if (!methodFn) { - respond(false, null, `unknown method: ${method}`); + respondError(`unknown method: ${method}`); return; } const result = await methodFn(...(params ?? [])); - respond(true, result); + respondSuccess(result); } catch (err) { - respond(false, null, err instanceof Error ? err.message : String(err)); + respondError(err instanceof Error ? err.message : String(err)); } };