diff --git a/worker/README.md b/worker/README.md index 4b309e22..d0382b74 100644 --- a/worker/README.md +++ b/worker/README.md @@ -14,11 +14,61 @@ thread. ## Features - Establishes two-way communication between the main and the worker threads -- Gracefully shutdowns the worker from the main thread +- Gracefully or forcibly shuts down Workers with contextual policies - Propagates errors from the worker to the main thread - Type-safe message handling with TypeScript - Supports worker-initiated requests handled by the host +## Shutdown + +Workers shut down gracefully by default. When their host scope shuts down, +`useWorker()` sends a close message and waits for Worker-side teardown and the +final result. + +CPU-bound or otherwise non-cooperative Workers cannot process that close +message. Use `shutdown: "forced"` when cancellation should terminate the Worker +immediately: + +```ts +const worker = yield* useWorker("./worker.ts", { + type: "module", + shutdown: "forced", +}); +``` + +A generator policy first requests graceful shutdown, then chooses whether to +keep waiting or force termination. It runs in the Worker's host evaluation +scope, so it can use application context. For example, a host-owned heartbeat +monitor can expose the point when the Worker's control channel stops responding: + +```ts +import { createContext, type Operation } from "effection"; + +interface WorkerHealth { + controlChannelUnresponsive: Operation; +} + +const workerHealth = createContext("worker health"); + +const worker = yield* useWorker("./worker.ts", { + type: "module", + *shutdown() { + const health = yield* workerHealth.expect(); + yield* health.controlChannelUnresponsive; + return "forced"; + }, +}); +``` + +The heartbeat monitor is application-owned because the package cannot infer +whether CPU use or delayed messages mean that a particular Worker is unhealthy. +If the Worker completes while the policy is pending, Effection cancels the +policy and finishes gracefully. Returning `"graceful"` also keeps waiting for +Worker-side teardown. No timeout is imposed by the package. + +Forced termination does not run Worker-side finalizers, so durable cleanup for +a forcibly terminated Worker must be owned by the host. + ## Usage: Get worker's return value The return value of the worker is the return value of the function passed to diff --git a/worker/package.json b/worker/package.json index de70dc83..b5f9409d 100644 --- a/worker/package.json +++ b/worker/package.json @@ -1,7 +1,7 @@ { "name": "@effectionx/worker", - "description": "Web Worker integration with two-way messaging and graceful shutdown", - "version": "0.5.4", + "description": "Web Worker integration with two-way messaging and policy-driven shutdown", + "version": "0.6.0", "keywords": ["platform"], "type": "module", "main": "./dist/mod.js", diff --git a/worker/test-assets/cpu-bound-worker.ts b/worker/test-assets/cpu-bound-worker.ts new file mode 100644 index 00000000..6b97e83f --- /dev/null +++ b/worker/test-assets/cpu-bound-worker.ts @@ -0,0 +1,11 @@ +import { workerMain } from "../worker-main.ts"; + +await workerMain(function* ({ data }) { + let state = new Int32Array(data); + Atomics.store(state, 0, 1); + Atomics.notify(state, 0); + + while (true) { + Atomics.load(state, 0); + } +}); diff --git a/worker/worker.test.ts b/worker/worker.test.ts index 9814b0ff..23e604a5 100644 --- a/worker/worker.test.ts +++ b/worker/worker.test.ts @@ -1,10 +1,12 @@ import { access, mkdir, readFile, rm } from "node:fs/promises"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; -import { beforeEach, describe, it } from "@effectionx/vitest"; import { when } from "@effectionx/converge"; +import { beforeEach, describe, it } from "@effectionx/vitest"; import { all, + createContext, + type Operation, scoped, sleep, spawn, @@ -15,7 +17,11 @@ import { import { expect } from "expect"; import type { ShutdownWorkerParams } from "./test-assets/shutdown-worker.ts"; -import { useWorker } from "./worker.ts"; +import { + type UseWorkerOptions, + type WorkerShutdownPolicy, + useWorker, +} from "./worker.ts"; describe("worker", () => { it("sends and receive messages in synchrony", function* () { @@ -83,7 +89,43 @@ describe("worker", () => { url = import.meta.resolve("./test-assets/shutdown-worker.ts"); }); - it("shuts down gracefully", function* () { + function* haltCPUWorker( + shutdown: UseWorkerOptions["shutdown"], + ): Operation { + let state = new Int32Array( + new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT), + ); + let task = yield* spawn(function* () { + yield* useWorker( + import.meta.resolve("./test-assets/cpu-bound-worker.ts"), + { + type: "module", + data: state.buffer, + shutdown, + }, + ); + yield* suspend(); + }); + + yield* when( + function* () { + if (Atomics.load(state, 0) !== 1) { + throw new Error("worker has not started spinning"); + } + }, + { timeout: 10_000 }, + ); + + yield* task.halt(); + + try { + yield* task; + } catch (error) { + return error as Error; + } + } + + it("shuts down gracefully by default", function* () { let task = yield* spawn(function* () { yield* useWorker(url, { type: "module", @@ -96,7 +138,6 @@ describe("worker", () => { yield* suspend(); }); - // Wait for worker to start yield* when( function* () { let exists = yield* until( @@ -105,27 +146,95 @@ describe("worker", () => { () => false, ), ); - if (!exists) throw new Error("start file not found"); - return true; + if (!exists) { + throw new Error("worker has not started"); + } }, { timeout: 10_000 }, ); yield* task.halt(); - // Wait for the end file to be written with expected content - let { value: content } = yield* when( + expect(yield* until(readFile(endFile, "utf-8"))).toEqual( + "goodbye cruel world!", + ); + }); + + it("cancels its shutdown policy when graceful shutdown completes", function* () { + let shutdownContext = createContext("worker shutdown test"); + yield* shutdownContext.set("available during shutdown"); + + let policyContext: string | undefined; + let terminated = false; + let task = yield* spawn(function* () { + yield* useWorker(url, { + type: "module", + data: { + startFile, + endFile, + endText: "graceful", + } satisfies ShutdownWorkerParams, + *shutdown() { + policyContext = yield* shutdownContext.expect(); + yield* suspend(); + terminated = true; + return "forced"; + }, + }); + yield* suspend(); + }); + + yield* when( function* () { - let text = yield* until(readFile(endFile, "utf-8").catch(() => "")); - if (text !== "goodbye cruel world!") { - throw new Error(`expected "goodbye cruel world!", got "${text}"`); + let exists = yield* until( + access(startFile).then( + () => true, + () => false, + ), + ); + if (!exists) { + throw new Error("worker has not started"); } - return text; }, - { timeout: 500 }, + { timeout: 10_000 }, ); - expect(content).toEqual("goodbye cruel world!"); + yield* task.halt(); + + expect(yield* until(readFile(endFile, "utf-8"))).toEqual("graceful"); + expect(policyContext).toEqual("available during shutdown"); + expect(terminated).toEqual(false); + }); + + it("terminates a CPU-bound worker in forced mode", function* () { + expect.assertions(1); + const taskError = yield* haltCPUWorker("forced"); + expect(taskError?.message).toContain("halted"); + }); + + it("can force a CPU-bound worker from host health state", function* () { + expect.assertions(2); + const workerHealth = createContext<{ + controlChannelUnresponsive: Operation; + }>("worker health"); + const controlChannelUnresponsive = withResolvers(); + yield* workerHealth.set({ + controlChannelUnresponsive: controlChannelUnresponsive.operation, + }); + controlChannelUnresponsive.resolve(); + + let observedHealth = false; + const shutdown: WorkerShutdownPolicy = function* () { + const health = yield* workerHealth.expect(); + observedHealth = true; + yield* health.controlChannelUnresponsive; + return "forced"; + }; + + const taskError = yield* haltCPUWorker(shutdown); + + expect(observedHealth).toEqual(true); + expect(taskError?.message).toContain("halted"); }); }); diff --git a/worker/worker.ts b/worker/worker.ts index a3308326..9d62e4f2 100644 --- a/worker/worker.ts +++ b/worker/worker.ts @@ -7,6 +7,7 @@ import { ensure, on, once, + race, resource, spawn, withResolvers, @@ -69,6 +70,20 @@ export interface WorkerResource ): Operation; } +/** How an active resource should shut down with its owning scope. */ +export type ShutdownMode = "graceful" | "forced"; + +/** Selects a Worker shutdown mode from application state. */ +export type WorkerShutdownPolicy = () => Operation; + +/** Options for creating and shutting down a Worker. */ +export interface UseWorkerOptions extends WorkerOptions { + /** Data passed to `workerMain()` during initialization. */ + data?: TData; + /** Selects graceful, forced, or policy-driven shutdown. */ + shutdown?: ShutdownMode | WorkerShutdownPolicy; +} + /** * Use on the main thread to create and exeecute a well behaved web worker. * @@ -110,7 +125,7 @@ export interface WorkerResource * ``` * * @param url URL or string of script - * @param options WorkerOptions + * @param options Worker construction and shutdown options * @template TSend - value main thread will send to the worker * @template TRecv - value main thread will receive from the worker * @template TReturn - worker operation return value @@ -119,9 +134,10 @@ export interface WorkerResource */ export function useWorker( url: string | URL, - options?: WorkerOptions & { data?: TData }, + options?: UseWorkerOptions, ): Operation> { return resource(function* (provide) { + let { data, shutdown = "graceful", ...workerOptions } = options ?? {}; let outcome = withResolvers(); let outcomeSettled = false; @@ -141,7 +157,14 @@ export function useWorker( outcome.reject(error); }; - let worker = new Worker(url, options); + let worker = new Worker(url, workerOptions); + const terminate = (error = new Error("worker terminated")) => { + if (!outcomeSettled) { + worker.terminate(); + rejectOutcome(error); + } + }; + let subscription = yield* on(worker, "message"); // Channel for worker-initiated requests (buffered via eager subscription) @@ -217,21 +240,25 @@ export function useWorker( }); yield* ensure(function* () { - worker.postMessage({ type: "close" }); if (!outcomeSettled) { - while (!outcomeSettled) { - const event = yield* once(worker, "message"); - const msg = event.data; - if (msg.type === "close") { - const { result } = msg as { result: Result }; - if (result.ok) { - resolveOutcome(result.value); - } else { - const serializedError = - result.error as unknown as SerializedError; - rejectOutcome( - errorFromSerialized("Worker failed", serializedError), - ); + if (shutdown === "forced") { + terminate(); + } else { + worker.postMessage({ type: "close" }); + if (typeof shutdown === "function") { + let mode: ShutdownMode = "graceful"; + try { + mode = yield* race([ + gracefulCompletion(outcome.operation), + shutdown(), + ]); + } catch (error) { + if (!outcomeSettled) { + terminate(error as Error); + } + } + if (mode === "forced") { + terminate(); } } } @@ -241,7 +268,7 @@ export function useWorker( worker.postMessage({ type: "init", - data: options?.data, + data, }); yield* provide({ @@ -329,6 +356,13 @@ export function useWorker( }); } +function* gracefulCompletion( + outcome: Operation, +): Operation { + yield* outcome; + return "graceful"; +} + function settled(operation: Operation): Operation> { return { *[Symbol.iterator]() {