diff --git a/.changeset/lean-mailboxes-drain.md b/.changeset/lean-mailboxes-drain.md new file mode 100644 index 0000000..50c9f12 --- /dev/null +++ b/.changeset/lean-mailboxes-drain.md @@ -0,0 +1,8 @@ +--- +"@typeonce/effect-machine": patch +--- + +Use a compact FIFO mailbox for on-demand compiled statecharts while retaining +Effect Queue for persistent custom process logic. This reduces idle heap for +machines and invoked families while preserving FIFO delivery, terminal send +rejection, and wake-up behavior. diff --git a/src/internal/machineProcess.ts b/src/internal/machineProcess.ts index eb74b7d..7e24336 100644 --- a/src/internal/machineProcess.ts +++ b/src/internal/machineProcess.ts @@ -153,7 +153,8 @@ const makeProcessLogic: < drain: (context: internalRuntime.ProcessContext, Machine.EventOf>) => internalRuntime.provideMachineRuntime( Effect.gen(function*() { - const { mailbox, state, setState } = context + const poll = context.poll ?? Queue.poll(context.mailbox!) + const { state, setState } = context let current = yield* state if (internalPlanner.isFinalState(machine, current)) { return Option.some( @@ -380,7 +381,7 @@ const makeProcessLogic: < } while (true) { - const pending = yield* Queue.poll(mailbox) + const pending = yield* poll if (Option.isNone(pending)) { return Option.none() } @@ -447,7 +448,8 @@ const makeProcessLogic: < run: (context) => internalRuntime.provideMachineRuntime( Effect.gen(function*() { - const { mailbox, receive, state, setState } = context + const poll = context.poll ?? Queue.poll(context.mailbox!) + const { receive, state, setState } = context let terminal: { readonly output: Output } | undefined let current = yield* state @@ -506,7 +508,7 @@ const makeProcessLogic: < } if (terminal === undefined) { - pendingEvent = yield* (pollEvent ??= Queue.poll(mailbox)) + pendingEvent = yield* (pollEvent ??= poll) if (Option.isNone(pendingEvent)) { configuration = undefined pollEvent = undefined @@ -785,7 +787,7 @@ const makeProcessLogic: < } if (terminal === undefined) { - pendingEvent = yield* (pollEvent ??= Queue.poll(mailbox)) + pendingEvent = yield* (pollEvent ??= poll) if (Option.isNone(pendingEvent)) { configuration = undefined pollEvent = undefined diff --git a/src/internal/machineRuntime.ts b/src/internal/machineRuntime.ts index 9e97b01..3f98f49 100644 --- a/src/internal/machineRuntime.ts +++ b/src/internal/machineRuntime.ts @@ -157,7 +157,10 @@ export interface ProcessScope { export interface ProcessContext extends ProcessScope { readonly receive: Effect.Effect - readonly mailbox: Queue.Dequeue + /** @internal */ + readonly mailbox?: Queue.Dequeue + /** @internal */ + readonly poll?: Effect.Effect> readonly state: Effect.Effect readonly setState: (state: State) => Effect.Effect readonly updateState: ( @@ -165,6 +168,37 @@ export interface ProcessContext extends ProcessScope { ) => Effect.Effect } +interface CompactProcessMailbox { + items: Array | undefined + index: number + closed: boolean +} + +const offerCompactMailbox = (mailbox: CompactProcessMailbox, event: Event): void => { + const items = mailbox.items ?? [] + mailbox.items = items + items.push(event) +} + +const pollCompactMailbox = (mailbox: CompactProcessMailbox): Option.Option => { + if (mailbox.items === undefined) { + return Option.none() + } + const event = mailbox.items[mailbox.index]! + mailbox.index += 1 + if (mailbox.index === mailbox.items.length) { + mailbox.items = undefined + mailbox.index = 0 + } + return Option.some(event) +} + +const closeCompactMailbox = (mailbox: CompactProcessMailbox): void => { + mailbox.closed = true + mailbox.items = undefined + mailbox.index = 0 +} + export interface ProcessLogic< State, Event, @@ -1145,11 +1179,24 @@ const startCompiledInternal: < const sessionId = yield* options.runtime.nextSessionId const id = options.id ?? sessionId - const queue = yield* Queue.unbounded() + const onDemand = logic.drain !== undefined + const queue = onDemand ? undefined : yield* Queue.unbounded() + // An on-demand drain never blocks on mailbox input: send schedules its owner + // whenever the FIFO becomes non-empty. Keep persistent custom processes on + // Queue, but avoid retaining Queue's waiting/backpressure machinery for the + // compiled protocol that only needs synchronous offer and poll operations. + const compactMailbox: CompactProcessMailbox | undefined = onDemand + ? { items: undefined, index: 0, closed: false } + : undefined + const poll = compactMailbox === undefined + ? Queue.poll(queue!) + : Effect.sync(() => pollCompactMailbox(compactMailbox)) + const shutdownMailbox = compactMailbox === undefined + ? Queue.shutdown(queue!) + : Effect.sync(() => closeCompactMailbox(compactMailbox)) const termination = yield* Deferred.make() const done = yield* Deferred.make() const awaitCompletion = Deferred.await(done).pipe(Effect.exit, Effect.asVoid) - const onDemand = logic.drain !== undefined const drainServices = onDemand ? yield* Effect.context() : undefined let worker: Fiber.Fiber | undefined let draining = false @@ -1158,10 +1205,12 @@ const startCompiledInternal: < let terminationRequested = false let requestRuntimeTermination = (requested: ProcessTermination): Effect.Effect => Deferred.succeed(termination, requested) - let sendEvent = (event: Event): Effect.Effect => - Queue.offer(queue, event).pipe( - Effect.flatMap((accepted) => accepted ? Effect.void : Effect.fail(new StoppedError())) - ) + let sendEvent = compactMailbox !== undefined + ? (event: Event): Effect.Effect => Effect.sync(() => offerCompactMailbox(compactMailbox, event)) + : (event: Event): Effect.Effect => + Queue.offer(queue!, event).pipe( + Effect.flatMap((accepted) => accepted ? Effect.void : Effect.fail(new StoppedError())) + ) let settleRequestedTermination = (_requested: ProcessTermination): Effect.Effect => Effect.void const interruptWorker: Effect.Effect = Effect.suspend(() => worker === undefined @@ -1410,7 +1459,7 @@ const startCompiledInternal: < Effect.asVoid ) return Effect.uninterruptible( - Queue.shutdown(queue).pipe( + shutdownMailbox.pipe( Effect.andThen(closeChildren(exit)), Effect.andThen(setAndPublishSnapshot(snapshot)), Effect.andThen(notifyOutcome), @@ -1536,8 +1585,8 @@ const startCompiledInternal: < const context: ProcessContext = { ...scope, - receive: Queue.take(queue), - mailbox: queue, + receive: queue === undefined ? Effect.never : Queue.take(queue), + poll, state: getCurrent.pipe(Effect.map((current) => current.snapshot.state)), setState: setActiveState, updateState: (f) => @@ -1683,13 +1732,11 @@ const startCompiledInternal: < sendEvent = (event) => Effect.uninterruptible( Effect.suspend(() => { - if (!Queue.offerUnsafe(queue, event)) { + if (compactMailbox!.closed || terminationRequested) { return Effect.fail(new StoppedError()) } + offerCompactMailbox(compactMailbox!, event) offerRevision += 1 - if (terminationRequested) { - return Effect.fail(new StoppedError()) - } if (draining) { return Effect.void } diff --git a/test/MachineProcessLifecycle.test.ts b/test/MachineProcessLifecycle.test.ts index 49248de..feb0921 100644 --- a/test/MachineProcessLifecycle.test.ts +++ b/test/MachineProcessLifecycle.test.ts @@ -1,9 +1,44 @@ import { assert, describe, it } from "@effect/vitest" -import { Cause, Deferred, Effect, Exit, Fiber, Option, Queue, Ref, Stream } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, Option, Ref, Stream } from "effect" import { Machine } from "../src/index.js" import * as MachineRuntime from "../src/internal/machineRuntime.js" describe("machine process lifecycle", () => { + it.effect("preserves FIFO order while draining a compact compiled mailbox", () => + Effect.gen(function*() { + type Event = + | { readonly _tag: "Value"; readonly value: number } + | { readonly _tag: "Done" } + const observed: Array = [] + const ref = yield* MachineRuntime.startProcess>({ + [MachineRuntime.childlessProcess]: true, + [MachineRuntime.compiledProcess]: true, + initial: () => Effect.succeed(undefined), + run: () => Effect.never, + drain: (context) => + Effect.gen(function*() { + while (true) { + const event = yield* context.poll! + if (Option.isNone(event)) { + return Option.none() + } + if (event.value._tag === "Done") { + return Option.some(observed) + } + observed.push(event.value.value) + } + }) + }) + + for (let value = 0; value < 1_000; value += 1) { + yield* ref.send({ _tag: "Value", value }) + } + yield* ref.send({ _tag: "Done" }) + + assert.deepStrictEqual(yield* ref.join, Array.from({ length: 1_000 }, (_, value) => value)) + assert.instanceOf(yield* Effect.flip(ref.send({ _tag: "Done" })), Machine.StoppedError) + })) + it.effect("wakes an on-demand compiled process across consecutive idle periods", () => Effect.gen(function*() { type Event = { @@ -18,7 +53,7 @@ describe("machine process lifecycle", () => { drain: (context) => Effect.gen(function*() { while (true) { - const event = yield* Queue.poll(context.mailbox) + const event = yield* context.poll! if (Option.isNone(event)) { return Option.none() } @@ -296,7 +331,7 @@ describe("machine process lifecycle", () => { run: () => Effect.never, drain: (context) => Effect.gen(function*() { - const event = yield* Queue.poll(context.mailbox) + const event = yield* context.poll! if (Option.isNone(event)) { return Option.none() }