From b459a598aed124a981d0f3f04d97ff5aaaa38921 Mon Sep 17 00:00:00 2001 From: kyscott18 <43524469+kyscott18@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:27:15 +0000 Subject: [PATCH 1/2] perf(runtime): wake the FIFO batch loop when a mutation arrives FIFO mode polled the received-mutation queue every 4ms, so each mutation waited for the next tick. That wait was most of its latency, and the loop woke 250 times per second when idle. FIFO now blocks on Queue.takeAll and accepts everything that queued during the previous batch. Batch mode keeps its fixed window for batchOrder. Both modes take mutations outside the speculative-state lock. Counter benchmark (sequential FIFO): 4.04ms -> 1.56ms average, 247 -> 641 mutations/s. --- packages/typewriter/src/runtime.ts | 201 ++++++++++++++++------------- 1 file changed, 108 insertions(+), 93 deletions(-) diff --git a/packages/typewriter/src/runtime.ts b/packages/typewriter/src/runtime.ts index c468739..bc3e614 100644 --- a/packages/typewriter/src/runtime.ts +++ b/packages/typewriter/src/runtime.ts @@ -76,7 +76,6 @@ import type { import type { LocalLog } from "./watch"; import { Watch } from "./watch"; -const FIFO_BATCH_INTERVAL_MS = 4; const SUBMIT_RETRY_TIMES = 5; type RuntimeExecuteInput = { @@ -636,99 +635,106 @@ export function createRuntimeEffect( }); } - const acceptBatch = withSpeculativeStateLock( - Effect.gen(function* () { - const batchStartedAtMs = startTimer(); - const submittedMutations = yield* Queue.clear(receivedMutationQueue); + type ReceivedMutationEntry = { + mutationId: number; + deferred: Deferred.Deferred; + }; - if (submittedMutations.length === 0) { - return; - } + const acceptBatch = (submittedMutations: ReceivedMutationEntry[]) => + withSpeculativeStateLock( + Effect.gen(function* () { + const batchStartedAtMs = startTimer(); - if (batchOrder.length > 0) { - submittedMutations.sort((a, b) => { - const aName = (mutationsById.get(a.mutationId)! as ReceivedMutation) - .name; - const bName = (mutationsById.get(b.mutationId)! as ReceivedMutation) - .name; - return ( - (batchOrderByName.get(aName) ?? batchOrder.length) - - (batchOrderByName.get(bName) ?? batchOrder.length) - ); - }); - } + if (submittedMutations.length === 0) { + return; + } - const acceptedMutations: { - mutation: AcceptedMutation; - executeResult: ExecuteResult; - preimages: readonly Hex.Hex[]; - deferred: Deferred.Deferred; - }[] = []; - - for (const { mutationId, deferred } of submittedMutations) { - const mutation = mutationsById.get(mutationId)! as ReceivedMutation; - const result = yield* Effect.result(acceptMutation(mutation)); - if (Result.isSuccess(result)) { - acceptedMutations.push({ ...result.success, deferred }); - } else { - yield* Deferred.fail(deferred, result.failure); + if (batchOrder.length > 0) { + submittedMutations.sort((a, b) => { + const aName = ( + mutationsById.get(a.mutationId)! as ReceivedMutation + ).name; + const bName = ( + mutationsById.get(b.mutationId)! as ReceivedMutation + ).name; + return ( + (batchOrderByName.get(aName) ?? batchOrder.length) - + (batchOrderByName.get(bName) ?? batchOrder.length) + ); + }); } - } - if (acceptedMutations.length === 0) { - return; - } + const acceptedMutations: { + mutation: AcceptedMutation; + executeResult: ExecuteResult; + preimages: readonly Hex.Hex[]; + deferred: Deferred.Deferred; + }[] = []; - const batch: RuntimeBatch = { - status: "accepted", - id: batchId++, - position: batchPosition++, - mutations: acceptedMutations.map(({ mutation }) => mutation), - }; + for (const { mutationId, deferred } of submittedMutations) { + const mutation = mutationsById.get(mutationId)! as ReceivedMutation; + const result = yield* Effect.result(acceptMutation(mutation)); + if (Result.isSuccess(result)) { + acceptedMutations.push({ ...result.success, deferred }); + } else { + yield* Deferred.fail(deferred, result.failure); + } + } - yield* Effect.logDebug("batched mutations").pipe( - Effect.annotateLogs({ - id: batch.id, - mutationCount: batch.mutations.length, - duration: durationMs(batchStartedAtMs), - }), - ); + if (acceptedMutations.length === 0) { + return; + } - batchesById.set(batch.id, batch); + const batch: RuntimeBatch = { + status: "accepted", + id: batchId++, + position: batchPosition++, + mutations: acceptedMutations.map(({ mutation }) => mutation), + }; - emitBatch(batchToEvent(batch)); + yield* Effect.logDebug("batched mutations").pipe( + Effect.annotateLogs({ + id: batch.id, + mutationCount: batch.mutations.length, + duration: durationMs(batchStartedAtMs), + }), + ); - const persistStartedAtMs = startTimer(); - yield* db.transaction((tx) => - Effect.gen(function* () { - yield* insertMutations( - tx, - schema, - acceptedMutations.map((entry) => entry.mutation), - ); - yield* insertSlotWritesMany(tx, schema, acceptedMutations); - yield* insertKeccakPreimages( - tx, - schema, - acceptedMutations.flatMap(({ preimages }) => preimages), - ); - }), - ); + batchesById.set(batch.id, batch); - for (const { mutation, deferred } of acceptedMutations) { - yield* Deferred.succeed(deferred, mutation); - } + emitBatch(batchToEvent(batch)); - yield* Effect.logDebug("persisted accepted mutations").pipe( - Effect.annotateLogs({ - mutationCount: acceptedMutations.length, - duration: durationMs(persistStartedAtMs), - }), - ); + const persistStartedAtMs = startTimer(); + yield* db.transaction((tx) => + Effect.gen(function* () { + yield* insertMutations( + tx, + schema, + acceptedMutations.map((entry) => entry.mutation), + ); + yield* insertSlotWritesMany(tx, schema, acceptedMutations); + yield* insertKeccakPreimages( + tx, + schema, + acceptedMutations.flatMap(({ preimages }) => preimages), + ); + }), + ); - yield* Queue.offer(acceptedBatchesQueue, batch.id); - }), - ); + for (const { mutation, deferred } of acceptedMutations) { + yield* Deferred.succeed(deferred, mutation); + } + + yield* Effect.logDebug("persisted accepted mutations").pipe( + Effect.annotateLogs({ + mutationCount: acceptedMutations.length, + duration: durationMs(persistStartedAtMs), + }), + ); + + yield* Queue.offer(acceptedBatchesQueue, batch.id); + }), + ); const submit = Effect.gen(function* () { const simulation = yield* withSpeculativeStateLock( @@ -1086,20 +1092,29 @@ export function createRuntimeEffect( ); }).pipe(Effect.withLogSpan("watch")); - const batchIntervalMs = - app.sequencing.order === "batch" - ? app.sequencing.batchIntervalMs - : FIFO_BATCH_INTERVAL_MS; const submitIntervalMs = app.sequencing.submitIntervalMs; - const batchProgram = Effect.sleep(Duration.millis(batchIntervalMs)).pipe( - Effect.andThen( - Effect.repeat( - acceptBatch, - Schedule.fixed(Duration.millis(batchIntervalMs)), - ), - ), - ); + // FIFO wakes as soon as a mutation arrives and accepts everything that + // queued while the previous batch ran. Batch mode collects for a fixed + // window so it can reorder by `batchOrder`. Both take mutations outside + // the state lock, so a wait never blocks submit. + const batchProgram = + app.sequencing.order === "batch" + ? Effect.sleep(Duration.millis(app.sequencing.batchIntervalMs)).pipe( + Effect.andThen( + Effect.repeat( + Queue.clear(receivedMutationQueue).pipe( + Effect.flatMap(acceptBatch), + ), + Schedule.fixed(Duration.millis(app.sequencing.batchIntervalMs)), + ), + ), + ) + : Effect.forever( + Queue.takeAll(receivedMutationQueue).pipe( + Effect.flatMap(acceptBatch), + ), + ); const submitProgram = Effect.sleep(Duration.millis(submitIntervalMs)).pipe( Effect.andThen( Effect.repeat( From 5bbc65a77d07db048819503a11cfaf2d4ddd4b11 Mon Sep 17 00:00:00 2001 From: kyscott18 <43524469+kyscott18@users.noreply.github.com> Date: Tue, 13 Oct 2026 20:39:08 +0000 Subject: [PATCH 2/2] refactor(runtime): inline the accept batch entry type --- packages/typewriter/src/runtime.ts | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/packages/typewriter/src/runtime.ts b/packages/typewriter/src/runtime.ts index bc3e614..713057e 100644 --- a/packages/typewriter/src/runtime.ts +++ b/packages/typewriter/src/runtime.ts @@ -635,12 +635,12 @@ export function createRuntimeEffect( }); } - type ReceivedMutationEntry = { - mutationId: number; - deferred: Deferred.Deferred; - }; - - const acceptBatch = (submittedMutations: ReceivedMutationEntry[]) => + const acceptBatch = ( + submittedMutations: { + mutationId: number; + deferred: Deferred.Deferred; + }[], + ) => withSpeculativeStateLock( Effect.gen(function* () { const batchStartedAtMs = startTimer(); @@ -1094,10 +1094,6 @@ export function createRuntimeEffect( const submitIntervalMs = app.sequencing.submitIntervalMs; - // FIFO wakes as soon as a mutation arrives and accepts everything that - // queued while the previous batch ran. Batch mode collects for a fixed - // window so it can reorder by `batchOrder`. Both take mutations outside - // the state lock, so a wait never blocks submit. const batchProgram = app.sequencing.order === "batch" ? Effect.sleep(Duration.millis(app.sequencing.batchIntervalMs)).pipe(