Compact child lifecycle (#58)

Author: SandroMaglione

.changeset/fast-children-close.md

@@ -0,0 +1,5 @@
+---
+"@typeonce/effect-machine": patch
+---
+
+Reduce child lifecycle overhead by reserving child starts atomically and specializing zero- and one-item invoke cleanup without weakening parallel finalization.

src/internal/machineProcess.ts

@@ -56,6 +56,24 @@ const getInvokes = (
   return Array.isArray(invokes) ? invokes as ReadonlyArray<AnyInvokeConfig> : [invokes as AnyInvokeConfig]
 }
 
+const runSequentialDiscard = <E, R>(
+  effects: ReadonlyArray<Effect.Effect<void, E, R>>
+): Effect.Effect<void, E, R> =>
+  effects.length === 0
+    ? Effect.void
+    : effects.length === 1
+    ? effects[0]!
+    : Effect.all(effects, { discard: true })
+
+const runParallelDiscard = <E, R>(
+  effects: ReadonlyArray<Effect.Effect<void, E, R>>
+): Effect.Effect<void, E, R> =>
+  effects.length === 0
+    ? Effect.void
+    : effects.length === 1
+    ? effects[0]!
+    : Effect.all(effects, { discard: true, concurrency: "unbounded" })
+
 const invokeCapabilityCache = new WeakMap<Machine.Any, boolean>()
 
 const hasInvokeCapability = (machine: Machine.Any): boolean => {
@@ -282,18 +300,11 @@ const makeProcessLogic: <
                 )
               )
             const stopInvoke = (key: string): Effect.Effect<void> => removeInvoke(key, undefined)
-            stopAllInvokes = Effect.sync(() => {
-              const sessions = Array.from(invokeSessions.values())
+            stopAllInvokes = Effect.suspend(() => {
+              const effects = Array.from(invokeSessions.values(), stopInvokeSession)
               invokeSessions.clear()
-              return sessions
-            }).pipe(
-              Effect.flatMap((sessions) =>
-                Effect.all(
-                  sessions.map((session) => stopInvokeSession(session)),
-                  { discard: true, concurrency: "unbounded" }
-                )
-              )
-            )
+              return runParallelDiscard(effects)
+            })
             const startInvoke = Effect.fnUntraced(function*<StateId extends Machine.StateIdentifier<States>>(
               path: StateId,
               config: AnyInvokeConfig
@@ -347,12 +358,12 @@ const makeProcessLogic: <
                 )
               )
             })
-            startInvokes = Effect.fnUntraced(function*(
+            startInvokes = (
               configuration: Model.ActiveConfiguration,
               paths: ReadonlyArray<string>,
               event: Machine.LifecycleEvent<Events>
-            ) {
-              yield* Effect.all(
+            ) =>
+              runSequentialDiscard(
                 internalPlanner.sortEntryPaths(machine, paths)
                   .filter((path) => configuration.active.has(path))
                   .flatMap((path) =>
@@ -367,22 +378,15 @@ const makeProcessLogic: <
                         config
                       ) as Effect.Effect<void, E | MachineSchemaDecodeError, R>
                     )
-                  ),
-                { discard: true }
+                  )
               )
-            })
             stopInvokes = (paths) =>
-              Effect.sync(() =>
-                internalPlanner.sortExitPaths(machine, paths).flatMap((path) =>
-                  Array.from(invokeSessions.entries())
-                    .filter(([, session]) => session.path === path)
-                    .map(([key]) => key)
-                )
-              ).pipe(
-                Effect.flatMap((keys) =>
-                  Effect.all(
-                    keys.map(stopInvoke),
-                    { discard: true, concurrency: "unbounded" }
+              Effect.suspend(() =>
+                runParallelDiscard(
+                  internalPlanner.sortExitPaths(machine, paths).flatMap((path) =>
+                    Array.from(invokeSessions.entries())
+                      .filter(([, session]) => session.path === path)
+                      .map(([key]) => stopInvoke(key))
                   )
                 )
               )
@@ -569,18 +573,11 @@ const makeProcessLogic: <
               )
             )
           const stopInvoke = (key: string): Effect.Effect<void> => removeInvoke(key, undefined)
-          const stopAllInvokes: Effect.Effect<void> = Effect.sync(() => {
-            const sessions = Array.from(invokeSessions.values())
+          const stopAllInvokes: Effect.Effect<void> = Effect.suspend(() => {
+            const effects = Array.from(invokeSessions.values(), stopInvokeSession)
             invokeSessions.clear()
-            return sessions
-          }).pipe(
-            Effect.flatMap((sessions) =>
-              Effect.all(
-                sessions.map((session) => stopInvokeSession(session)),
-                { discard: true, concurrency: "unbounded" }
-              )
-            )
-          )
+            return runParallelDiscard(effects)
+          })
           const startInvoke = Effect.fnUntraced(function*<StateId extends Machine.StateIdentifier<States>>(
             path: StateId,
             config: AnyInvokeConfig
@@ -643,7 +640,7 @@ const makeProcessLogic: <
             paths: ReadonlyArray<string>,
             event: Machine.LifecycleEvent<Events>
           ) {
-            yield* Effect.all(
+            yield* runSequentialDiscard(
               internalPlanner.sortEntryPaths(machine, paths)
                 .filter((path) => configuration.active.has(path))
                 .flatMap((path) =>
@@ -658,22 +655,16 @@ const makeProcessLogic: <
                       config
                     ) as Effect.Effect<void, E | MachineSchemaDecodeError, R>
                   )
-                ),
-              { discard: true }
+                )
             )
           })
           const stopInvokes = (paths: ReadonlyArray<string>): Effect.Effect<void> =>
