Skip to content

Commit ea1ea8e

Browse files
SandroMaglioneSandroMaglione
authored
Compact invoked child ownership (#56)
1 parent 1e7821a commit ea1ea8e

4 files changed

Lines changed: 266 additions & 179 deletions

File tree

‎.changeset/calm-children-invoke.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@typeonce/effect-machine": patch
3+
---
4+
5+
Reduce the retained memory of invoked machines by running the original child logic with a compact guarded `sendParent` channel instead of allocating a wrapper process for every invocation.

‎src/internal/machineProcess.ts‎

Lines changed: 108 additions & 144 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,78 @@ const hasInvokeCapability = (machine: Machine.Any): boolean => {
7070
return hasInvokes
7171
}
7272

73+
const isCurrentInvoke = (
74+
sessions: Map<string, InvokeSession>,
75+
key: string,
76+
token: symbol
77+
): Effect.Effect<boolean> => Effect.sync(() => sessions.get(key)?.token === token)
78+
79+
const makeInvokeSendParent = (
80+
sessions: Map<string, InvokeSession>,
81+
self: internalRuntime.ProcessScope<any>["self"],
82+
key: string,
83+
token: symbol
84+
): (event: unknown) => Effect.Effect<void, StoppedError> => {
85+
return (event) =>
86+
isCurrentInvoke(sessions, key, token).pipe(
87+
Effect.flatMap((isCurrent) => isCurrent ? self.send(event) : Effect.void)
88+
)
89+
}
90+
91+
const makeInvokeOutcomeHandler = (
92+
sessions: Map<string, InvokeSession>,
93+
self: internalRuntime.ProcessScope<any>["self"],
94+
failCause: internalRuntime.ProcessScope<any>["failCause"],
95+
config: AnyInvokeConfig,
96+
key: string,
97+
token: symbol
98+
): (outcome: internalRuntime.RuntimeOutcome<any, any, any>) => Effect.Effect<void> => {
99+
return (outcome) => {
100+
if (outcome._tag === "Stopped") {
101+
return Effect.void
102+
}
103+
return isCurrentInvoke(sessions, key, token).pipe(
104+
Effect.flatMap((isCurrent) => {
105+
if (!isCurrent) {
106+
return Effect.void
107+
}
108+
if (outcome._tag === "Done") {
109+
const mappedEvent = config.onDone === undefined
110+
? outcome.output
111+
: config.onDone({ id: config.id, output: outcome.output })
112+
return mappedEvent === undefined
113+
? Effect.void
114+
: self.send(mappedEvent).pipe(Effect.catchTag("StoppedError", () => Effect.void))
115+
}
116+
return failCause(outcome.cause)
117+
})
118+
)
119+
}
120+
}
121+
122+
const makeInvokeSnapshotHandler = (
123+
sessions: Map<string, InvokeSession>,
124+
self: internalRuntime.ProcessScope<any>["self"],
125+
config: AnyInvokeConfig,
126+
key: string,
127+
token: symbol
128+
): (
129+
snapshot: Extract<internalRuntime.RuntimeSnapshot<any, any, any>, { readonly status: "active" }>
130+
) => Effect.Effect<void> => {
131+
return (snapshot) =>
132+
isCurrentInvoke(sessions, key, token).pipe(
133+
Effect.flatMap((isCurrent) => {
134+
if (!isCurrent || config.snapshot === undefined) {
135+
return Effect.void
136+
}
137+
const mappedEvent = config.snapshot({ id: config.id, snapshot })
138+
return mappedEvent === undefined
139+
? Effect.void
140+
: self.send(mappedEvent).pipe(Effect.catchTag("StoppedError", () => Effect.void))
141+
})
142+
)
143+
}
144+
73145
const makeProcessLogic: <
74146
const States extends Machine.StateSchemas,
75147
const Events extends ReadonlyArray<Machine.TaggedSchema>,
@@ -189,8 +261,6 @@ const makeProcessLogic: <
189261
const makeInvokeSessionKey = (path: string, id: string): string => `${path.length}:${path}${id}`
190262
const makeInvokeChildId = (path: string, id: string): string =>
191263
`Machine.invoke:${makeInvokeSessionKey(path, id)}`
192-
const isCurrentInvoke = (key: string, token: symbol): Effect.Effect<boolean> =>
193-
Effect.sync(() => invokeSessions.get(key)?.token === token)
194264
const stopInvokeSession = (session: InvokeSession): Effect.Effect<void> =>
195265
context.stopChild(session.childId)
196266
const removeInvoke = (
@@ -224,53 +294,6 @@ const makeProcessLogic: <
224294
)
225295
)
226296
)
227-
const handleInvokeOutcome = (
228-
config: AnyInvokeConfig,
229-
key: string,
230-
token: symbol,
231-
outcome: internalRuntime.RuntimeOutcome<any, any, any>
232-
): Effect.Effect<void> => {
233-
if (outcome._tag === "Stopped") {
234-
return Effect.void
235-
}
236-
return isCurrentInvoke(key, token).pipe(
237-
Effect.flatMap((isCurrent) => {
238-
if (!isCurrent) {
239-
return Effect.void
240-
}
241-
if (outcome._tag === "Done") {
242-
const mappedEvent = config.onDone === undefined
243-
? outcome.output
244-
: config.onDone({ id: config.id, output: outcome.output })
245-
return mappedEvent === undefined
246-
? Effect.void
247-
: context.self.send(mappedEvent as Machine.EventOf<Events>).pipe(
248-
Effect.catchTag("StoppedError", () => Effect.void)
249-
)
250-
}
251-
return context.failCause(outcome.cause)
252-
})
253-
)
254-
}
255-
const handleInvokeSnapshot = (
256-
config: AnyInvokeConfig,
257-
key: string,
258-
token: symbol,
259-
snapshot: Extract<internalRuntime.RuntimeSnapshot<any, any, any>, { readonly status: "active" }>
260-
): Effect.Effect<void> =>
261-
isCurrentInvoke(key, token).pipe(
262-
Effect.flatMap((isCurrent) => {
263-
if (!isCurrent || config.snapshot === undefined) {
264-
return Effect.void
265-
}
266-
const mappedEvent = config.snapshot({ id: config.id, snapshot })
267-
return mappedEvent === undefined
268-
? Effect.void
269-
: context.self.send(mappedEvent as Machine.EventOf<Events>).pipe(
270-
Effect.catchTag("StoppedError", () => Effect.void)
271-
)
272-
})
273-
)
274297
const startInvoke = Effect.fnUntraced(function*<StateId extends Machine.StateIdentifier<States>>(
275298
path: StateId,
276299
config: AnyInvokeConfig
@@ -291,34 +314,29 @@ const makeProcessLogic: <
291314
}
292315
const logic = config.src()
293316
const processLogic = logic as internalRuntime.ProcessLogic<any, any, any, any, any, any>
294-
const sendParent = (event: unknown): Effect.Effect<void, StoppedError> =>
295-
isCurrentInvoke(key, token).pipe(
296-
Effect.flatMap((isCurrent) =>
297-
isCurrent ? context.self.send(event as Machine.EventOf<Events>) : Effect.void
298-
)
299-
)
317+
const sendParent = makeInvokeSendParent(invokeSessions, context.self, key, token)
300318
yield* context.spawn(
301-
{
302-
...(processLogic[internalRuntime.childlessProcess] === true
303-
? { [internalRuntime.childlessProcess]: true as const }
304-
: undefined),
305-
...(processLogic[internalRuntime.compiledProcess] === true
306-
? { [internalRuntime.compiledProcess]: true as const }
307-
: undefined),
308-
initial: (childScope) => logic.initial({ ...childScope, sendParent }),
309-
run: (childContext) => logic.run({ ...childContext, sendParent }),
310-
...(processLogic.drain === undefined ? undefined : {
311-
drain: (childContext: internalRuntime.ProcessContext<any, any>) =>
312-
processLogic.drain!({ ...childContext, sendParent })
313-
})
314-
},
319+
processLogic,
315320
{
316321
id: childId,
317322
...(config.descriptor === undefined ? undefined : { descriptor: config.descriptor }),
318-
onOutcome: (outcome) => handleInvokeOutcome(config, key, token, outcome),
323+
[internalRuntime.sendParentOverride]: sendParent,
324+
onOutcome: makeInvokeOutcomeHandler(
325+
invokeSessions,
326+
context.self,
327+
context.failCause,
328+
config,
329+
key,
330+
token
331+
),
319332
...(config.snapshot === undefined ? undefined : {
320-
[internalRuntime.activeSnapshotObserver]: (snapshot) =>
321-
handleInvokeSnapshot(config, key, token, snapshot)
333+
[internalRuntime.activeSnapshotObserver]: makeInvokeSnapshotHandler(
334+
invokeSessions,
335+
context.self,
336+
config,
337+
key,
338+
token
339+
)
322340
})
323341
}
324342
).pipe(
@@ -531,8 +549,6 @@ const makeProcessLogic: <
531549
const makeInvokeSessionKey = (path: string, id: string): string => `${path.length}:${path}${id}`
532550
const makeInvokeChildId = (path: string, id: string): string =>
533551
`Machine.invoke:${makeInvokeSessionKey(path, id)}`
534-
const isCurrentInvoke = (key: string, token: symbol): Effect.Effect<boolean> =>
535-
Effect.sync(() => invokeSessions.get(key)?.token === token)
536552
const stopInvokeSession = (session: InvokeSession): Effect.Effect<void> => context.stopChild(session.childId)
537553
const removeInvoke = (
538554
key: string,
@@ -565,53 +581,6 @@ const makeProcessLogic: <
565581
)
566582
)
567583
)
568-
const handleInvokeOutcome = (
569-
config: AnyInvokeConfig,
570-
key: string,
571-
token: symbol,
572-
outcome: internalRuntime.RuntimeOutcome<any, any, any>
573-
): Effect.Effect<void> => {
574-
if (outcome._tag === "Stopped") {
575-
return Effect.void
576-
}
577-
return isCurrentInvoke(key, token).pipe(
578-
Effect.flatMap((isCurrent) => {
579-
if (!isCurrent) {
580-
return Effect.void
581-
}
582-
if (outcome._tag === "Done") {
583-
const mappedEvent = config.onDone === undefined
584-
? outcome.output
585-
: config.onDone({ id: config.id, output: outcome.output })
586-
return mappedEvent === undefined
587-
? Effect.void
588-
: context.self.send(mappedEvent as Machine.EventOf<Events>).pipe(
589-
Effect.catchTag("StoppedError", () => Effect.void)
590-
)
591-
}
592-
return context.failCause(outcome.cause)
593-
})
594-
)
595-
}
596-
const handleInvokeSnapshot = (
597-
config: AnyInvokeConfig,
598-
key: string,
599-
token: symbol,
600-
snapshot: Extract<internalRuntime.RuntimeSnapshot<any, any, any>, { readonly status: "active" }>
601-
): Effect.Effect<void> =>
602-
isCurrentInvoke(key, token).pipe(
603-
Effect.flatMap((isCurrent) => {
604-
if (!isCurrent || config.snapshot === undefined) {
605-
return Effect.void
606-
}
607-
const mappedEvent = config.snapshot({ id: config.id, snapshot })
608-
return mappedEvent === undefined
609-
? Effect.void
610-
: context.self.send(mappedEvent as Machine.EventOf<Events>).pipe(
611-
Effect.catchTag("StoppedError", () => Effect.void)
612-
)
613-
})
614-
)
615584
const startInvoke = Effect.fnUntraced(function*<StateId extends Machine.StateIdentifier<States>>(
616585
path: StateId,
617586
config: AnyInvokeConfig
@@ -632,34 +601,29 @@ const makeProcessLogic: <
632601
}
633602
const logic = config.src()
634603
const processLogic = logic as internalRuntime.ProcessLogic<any, any, any, any, any, any>
635-
const sendParent = (event: unknown): Effect.Effect<void, StoppedError> =>
636-
isCurrentInvoke(key, token).pipe(
637-
Effect.flatMap((isCurrent) =>
638-
isCurrent ? context.self.send(event as Machine.EventOf<Events>) : Effect.void
639-
)
640-
)
604+
const sendParent = makeInvokeSendParent(invokeSessions, context.self, key, token)
641605
yield* context.spawn(
642-
{
643-
...(processLogic[internalRuntime.childlessProcess] === true
644-
? { [internalRuntime.childlessProcess]: true as const }
645-
: undefined),
646-
...(processLogic[internalRuntime.compiledProcess] === true
647-
? { [internalRuntime.compiledProcess]: true as const }
648-
: undefined),
649-
initial: (childScope) => logic.initial({ ...childScope, sendParent }),
650-
run: (childContext) => logic.run({ ...childContext, sendParent }),
651-
...(processLogic.drain === undefined ? undefined : {
652-
drain: (childContext: internalRuntime.ProcessContext<any, any>) =>
653-
processLogic.drain!({ ...childContext, sendParent })
654-
})
655-
},
606+
processLogic,
656607
{
657608
id: childId,
658609
...(config.descriptor === undefined ? undefined : { descriptor: config.descriptor }),
659-
onOutcome: (outcome) => handleInvokeOutcome(config, key, token, outcome),
610+
[internalRuntime.sendParentOverride]: sendParent,
611+
onOutcome: makeInvokeOutcomeHandler(
612+
invokeSessions,
613+
context.self,
614+
context.failCause,
615+
config,
616+
key,
617+
token
618+
),
660619
...(config.snapshot === undefined ? undefined : {
661-
[internalRuntime.activeSnapshotObserver]: (snapshot) =>
662-
handleInvokeSnapshot(config, key, token, snapshot)
620+
[internalRuntime.activeSnapshotObserver]: makeInvokeSnapshotHandler(
621+
invokeSessions,
622+
context.self,
623+
config,
624+
key,
625+
token
626+
)
663627
})
664628
}
665629
).pipe(

0 commit comments

Comments
 (0)