-            Effect.sync(() =>
-              internalPlanner.sortExitPaths(machine, paths).flatMap((path) =>
-                Array.from(invokeSessions.entries())
-                  .filter(([, session]) => session.path === path)
-                  .map(([key]) => key)
-              )
-            ).pipe(
-              Effect.flatMap((keys) =>
-                Effect.all(
-                  keys.map(stopInvoke),
-                  { discard: true, concurrency: "unbounded" }
+            Effect.suspend(() =>
+              runParallelDiscard(
+                internalPlanner.sortExitPaths(machine, paths).flatMap((path) =>
+                  Array.from(invokeSessions.entries())
+                    .filter(([, session]) => session.path === path)
+                    .map(([key]) => stopInvoke(key))
                 )
               )
             )

src/internal/machineRuntime.ts

@@ -469,56 +469,43 @@ const makeChildRuntime = (
       }
     }
 
-    const close = <A, E>(_exit: Exit.Exit<A, E>): Effect.Effect<void> =>
-      Effect.sync(() => {
+    const close = <A, E>(exit: Exit.Exit<A, E>): Effect.Effect<void> =>
+      Effect.suspend(() => {
         if (registry.closed) {
-          return undefined
+          return Effect.void
         }
         registry.closed = true
         if (registry.scope === undefined) {
-          return undefined
+          return Effect.void
         }
-        const children = Array.from(registry.children.values()).flatMap((entry) =>
-          entry._tag === "Started" ? [entry.ref] : []
-        )
-        return { children, scope: registry.scope }
-      }).pipe(
-        Effect.flatMap((resources) =>
-          resources === undefined
-            ? Effect.void
-            : Effect.all(
-              [
-                ...resources.children.map((child) => child.stop),
-                ...(resources.scope === undefined ? [] : [Scope.close(resources.scope, _exit)])
-              ],
-              { concurrency: "unbounded", discard: true }
-            )
-        )
-      )
-
-    const getOrCreateScope: Effect.Effect<Scope.Closeable | undefined> = Effect.sync(() => {
-      if (registry.closed) {
-        return undefined
-      }
-      if (registry.scope === undefined) {
-        registry.scope = Scope.makeUnsafe("parallel")
-      }
-      return registry.scope
-    })
-
-    const reserve = (
-      key: ChildKey,
-      token: symbol
-    ): Effect.Effect<boolean, ChildAlreadyExistsError> =>
-      Effect.suspend(() => {
-        if (registry.closed) {
-          return Effect.succeed(false)
+        const finalizers = Scope.closeUnsafe(registry.scope, exit)
+        let first: Effect.Effect<void> | undefined
+        let rest: Array<Effect.Effect<void>> | undefined
+        for (const entry of registry.children.values()) {
+          if (entry._tag !== "Started") {
+            continue
+          }
+          if (first === undefined) {
+            first = entry.ref.stop
+          } else {
+            rest ??= [first]
+            rest.push(entry.ref.stop)
+          }
         }
-        if (typeof key === "string" && registry.children.has(key)) {
-          return Effect.fail(new ChildAlreadyExistsError({ id: key }))
+        if (finalizers !== undefined) {
+          if (first === undefined) {
+            first = finalizers
+          } else {
+            rest ??= [first]
+            rest.push(finalizers)
+          }
         }
-        registry.children.set(key, { _tag: "Starting", token })
-        return Effect.succeed(true)
+        const cleanup = rest ?? first
+        return cleanup === undefined
+          ? Effect.void
+          : Array.isArray(cleanup)
+          ? Effect.all(cleanup, { concurrency: "unbounded", discard: true })
+          : cleanup
       })
 
     const unregister = (
@@ -681,47 +668,50 @@ const makeChildRuntime = (
       const token = Symbol()
       const key = spawnOptions?.id ?? token
       let startedChild: MachineRef<any, any, any, any> | undefined
-      return getOrCreateScope.pipe(
-        Effect.flatMap((childScope) =>
-          childScope === undefined
-            ? Effect.interrupt
-            : Effect.gen(function*() {
-              const reserved = yield* reserve(key, token)
-              if (!reserved) {
-                return yield* Effect.interrupt
-              }
-              return yield* startLogicInternal(logic, {
-                detached: true,
-                ...(spawnOptions?.id === undefined ? undefined : { id: spawnOptions.id }),
-                ...(spawnOptions?.onOutcome === undefined ? undefined : { onOutcome: spawnOptions.onOutcome }),
-                ...(spawnOptions?.[activeSnapshotObserver] === undefined
-                  ? undefined
-                  : { onSnapshot: spawnOptions[activeSnapshotObserver] }),
-                ...(spawnOptions?.[sendParentOverride] === undefined
-                  ? undefined
-                  : { sendParent: spawnOptions[sendParentOverride] }),
-                onReady: (child, requestChildStop) =>
-                  Effect.sync(() => {
-                    startedChild = child
-                  }).pipe(
-                    Effect.andThen(register(key, token, child, spawnOptions?.descriptor)),
-                    Effect.flatMap((registered) => registered ? Effect.void : requestChildStop)
-                  ),
-                onStop: unregister(key, token),
-                parent: self,
-                runtime
-              }).pipe(
-                Effect.onExit((exit) =>
-                  Exit.isFailure(exit)
-                    ? unregister(key, token).pipe(
-                      Effect.andThen(startedChild === undefined ? Effect.void : startedChild.stop)
-                    )
-                    : Effect.void
-                )
+      return Effect.suspend((): Effect.Effect<
+        MachineRef<ChildState, ChildEvent, ChildError, ChildOutput>,
+        ChildAlreadyExistsError | ChildInitialError,
+        Exclude<ChildRequirements, Scope.Scope>
+      > => {
+        if (registry.closed) {
+          return Effect.interrupt
+        }
+        if (typeof key === "string" && registry.children.has(key)) {
+          return Effect.fail(new ChildAlreadyExistsError({ id: key }))
+        }
+        registry.scope ??= Scope.makeUnsafe("parallel")
+        registry.children.set(key, { _tag: "Starting", token })
+        return startLogicInternal(logic, {
+          detached: true,
+          ...(spawnOptions?.id === undefined ? undefined : { id: spawnOptions.id }),
+          ...(spawnOptions?.onOutcome === undefined ? undefined : { onOutcome: spawnOptions.onOutcome }),
+          ...(spawnOptions?.[activeSnapshotObserver] === undefined
+            ? undefined
+            : { onSnapshot: spawnOptions[activeSnapshotObserver] }),
+          ...(spawnOptions?.[sendParentOverride] === undefined
+            ? undefined
+            : { sendParent: spawnOptions[sendParentOverride] }),
+          onReady: (child, requestChildStop) =>
+            Effect.sync(() => {
+              startedChild = child
+            }).pipe(
+              Effect.andThen(register(key, token, child, spawnOptions?.descriptor)),
+              Effect.flatMap((registered) => registered ? Effect.void : requestChildStop)
+            ),
+          onStop: unregister(key, token),
+          parent: self,
+          runtime
+        }).pipe(
+          Effect.onExit((exit) =>
+            Exit.isFailure(exit)
+              ? unregister(key, token).pipe(
+                Effect.andThen(startedChild === undefined ? Effect.void : startedChild.stop)
               )
-            }).pipe(Scope.provide(childScope))
+              : Effect.void
+          ),
+          Scope.provide(registry.scope)
         )
-      ) as Effect.Effect<
+      }) as Effect.Effect<
         MachineRef<ChildState, ChildEvent, ChildError, ChildOutput>,
         ChildAlreadyExistsError | ChildInitialError,
         Exclude<ChildRequirements, Scope.Scope>

test/MachineProcessLifecycle.test.ts

@@ -647,6 +647,47 @@ describe("machine process lifecycle", () => {
       assert.deepStrictEqual(yield* children.anonymous.snapshot, { status: "stopped", state: 0 })
     }))
 
+  it.effect("keeps child interruption parallel with scoped resource finalization", () =>
+    Effect.gen(function*() {
+      const parentScope = yield* Deferred.make<MachineRuntime.ProcessScope<never>>()
+      const childInterrupted = yield* Deferred.make<void>()
+      const resourceFinalized = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      const parent = yield* MachineRuntime.startProcess({
+        initial: (scope) => Deferred.succeed(parentScope, scope).pipe(Effect.as(undefined)),
+        run: () => Effect.never
+      })
+      const child = yield* (yield* Deferred.await(parentScope)).spawn(
+        Machine.logic({
+          initial: () =>
+            Effect.acquireRelease(
+              Effect.void,
+              () =>
+                Deferred.succeed(resourceFinalized, undefined).pipe(
+                  Effect.andThen(Deferred.await(release))
+                )
+            ).pipe(Effect.as(0)),
+          run: () =>
+            Effect.never.pipe(
+              Effect.ensuring(
+                Deferred.succeed(childInterrupted, undefined).pipe(
+                  Effect.andThen(Deferred.await(release))
+                )
+              )
+            )
+        }),
+        { id: "child" }
+      )
+      const stopping = yield* parent.stop.pipe(Effect.forkChild)
+
+      yield* Deferred.await(childInterrupted)
+      yield* Deferred.await(resourceFinalized)
+      yield* Deferred.succeed(release, undefined)
+      yield* Fiber.join(stopping)
+
+      assert.deepStrictEqual(yield* child.snapshot, { status: "stopped", state: 0 })
+    }))
+
   it.effect("does not orphan a child when parent stop races child initialization", () =>
     Effect.gen(function*() {
       const parentScope = yield* Deferred.make<MachineRuntime.ProcessScope<never>>()