From 4fdd74fb09443761900fd95eb5e854d0d479318b Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 18 Oct 2026 19:59:07 +0200 Subject: [PATCH] fix!: correct task lifecycle and type export bugs (#3360) * fix: correct task lifecycle and type export bugs * test(types): isolate declaration probes from runtime coverage * fix(pool): synchronize replacement task function replay * fix: settle task function operations when workers exit * fix: retire worker message listeners before node removal * fix: preserve operation cohorts and replay readiness * [autofix.ci] apply automated fixes * fix: preserve task defaults and assimilate foreign promises Restore acknowledged defaults before readiness and preserve empty-cohort no-op semantics. Bound replay listeners with sequential operations and stable worker identity. Normalize foreign promises and thenables across tasks, cancellation and cleanup without changing protected execution contracts. * fix: snapshot and serialize task function mutations Adapt relevant fixes from poolifier-web-worker#196 without replacing Node lifecycle handling. Normalize decorated native promises and avoid constructing weights when counting retries. * fix: complete task lifecycle port and replay guarantees * fix: normalize async results and thenables * fix: skip unready weighted workers * fix: keep terminating replay workers unready * [autofix.ci] apply automated fixes * build: align ECMAScript target with ES2024 lib * build: align TypeScript examples with ES2024 * docs: clarify user-facing task lifecycle rules * fix: preserve task function mutation and result contracts * fix: preserve queued operation identity and abortable results * docs: align task function signatures with thenable support * fix: scope bootstrap failures by worker affinity * fix!: preserve task operation and strategy invariants Reject lost operation identity without releasing live cohorts early. Keep ready peers' strategies registered during worker replay. BREAKING CHANGE: deferred custom operation handlers must retain request UUIDs or the original task-function properties reference. Replies without recoverable identity reject instead of leaving changes pending. * fix: preserve buffered task function operation results --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- README.md | 2 +- docs/api.md | 62 +- docs/worker-choice-strategies.md | 2 + .../typescript/http-client-pool/tsconfig.json | 2 +- .../express-cluster/tsconfig.json | 2 +- .../express-hybrid/tsconfig.json | 2 +- .../express-worker_threads/tsconfig.json | 2 +- .../fastify-cluster/tsconfig.json | 2 +- .../fastify-hybrid/tsconfig.json | 2 +- .../fastify-worker_threads/tsconfig.json | 2 +- .../typescript/smtp-client-pool/tsconfig.json | 2 +- .../ws-cluster/tsconfig.json | 2 +- .../ws-hybrid/tsconfig.json | 2 +- .../ws-worker_threads/tsconfig.json | 2 +- package.json | 1 - pnpm-lock.yaml | 62 - rollup.config.mjs | 17 - src/pools/abstract-pool.ts | 927 +++++++-- src/pools/pool.ts | 4 +- .../selection-strategies-utils.ts | 3 +- ...hted-round-robin-worker-choice-strategy.ts | 2 + src/pools/worker-node.ts | 47 +- src/utility-types.ts | 12 +- src/worker/abstract-worker.ts | 310 ++- src/worker/cluster-worker.ts | 10 +- src/worker/task-functions.ts | 12 +- src/worker/thread-worker.ts | 21 +- src/worker/worker-options.ts | 5 +- tests/declarations.test.mjs | 76 + tests/pools/abstract-pool.test.mjs | 16 - tests/pools/task-function-replay.test.mjs | 1773 +++++++++++++++++ tests/pools/task-lifecycle.test.mjs | 1172 +++++++++++ tests/pools/worker-exit.test.mjs | 104 +- .../cluster/legacyTaskFunctionWorker.mjs | 60 + .../thread/bootstrapAffinityWorker.mjs | 6 +- .../thread/taskLifecycleWorker.mjs | 285 +++ tsconfig.build.json | 1 + tsconfig.json | 3 +- 38 files changed, 4543 insertions(+), 474 deletions(-) create mode 100644 tests/declarations.test.mjs create mode 100644 tests/pools/task-function-replay.test.mjs create mode 100644 tests/pools/task-lifecycle.test.mjs create mode 100644 tests/worker-files/cluster/legacyTaskFunctionWorker.mjs create mode 100644 tests/worker-files/thread/taskLifecycleWorker.mjs diff --git a/README.md b/README.md index 97c1014b6..341ad3baa 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,7 @@ Please consult our [general guidelines](#general-guidelines). - Support for multiple task functions: - Queuing priority :white_check_mark: - Tasks distribution strategy :white_check_mark: - - Worker node affinity :white_check_mark: + - [Worker node affinity](./docs/api.md#worker-node-affinity-notes) :white_check_mark: - [CRUD](https://en.wikipedia.org/wiki/Create,_read,_update_and_delete) operations at runtime :white_check_mark: - General guidelines on pool choice :white_check_mark: - Error handling out of the box :white_check_mark: diff --git a/docs/api.md b/docs/api.md index fa7bc3d10..b06f56dda 100644 --- a/docs/api.md +++ b/docs/api.md @@ -43,28 +43,34 @@ `data` (optional) An object that you want to pass to your worker task function implementation. `name` (optional) A string with the task function name that you want to execute on the worker. Default: `'default'` `abortSignal` (optional) An abort signal to abort the task function execution. -`transferList` (optional) An array of transferable objects that you want to transfer to your [`ThreadWorker`](#class-yourworker-extends-threadworkerclusterworker) worker implementation. +`transferList` (optional) An array of transferable objects that you want to transfer to your [`ThreadWorker`](#class-yourworker-extends-threadworkerclusterworker) worker implementation. The array is copied at submission. This method is available on both pool implementations and returns a promise with the task function execution response. +Promise and thenable task results are awaited, including promises from other JavaScript realms. Functions need not be declared `async`. + +An already-aborted signal rejects the task before submission. + #### Task failures - If sending a task throws, its promise rejects with the original error, without failing other tasks. -- On unexpected worker exit, unfinished dispatched tasks reject with an `Error` and are not retried. Thread responses already posted to the pool retain their result or task error; cluster responses must reach the pool before exit. -- Queued tasks are redistributed to ready workers respecting worker node affinity. Worker node affinity is preserved while other workers initialize. Affinity uses current worker node indices after removal and any replacement attempt. Tasks without an eligible destination reject. Eligible handoffs complete before `backPressureEnd` is emitted, including when worker-removal event listeners disable task queues. -- Disabling task queues from a worker-removal event listener does not discard queued tasks awaiting redistribution: eligible tasks are dispatched directly. -- Queued tasks aborted after worker removal and before reassignment reject with the original abort reason if it is an `Error`, a new `Error` with that message if it is a string, or a task-specific abort `Error` otherwise. -- Non-`Error` native worker failures are wrapped in an `Error` with the original value in `cause`. The pool's `error` event retains the original value. -- Failure to construct a replacement worker does not prevent task settlement; the construction error still propagates. +- On unexpected worker exit, unfinished tasks already sent to it reject with an `Error` and are not retried. Thread responses sent before exit retain their result or task error; cluster responses must reach the pool before exit. +- Task errors reported by a worker reject the task promise and emit `taskError` when events are enabled. For tasks with transferred objects, the event's `data` is the original input object; reporting does not return ownership of those objects. Other task errors may omit input that cannot be copied back. +- If a worker cannot report a task error, it exits unexpectedly and unfinished tasks follow the rules above. +- Queued tasks move to ready workers allowed by worker node affinity, using current worker node indices after removal and any replacement attempt. Affinity still applies while other workers initialize. Tasks without an eligible destination reject. +- Eligible tasks are reassigned before `backPressureEnd` is emitted. Disabling task queues during reassignment sends eligible queued tasks directly instead of discarding them. +- Task cancellation preserves an `Error` reason, wraps a string reason in a new `Error`, or produces a task-specific abort `Error` otherwise. +- Worker failures emit the pool's `error` event with the original value when events are enabled. Rejections wrap non-`Error` values in an `Error` with the original value as `cause`. +- If a replacement worker cannot be created, affected task promises still settle; the creation error is not suppressed. -Unexpected-exit handling does not apply to intentional pool or worker termination. +Intentional termination alone does not guarantee rejection of unfinished task promises. Worker errors during termination still follow the failure rules above. ### `pool.mapExecute(data, name, abortSignals, transferList)` `data` An iterable of objects that you want to pass to your worker task function implementation. `name` (optional) A string with the task function name that you want to execute on the worker. Default: `'default'` -`abortSignals` (optional) An iterable of AbortSignal to abort the matching object task function execution. -`transferList` (optional) An array of transferable objects that you want to transfer to your [`ThreadWorker`](#class-yourworker-extends-threadworkerclusterworker) worker implementation. +`abortSignals` (optional) An iterable of AbortSignal matching the input order and length. One-shot iterables are supported. +`transferList` (optional) Same as [`pool.execute()`](#poolexecutedata-name-abortsignal-transferlist), with one array snapshot for the batch. This method is available on both pool implementations and returns a promise with the task function execution responses array. @@ -78,6 +84,8 @@ This method is available on both pool implementations and will start the minimum This method is available on both pool implementations and returns a promise for pool destruction. Workers that have already exited do not block destruction. +The pool waits up to 1000 milliseconds for each worker to confirm cleanup before terminating it. Cleanup failure or timeout does not prevent termination; unfinished cleanup can be interrupted. Reported cleanup failures emit the pool's `error` event when events are enabled. This timeout does not bound total pool destruction time. + ### `pool.hasTaskFunction(name)` `name` (mandatory) The task function name. @@ -87,22 +95,26 @@ This method is available on both pool implementations and returns a boolean. ### `pool.addTaskFunction(name, fn)` `name` (mandatory) The task function name. -`fn` (mandatory) The task function `(data?: Data) => Response | Promise` or task function object `{ taskFunction: (data?: Data) => Response | Promise, priority?: number, strategy?: WorkerChoiceStrategy, workerNodeKeys?: number[] }`. Priority range is the same as Unix nice levels. `workerNodeKeys` is an array of worker node keys to restrict task execution to specific workers (worker node affinity). +`fn` (mandatory) The task function `(data?: Data) => Response | PromiseLike` or task function object `{ taskFunction: (data?: Data) => Response | PromiseLike, priority?: number, strategy?: WorkerChoiceStrategy, workerNodeKeys?: number[] }`. Priority range is the same as Unix nice levels. `workerNodeKeys` restricts initial task dispatch to specific workers (worker node affinity). + +This method is available on both pool implementations and returns a boolean promise. See [Runtime task function changes](#runtime-task-function-changes). + +The function's source and options, including `workerNodeKeys`, are captured at call time. Later caller changes do not affect registration or replacement workers. #### Worker Node Affinity Notes - Worker node keys are validated at registration time against the pool's maximum size (`maximumNumberOfWorkers ?? minimumNumberOfWorkers`). - The number of worker node keys cannot exceed the pool's maximum size (`maximumNumberOfWorkers ?? minimumNumberOfWorkers`). - In dynamic pools, you can reference worker node keys up to the maximum pool size. Workers that don't exist yet are automatically created when a task targeting them is executed. -- At execution time, if no specified worker is ready, selection retries until one becomes available or retries are exhausted. - -This method is available on both pool implementations and returns a boolean promise. +- At execution time, if no specified worker is ready, selection retries until one becomes available or retries are exhausted. Initialization failures on ineligible workers do not reject waiting tasks. +- Task stealing on idle or under back pressure ignores worker node affinity. +- With task queues enabled, disable both `taskStealing` and `tasksStealingOnBackPressure` for strict worker node affinity. Both are enabled by default. ### `pool.removeTaskFunction(name)` `name` (mandatory) The task function name. -This method is available on both pool implementations and returns a boolean promise. +This method is available on both pool implementations and returns a boolean promise. See [Runtime task function changes](#runtime-task-function-changes). ### `pool.listTaskFunctionsProperties()` @@ -112,7 +124,19 @@ This method is available on both pool implementations and returns an array of th `name` (mandatory) The task function name. -This method is available on both pool implementations and returns a boolean promise. +This method is available on both pool implementations and returns a boolean promise. See [Runtime task function changes](#runtime-task-function-changes). + +With no workers, this method resolves to `true` without changing the default selection. + +### Runtime task function changes + +Pool-wide calls to `addTaskFunction()`, `removeTaskFunction()` and `setDefaultTaskFunction()` run in call order. A failed call does not block later calls or undo changes already applied to workers. + +If workers are present when a change starts, each must respond or exit before the call settles. At least one successful result is required; any reported failure rejects the change. Thread pools preserve operation results sent before an unexpected worker exit. + +New and replacement workers wait for ongoing task function changes. They accept tasks only after restoring successfully registered task functions, the selected default and scheduling options. Workers being terminated remain unavailable. + +Custom worker subclasses that defer task function operations must echo the request's `taskFunctionOperationId` in their acknowledgement, or retain the original `taskFunctionProperties` object when using the standard transport. An acknowledgement without a recoverable UUID rejects the operation. Pool-wide changes still wait for all original targets to respond or exit. Status-only overrides that discard this correlation must be updated. ### Pool options @@ -181,7 +205,7 @@ An object with these properties: ### `class YourWorker extends ThreadWorker/ClusterWorker` -`taskFunctions` (mandatory) The task function or task functions object `Record Response | Promise | { taskFunction: (data?: Data) => Response | Promise, priority?: number, strategy?: WorkerChoiceStrategy, workerNodeKeys?: number[] }>` that you want to execute on the worker. Priority range is the same as Unix nice levels. `workerNodeKeys` is an array of worker node keys to restrict task execution to specific workers (worker node affinity). See [Worker Node Affinity Notes](#worker-node-affinity-notes) above for validation behavior. +`taskFunctions` (mandatory) The task function or task functions object `Record Response | PromiseLike | { taskFunction: (data?: Data) => Response | PromiseLike, priority?: number, strategy?: WorkerChoiceStrategy, workerNodeKeys?: number[] }>` that you want to execute on the worker. Priority range is the same as Unix nice levels. `workerNodeKeys` restricts initial task dispatch to specific workers (worker node affinity). See [Worker Node Affinity Notes](#worker-node-affinity-notes) for affinity behavior. `opts` (optional) An object with these properties: - `killBehavior` (optional) - Dictates if your worker will be deleted in case a task is active on it. @@ -196,7 +220,7 @@ An object with these properties: If `killBehavior` is set to `KillBehaviors.SOFT` your tasks have no timeout and your workers will not be terminated until your task is completed. Default: `60000` -- `killHandler` (optional) - A function that will be called when a worker is killed. +- `killHandler` (optional) - A cleanup function called when the pool requests worker termination. Returned promises and thenables are awaited, subject to the [`pool.destroy()` timeout and failure rules](#pooldestroy). Default: `() => {}` #### `YourWorker.hasTaskFunction(name)` @@ -208,7 +232,7 @@ This method is available on both worker implementations and returns `{ status: b #### `YourWorker.addTaskFunction(name, fn)` `name` (mandatory) The task function name. -`fn` (mandatory) The task function `(data?: Data) => Response | Promise` or task function object `{ taskFunction: (data?: Data) => Response | Promise, priority?: number, strategy?: WorkerChoiceStrategy, workerNodeKeys?: number[] }`. Priority range is the same as Unix nice levels. `workerNodeKeys` is an array of worker node keys to restrict task execution to specific workers (worker node affinity). See [Worker Node Affinity Notes](#worker-node-affinity-notes) above for validation behavior. +`fn` (mandatory) The task function `(data?: Data) => Response | PromiseLike` or task function object `{ taskFunction: (data?: Data) => Response | PromiseLike, priority?: number, strategy?: WorkerChoiceStrategy, workerNodeKeys?: number[] }`. Priority range is the same as Unix nice levels. `workerNodeKeys` restricts initial task dispatch to specific workers (worker node affinity). See [Worker Node Affinity Notes](#worker-node-affinity-notes) for affinity behavior. This method is available on both worker implementations and returns `{ status: boolean, error?: Error }`. diff --git a/docs/worker-choice-strategies.md b/docs/worker-choice-strategies.md index 36d6cf70f..017d01e18 100644 --- a/docs/worker-choice-strategies.md +++ b/docs/worker-choice-strategies.md @@ -22,6 +22,8 @@ By default, the strategy uses the simple moving average task execution time for The worker weights are maximum tasks execution time. Once the worker has reached its maximum tasks execution time, the next task is assigned to the next worker. The default worker weight is the same for each and computed given the CPU cores speed and theirs numbers. +Workers that are not ready are skipped. + ### Interleaved weighted round robin (experimental) The worker weights are maximum tasks execution time. The rounds are the deduplicated worker weights in ascending order. diff --git a/examples/typescript/http-client-pool/tsconfig.json b/examples/typescript/http-client-pool/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/http-client-pool/tsconfig.json +++ b/examples/typescript/http-client-pool/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/http-server-pool/express-cluster/tsconfig.json b/examples/typescript/http-server-pool/express-cluster/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/http-server-pool/express-cluster/tsconfig.json +++ b/examples/typescript/http-server-pool/express-cluster/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/http-server-pool/express-hybrid/tsconfig.json b/examples/typescript/http-server-pool/express-hybrid/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/http-server-pool/express-hybrid/tsconfig.json +++ b/examples/typescript/http-server-pool/express-hybrid/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/http-server-pool/express-worker_threads/tsconfig.json b/examples/typescript/http-server-pool/express-worker_threads/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/http-server-pool/express-worker_threads/tsconfig.json +++ b/examples/typescript/http-server-pool/express-worker_threads/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/http-server-pool/fastify-cluster/tsconfig.json b/examples/typescript/http-server-pool/fastify-cluster/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/http-server-pool/fastify-cluster/tsconfig.json +++ b/examples/typescript/http-server-pool/fastify-cluster/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/http-server-pool/fastify-hybrid/tsconfig.json b/examples/typescript/http-server-pool/fastify-hybrid/tsconfig.json index 016d5a668..132f00ee6 100644 --- a/examples/typescript/http-server-pool/fastify-hybrid/tsconfig.json +++ b/examples/typescript/http-server-pool/fastify-hybrid/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/http-server-pool/fastify-worker_threads/tsconfig.json b/examples/typescript/http-server-pool/fastify-worker_threads/tsconfig.json index 016d5a668..132f00ee6 100644 --- a/examples/typescript/http-server-pool/fastify-worker_threads/tsconfig.json +++ b/examples/typescript/http-server-pool/fastify-worker_threads/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/smtp-client-pool/tsconfig.json b/examples/typescript/smtp-client-pool/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/smtp-client-pool/tsconfig.json +++ b/examples/typescript/smtp-client-pool/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/websocket-server-pool/ws-cluster/tsconfig.json b/examples/typescript/websocket-server-pool/ws-cluster/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/websocket-server-pool/ws-cluster/tsconfig.json +++ b/examples/typescript/websocket-server-pool/ws-cluster/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/websocket-server-pool/ws-hybrid/tsconfig.json b/examples/typescript/websocket-server-pool/ws-hybrid/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/websocket-server-pool/ws-hybrid/tsconfig.json +++ b/examples/typescript/websocket-server-pool/ws-hybrid/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/examples/typescript/websocket-server-pool/ws-worker_threads/tsconfig.json b/examples/typescript/websocket-server-pool/ws-worker_threads/tsconfig.json index 3eddea483..3fbdee824 100644 --- a/examples/typescript/websocket-server-pool/ws-worker_threads/tsconfig.json +++ b/examples/typescript/websocket-server-pool/ws-worker_threads/tsconfig.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", "module": "Node16", "moduleResolution": "Node16", "verbatimModuleSyntax": true, diff --git a/package.json b/package.json index 35a0963c8..7bb781205 100644 --- a/package.json +++ b/package.json @@ -126,7 +126,6 @@ "rollup-plugin-analyzer": "^4.0.0", "rollup-plugin-command": "^1.1.3", "rollup-plugin-delete": "^3.0.2", - "rollup-plugin-dts": "^6.5.1", "sinon": "^22.1.0", "tinybench": "^6.2.2", "tslib": "^2.8.1", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 26aa57bb7..943054db5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -235,9 +235,6 @@ importers: rollup-plugin-delete: specifier: ^3.0.2 version: 3.0.2(rollup@4.64.7) - rollup-plugin-dts: - specifier: ^6.5.1 - version: 6.5.1(rollup@4.64.7)(typescript@6.0.3) sinon: specifier: ^22.1.0 version: 22.1.0 @@ -264,10 +261,6 @@ packages: resolution: {integrity: sha512-Aup7aUOfpbAUg2ROOJN6Iw5f9DMBlzu0mIkm/malLQFN/YQgO48wCj0Kxa3sEHJvPVFg7siR+qRInwXd2qhQKw==} engines: {node: '>=6.9.0'} - '@babel/code-frame@8.0.6': - resolution: {integrity: sha512-uMHhvmDmwFiLuiozaoRYTZHqMFuu3fem/X9JY78lohAuLoh8LSGGC8JqUzo2cVuBsBxkjMZNk56CE7ock8Xxkw==} - engines: {node: ^22.18.0 || >=24.11.0} - '@babel/helper-string-parser@7.29.7': resolution: {integrity: sha512-Pb5ijPrZ89GDH8223L4UP8i6QApWxs04RbPQJTeWDV0/keR2E36MeKnyr6LYmUUvqRRI+Iv87SuF1W6ErINzYw==} engines: {node: '>=6.9.0'} @@ -276,10 +269,6 @@ packages: resolution: {integrity: sha512-qehxGkRj55h/ff8EMaJ+cYhyaKlHIxqYDn682wQD7RNp9UujOQsHog2uS0r2vzr4pW+sXf90NeeayjcNaX3fFg==} engines: {node: '>=6.9.0'} - '@babel/helper-validator-identifier@8.0.6': - resolution: {integrity: sha512-HYx0YmuchRxRSpnjvcNffnPS8DCauKNmnumIuHG3jnQAVewRA7kfJGwF13ayEpSZs7s4R/oBH4eppuXcAQd5JQ==} - engines: {node: ^22.18.0 || >=24.11.0} - '@babel/parser@7.29.9': resolution: {integrity: sha512-CjXrNHTnvqBVqHgdBysY3vk2T8tpJHb5/RMeHJBTyVa9xgugCB0CJTx/3oO8RV2QRQP391RWpB7D6hLjm8V9uA==} engines: {node: '>=6.0.0'} @@ -752,9 +741,6 @@ packages: '@jridgewell/gen-mapping@0.3.13': resolution: {integrity: sha512-2kkt/7niJ6MgEPxF0bYdQ6etZaA+fQvDcLKckhy1yIQOzaoKjBBjSj63/aLVjYE3qhRt5dvM+uUyfCg6UKCBbA==} - '@jridgewell/remapping@2.3.5': - resolution: {integrity: sha512-LI9u/+laYG4Ds1TDKSJW2YPrIlcVYOwi2fUC6xB43lueCjgxV4lffOCZCtYFiH6TNOX+tQKXx97T4IKHbhyHEQ==} - '@jridgewell/resolve-uri@3.1.2': resolution: {integrity: sha512-bRISgCIjP20/tbWSPWMEi54QVPRZExkuD9lJL+UIxUKtwVJA8wW1Trb1jMs1RFXo1CBTNZ/5hpC9QvmKWdopKw==} engines: {node: '>=6.0.0'} @@ -1338,9 +1324,6 @@ packages: engines: {node: '>=22'} hasBin: true - convert-source-map@2.0.0: - resolution: {integrity: sha512-Kvp459HrV2FEJ1CAsi1Ku+MY3kasH19TFykTz2xWmMeq6bk2NU3XXvfJ+Q61m0xktWwt+1HSYf3JZsTms3aRJg==} - cosmiconfig-typescript-loader@6.3.0: resolution: {integrity: sha512-Akr82WH1Wfqatyiqpj8HDkO2o2KmJRu1FhKfSNJP3K4IdXwHfEyL7MOb62i1AGQVLtIQM+iCE9CGOtrfhR+mmA==} engines: {node: '>=v18'} @@ -1915,9 +1898,6 @@ packages: lunr@2.3.9: resolution: {integrity: sha512-zTU3DaZaF3Rt9rhN3uBMGQD3dD2/vFQqnvZCDv4dl5iOzq2IZQqTxu90r4E5J+nP70J3ilqVCrbho2eWaeW8Ow==} - magic-string@0.30.21: - resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} - magic-string@1.4.3: resolution: {integrity: sha512-z12OxmaPGE0F4xlpGdjuAUckipLpQlTAuAmyGhNoD7ODpYUzvDfvtxUbjM/jwznyP178oju/KDdyi220wNJ0RQ==} @@ -2118,17 +2098,6 @@ packages: peerDependencies: rollup: '*' - rollup-plugin-dts@6.5.1: - resolution: {integrity: sha512-jODTXp3H7MK/Ur/ErtsrQ0G1GvaCmc3du+y5pNrdBMf6d7HlL2Nd/N6TkEr+f75CkUj01zEoEd7y2elH0eHi1Q==} - engines: {node: '>=20'} - peerDependencies: - '@typescript/typescript6': ^6 - rollup: ^3 || ^4 - typescript: ^4.5 || ^5 || ^6 || ^7 - peerDependenciesMeta: - '@typescript/typescript6': - optional: true - rollup@4.64.7: resolution: {integrity: sha512-aCiJONoZp0lNGJ6sYqspLkh6UXaWEey/7Ko2TZyXQ4kgzAhDUxAgbt/oulXKmD+zFHXw1k+RF6ZZWOrjp7DZnQ==} engines: {node: '>=18.0.0', npm: '>=8.0.0'} @@ -2454,19 +2423,10 @@ snapshots: js-tokens: 4.0.0 picocolors: 1.1.1 - '@babel/code-frame@8.0.6': - dependencies: - '@babel/helper-validator-identifier': 8.0.6 - js-tokens: 10.0.0 - optional: true - '@babel/helper-string-parser@7.29.7': {} '@babel/helper-validator-identifier@7.29.7': {} - '@babel/helper-validator-identifier@8.0.6': - optional: true - '@babel/parser@7.29.9': dependencies: '@babel/types': 7.29.8 @@ -2943,11 +2903,6 @@ snapshots: '@jridgewell/sourcemap-codec': 1.6.0 '@jridgewell/trace-mapping': 0.3.31 - '@jridgewell/remapping@2.3.5': - dependencies: - '@jridgewell/gen-mapping': 0.3.13 - '@jridgewell/trace-mapping': 0.3.31 - '@jridgewell/resolve-uri@3.1.2': {} '@jridgewell/source-map@0.3.11': @@ -3435,8 +3390,6 @@ snapshots: '@simple-libs/stream-utils': 2.0.0 argue-cli: 3.2.0 - convert-source-map@2.0.0: {} - cosmiconfig-typescript-loader@6.3.0(@types/node@24.19.2)(cosmiconfig@9.0.2(typescript@6.0.3))(typescript@6.0.3): dependencies: '@types/node': 24.19.2 @@ -3989,10 +3942,6 @@ snapshots: lunr@2.3.9: {} - magic-string@0.30.21: - dependencies: - '@jridgewell/sourcemap-codec': 1.6.0 - magic-string@1.4.3: dependencies: '@jridgewell/sourcemap-codec': 1.6.0 @@ -4184,17 +4133,6 @@ snapshots: del: 8.0.1 rollup: 4.64.7 - rollup-plugin-dts@6.5.1(rollup@4.64.7)(typescript@6.0.3): - dependencies: - '@jridgewell/remapping': 2.3.5 - '@jridgewell/sourcemap-codec': 1.6.0 - convert-source-map: 2.0.0 - magic-string: 0.30.21 - rollup: 4.64.7 - typescript: 6.0.3 - optionalDependencies: - '@babel/code-frame': 8.0.6 - rollup@4.64.7: dependencies: '@types/estree': 1.0.9 diff --git a/rollup.config.mjs b/rollup.config.mjs index ae63b99fc..12e0b3190 100644 --- a/rollup.config.mjs +++ b/rollup.config.mjs @@ -6,7 +6,6 @@ import { defineConfig } from 'rollup' import analyze from 'rollup-plugin-analyzer' import command from 'rollup-plugin-command' import del from 'rollup-plugin-delete' -import { dts } from 'rollup-plugin-dts' const availableParallelism = () => { let availableParallelism = 1 @@ -88,20 +87,4 @@ export default defineConfig([ ], strictDeprecations: true, }, - { - external: [/^node:*/], - input: './lib/index.d.ts', - output: [{ file: './lib/index.d.ts', format: 'esm' }], - plugins: [ - dts(), - del({ - hook: 'buildEnd', - targets: isDevelopmentBuild - ? ['./lib/**/*.d.ts', '!./lib/index.d.ts'] - : ['./lib/**/*.d.ts', '!./lib/index.d.ts', './lib/*/'], - }), - isAnalyzeBuild && analyze(), - ], - strictDeprecations: true, - }, ]) diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 57303027b..ec606b87e 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -441,11 +441,20 @@ export abstract class AbstractPool< */ private busyEventEmitted: boolean + /** + * The last explicit default acknowledged by an operation's worker cohort. + */ + private defaultTaskFunctionName?: string + + private pendingTaskFunctionOperations = 0 + /** * Whether the pool ready event has been emitted or not. */ private readyEventEmitted: boolean + private resolveTaskFunctionOperations?: (value: undefined) => void + /** * Whether the minimum number of workers is starting or not. */ @@ -455,20 +464,25 @@ export abstract class AbstractPool< * The start timestamp of the pool. */ private startTimestamp?: number - /** * Whether task dequeuing events are deferred until worker exit handoffs finish. */ private taskDequeuingEventsDeferred = false - + private taskFunctionOperationsFinished?: Promise + private taskFunctionOperationsQueue?: Promise /** * The task functions added at runtime map: * - `key`: The task function name. - * - `value`: The task function object. + * - `value`: The captured task function registration message. */ private readonly taskFunctions: Map< string, - TaskFunctionObject + Required< + Pick< + MessageValue, + 'taskFunction' | 'taskFunctionOperation' | 'taskFunctionProperties' + > + > > /** @@ -478,6 +492,16 @@ export abstract class AbstractPool< IWorkerNode >() + private readonly workerTaskFunctionBootstraps = new Map< + IWorkerNode, + { promise?: Promise; ready: boolean } + >() + + private readonly workerTaskFunctionOperationExitListeners = new Map< + IWorkerNode, + Set<() => void> + >() + /** * Whether the pool is ready or not. * @returns The pool readiness boolean status. @@ -567,7 +591,7 @@ export abstract class AbstractPool< this.setupHook() - this.taskFunctions = new Map>() + this.taskFunctions = new Map() this.started = false this.starting = false @@ -592,31 +616,51 @@ export abstract class AbstractPool< if (typeof name === 'string' && name.trim().length === 0) { throw new TypeError('name argument must not be an empty string') } - if (typeof fn === 'function') { - fn = { taskFunction: fn } satisfies TaskFunctionObject - } - if (typeof fn.taskFunction !== 'function') { + const taskFunction = typeof fn === 'function' ? fn : fn.taskFunction + if (typeof taskFunction !== 'function') { throw new TypeError('taskFunction property must be a function') } - checkValidPriority(fn.priority) - checkValidWorkerChoiceStrategy(fn.strategy) + const definition: TaskFunctionObject = { taskFunction } + if (typeof fn !== 'function') { + const { priority, strategy, workerNodeKeys } = fn + if (priority !== undefined) definition.priority = priority + if (strategy !== undefined) definition.strategy = strategy + if (workerNodeKeys !== undefined) { + definition.workerNodeKeys = Array.isArray(workerNodeKeys) + ? [...workerNodeKeys] + : workerNodeKeys + } + } + checkValidPriority(definition.priority) + checkValidWorkerChoiceStrategy(definition.strategy) checkValidWorkerNodeKeys( - fn.workerNodeKeys, + definition.workerNodeKeys, this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers ) - const opResult = await this.sendTaskFunctionOperationToWorkers({ - taskFunction: fn.taskFunction.toString(), + const message = { + taskFunction: taskFunction.toString(), taskFunctionOperation: 'add', - taskFunctionProperties: buildTaskFunctionProperties(name, fn), + taskFunctionProperties: buildTaskFunctionProperties(name, definition), + } satisfies MessageValue + return await this.queueTaskFunctionOperation(async () => { + while (this.workerTaskFunctionBootstraps.size > 0) { + await this.waitForWorkerTaskFunctionBootstraps() + } + this.beginTaskFunctionOperation() + try { + const opResult = await this.sendTaskFunctionOperationToWorkers(message) + this.taskFunctions.set(name, message) + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + this.getWorkerChoiceStrategies() + ) + for (const workerNodeKey of this.workerNodes.keys()) { + this.sendStatisticsMessageToWorker(workerNodeKey) + } + return opResult + } finally { + this.endTaskFunctionOperation() + } }) - this.taskFunctions.set(name, fn) - this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( - this.getWorkerChoiceStrategies() - ) - for (const workerNodeKey of this.workerNodes.keys()) { - this.sendStatisticsMessageToWorker(workerNodeKey) - } - return opResult } /** @inheritDoc */ @@ -693,6 +737,10 @@ export abstract class AbstractPool< if (transferList != null && !Array.isArray(transferList)) { throw new TypeError('transferList argument must be an array') } + transferList = + transferList != null && transferList.length > 0 + ? Array.from(transferList) + : undefined return await this.internalExecute(data, name, abortSignal, transferList) } @@ -705,15 +753,15 @@ export abstract class AbstractPool< /** @inheritDoc */ public listTaskFunctionsProperties (): TaskFunctionProperties[] { + let fallback: TaskFunctionProperties[] | undefined for (const workerNode of this.workerNodes) { - if ( - Array.isArray(workerNode.info.taskFunctionsProperties) && - workerNode.info.taskFunctionsProperties.length > 0 - ) { - return workerNode.info.taskFunctionsProperties + const properties = workerNode.info.taskFunctionsProperties + if (Array.isArray(properties) && properties.length > 0) { + if (workerNode.info.ready) return properties + fallback ??= properties } } - return [] + return fallback ?? [] } /** @inheritDoc */ @@ -749,6 +797,9 @@ export abstract class AbstractPool< if (typeof abortSignals[Symbol.iterator] !== 'function') { throw new TypeError('abortSignals argument must be an iterable') } + if (!Array.isArray(abortSignals)) { + abortSignals = [...abortSignals] + } for (const abortSignal of abortSignals) { if (!(abortSignal instanceof AbortSignal)) { throw new TypeError( @@ -756,9 +807,6 @@ export abstract class AbstractPool< ) } } - if (!Array.isArray(abortSignals)) { - abortSignals = [...abortSignals] - } if ((data as Data[]).length !== (abortSignals as AbortSignal[]).length) { throw new Error( 'data and abortSignals arguments must have the same length' @@ -768,6 +816,10 @@ export abstract class AbstractPool< if (transferList != null && !Array.isArray(transferList)) { throw new TypeError('transferList argument must be an array') } + transferList = + transferList != null && transferList.length > 0 + ? Array.from(transferList) + : undefined const tasks: [Data, AbortSignal | undefined][] = Array.from( { length: (data as Data[]).length }, (_, i) => [ @@ -784,39 +836,61 @@ export abstract class AbstractPool< /** @inheritDoc */ public async removeTaskFunction (name: string): Promise { - if (!this.taskFunctions.has(name)) { - throw new Error( - 'Cannot remove a task function not handled on the pool side' - ) - } - const opResult = await this.sendTaskFunctionOperationToWorkers({ - taskFunctionOperation: 'remove', - taskFunctionProperties: buildTaskFunctionProperties( - name, - this.taskFunctions.get(name) - ), + return await this.queueTaskFunctionOperation(async () => { + if (!this.taskFunctions.has(name)) { + throw new Error( + 'Cannot remove a task function not handled on the pool side' + ) + } + while (this.workerTaskFunctionBootstraps.size > 0) { + await this.waitForWorkerTaskFunctionBootstraps() + } + this.beginTaskFunctionOperation() + try { + const opResult = await this.sendTaskFunctionOperationToWorkers({ + taskFunctionOperation: 'remove', + taskFunctionProperties: this.taskFunctions.get(name) + ?.taskFunctionProperties ?? { name }, + }) + for (const workerNode of this.workerNodes) { + workerNode.deleteTaskFunctionWorkerUsage(name) + } + this.taskFunctions.delete(name) + if (this.defaultTaskFunctionName === name) { + this.defaultTaskFunctionName = undefined + } + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + this.getWorkerChoiceStrategies() + ) + for (const workerNodeKey of this.workerNodes.keys()) { + this.sendStatisticsMessageToWorker(workerNodeKey) + } + return opResult + } finally { + this.endTaskFunctionOperation() + } }) - for (const workerNode of this.workerNodes) { - workerNode.deleteTaskFunctionWorkerUsage(name) - } - this.taskFunctions.delete(name) - this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( - this.getWorkerChoiceStrategies() - ) - for (const workerNodeKey of this.workerNodes.keys()) { - this.sendStatisticsMessageToWorker(workerNodeKey) - } - return opResult } /** @inheritDoc */ public async setDefaultTaskFunction (name: string): Promise { - return await this.sendTaskFunctionOperationToWorkers({ - taskFunctionOperation: 'default', - taskFunctionProperties: buildTaskFunctionProperties( - name, - this.taskFunctions.get(name) - ), + return await this.queueTaskFunctionOperation(async () => { + while (this.workerTaskFunctionBootstraps.size > 0) { + await this.waitForWorkerTaskFunctionBootstraps() + } + this.beginTaskFunctionOperation() + try { + const hadTargets = this.workerNodes.length > 0 + const opResult = await this.sendTaskFunctionOperationToWorkers({ + taskFunctionOperation: 'default', + taskFunctionProperties: this.taskFunctions.get(name) + ?.taskFunctionProperties ?? { name }, + }) + if (hadTargets) this.defaultTaskFunctionName = name + return opResult + } finally { + this.endTaskFunctionOperation() + } }) } @@ -990,7 +1064,7 @@ export abstract class AbstractPool< */ protected afterWorkerNodeSetup (workerNodeKey: number): void { // Listen to worker messages. - this.registerWorkerMessageListener( + this.registerWorkerLifecycleMessageListener( workerNodeKey, this.workerMessageListener ) @@ -1073,8 +1147,13 @@ export abstract class AbstractPool< */ protected createAndSetupDynamicWorkerNode (): number { const workerNodeKey = this.createAndSetupWorkerNode() - this.registerWorkerMessageListener(workerNodeKey, message => { - if (this.destroying) { + this.registerWorkerLifecycleMessageListener(workerNodeKey, message => { + if ( + this.destroying || + message.kill === 'success' || + message.kill === 'failure' || + message.taskFunctionOperationStatus != null + ) { return } this.checkMessageWorkerId(message) @@ -1096,25 +1175,12 @@ export abstract class AbstractPool< this.sendToWorker(workerNodeKey, { checkActive: true, }) - if (this.taskFunctions.size > 0) { - for (const [taskFunctionName, taskFunctionObject] of this.taskFunctions) { - this.sendTaskFunctionOperationToWorker(workerNodeKey, { - taskFunction: taskFunctionObject.taskFunction.toString(), - taskFunctionOperation: 'add', - taskFunctionProperties: buildTaskFunctionProperties( - taskFunctionName, - taskFunctionObject - ), - }).catch((error: unknown) => { - this.emitter?.emit(PoolEvents.error, error) - }) - } - } const workerNode = this.workerNodes[workerNodeKey] workerNode.info.dynamic = true if ( this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerReady === - true + true && + !this.workerTaskFunctionBootstraps.has(workerNode) ) { workerNode.info.ready = true } @@ -1182,6 +1248,13 @@ export abstract class AbstractPool< if (settlingTasks) { taskResponses = this.captureTaskResponses(workerNode, taskResponses) } + const operationExitListeners = + this.workerTaskFunctionOperationExitListeners.get(workerNode) + if (operationExitListeners != null) { + this.workerTaskFunctionOperationExitListeners.delete(workerNode) + // Settle in dispatch order before removal or fallible replacement setup. + for (const listener of operationExitListeners) listener() + } workerNode.info.ready = false let workerNodeTasks: | Map> @@ -1236,6 +1309,24 @@ export abstract class AbstractPool< } }) const workerNodeKey = this.addWorkerNode(workerNode) + if ( + this.pendingTaskFunctionOperations > 0 || + this.taskFunctions.size > 0 || + this.defaultTaskFunctionName != null + ) { + const bootstrap: { promise?: Promise; ready: boolean } = { + ready: false, + } + workerNode.info.ready = false + this.workerTaskFunctionBootstraps.set(workerNode, bootstrap) + bootstrap.promise = this.replayTaskFunctionsAfterOperations( + workerNode, + bootstrap + ) + bootstrap.promise.catch((error: unknown) => { + this.emitter?.emit(PoolEvents.error, error) + }) + } this.afterWorkerNodeSetup(workerNodeKey) return workerNodeKey } @@ -1257,10 +1348,10 @@ export abstract class AbstractPool< * @param workerNodeKey - The worker node key. */ protected async destroyWorkerNode (workerNodeKey: number): Promise { - this.terminatingWorkerNodes.add(this.workerNodes[workerNodeKey]) + const workerNode = this.workerNodes[workerNodeKey] + this.terminatingWorkerNodes.add(workerNode) this.flagWorkerNodeAsNotReady(workerNodeKey) const flushedTasks = this.flushTasksQueue(workerNodeKey) - const workerNode = this.workerNodes[workerNodeKey] await waitWorkerNodeEvents( workerNode, 'taskFinished', @@ -1271,8 +1362,14 @@ export abstract class AbstractPool< ).tasksFinishedTimeout, false ) - await this.sendKillMessageToWorker(workerNodeKey) - await workerNode.terminate() + try { + const currentWorkerNodeKey = this.workerNodes.indexOf(workerNode) + if (currentWorkerNodeKey !== -1) { + await this.sendKillMessageToWorker(currentWorkerNodeKey) + } + } finally { + await workerNode.terminate() + } } protected flagWorkerNodeAsNotReady (workerNodeKey: number): void { @@ -1410,8 +1507,8 @@ export abstract class AbstractPool< if (this.destroying && workerReadyMessage) { return } - // Kill messages responses are handled in dedicated listeners - if (kill != null) { + // Kill and task-function operation responses have dedicated listeners. + if (kill != null || message.taskFunctionOperationStatus != null) { return } this.checkMessageWorkerId(message) @@ -1438,12 +1535,14 @@ export abstract class AbstractPool< return } const { taskId, workerId } = eventDetail - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const promiseResponse = this.promiseResponseMap.get(taskId!) + if (taskId == null) { + return + } + const promiseResponse = this.promiseResponseMap.get(taskId) if (promiseResponse == null) { return } - const { abortSignal, reject } = promiseResponse + const { abortSignal } = promiseResponse if (abortSignal?.aborted === false) { return } @@ -1458,11 +1557,13 @@ export abstract class AbstractPool< if (taskId === task.taskId && abortable === true) { workerNode.info.queuedTaskAbortion = true workerNode.deleteTask(task) - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.promiseResponseMap.delete(taskId!) workerNode.info.queuedTaskAbortion = false - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - reject(this.getAbortError(name!, taskId!)) + + this.rejectTask( + taskId, + promiseResponse, + this.getAbortError(name ?? DEFAULT_TASK_NAME, taskId, abortSignal) + ) return } } @@ -1485,6 +1586,14 @@ export abstract class AbstractPool< return workerNodeKey } + private beginTaskFunctionOperation (): void { + if (this.pendingTaskFunctionOperations++ === 0) { + const { promise, resolve } = Promise.withResolvers() + this.taskFunctionOperationsFinished = promise + this.resolveTaskFunctionOperations = resolve + } + } + private buildTasksQueueOptions ( tasksQueueOptions: TasksQueueOptions | undefined ): TasksQueueOptions { @@ -1520,6 +1629,8 @@ export abstract class AbstractPool< if (message.taskId != null) { this.checkMessageWorkerId(message) ;(taskResponses ??= []).push(message) + } else if (message.taskFunctionOperationStatus != null) { + workerNode.messageChannel.port1.emit('message', message) } } return taskResponses @@ -1684,9 +1795,9 @@ export abstract class AbstractPool< /** * Chooses a worker node for the next task. * @param name - The task function name. - * @returns The chosen worker node key. + * @returns The chosen worker node key, or `undefined` while runtime task functions bootstrap. */ - private chooseWorkerNode (name?: string): number { + private chooseWorkerNode (name?: string): number | undefined { const workerNodeKeys = this.getTaskFunctionWorkerNodeKeys(name) if (workerNodeKeys != null) { const maxPoolSize = @@ -1704,18 +1815,59 @@ export abstract class AbstractPool< const workerNodeKey = this.createAndSetupDynamicWorkerNode() if ( this.workerChoiceStrategiesContext?.getPolicy().dynamicWorkerUsage === - true + true && + !this.workerTaskFunctionBootstraps.has(this.workerNodes[workerNodeKey]) ) { return workerNodeKey } } + const eligibleWorkerNodeKeys = + workerNodeKeys != null ? new Set(workerNodeKeys) : undefined + if (this.workerTaskFunctionBootstraps.size > 0) { + let readyWorkerFound = false + for ( + let workerNodeKey = 0; + workerNodeKey < this.workerNodes.length; + ++workerNodeKey + ) { + if ( + this.workerNodes[workerNodeKey].info.ready && + (eligibleWorkerNodeKeys == null || + eligibleWorkerNodeKeys.has(workerNodeKey)) + ) { + readyWorkerFound = true + break + } + } + if (!readyWorkerFound) return + } // eslint-disable-next-line @typescript-eslint/no-non-null-assertion return this.workerChoiceStrategiesContext!.execute( this.getTaskFunctionWorkerChoiceStrategy(name), - workerNodeKeys != null ? new Set(workerNodeKeys) : undefined + eligibleWorkerNodeKeys ) } + private createWorkerMessageListenerCleanup ( + workerNode: IWorkerNode, + listener: (message: MessageValue) => void + ): () => void { + const messageTarget = workerNode.messageChannel?.port1 ?? workerNode.worker + let cleaned = false + return () => { + if (cleaned) return + cleaned = true + try { + const workerNodeKey = this.workerNodes.indexOf(workerNode) + if (workerNodeKey !== -1) { + this.deregisterWorkerMessageListener(workerNodeKey, listener) + } + } finally { + messageTarget.off('message', listener) + } + } + } + /** * Creates a worker node. * @returns The created worker node. @@ -1753,6 +1905,29 @@ export abstract class AbstractPool< return task } + private deregisterTaskFunctionOperationExitListener ( + workerNode: IWorkerNode, + listener: () => void + ): void { + const listeners = + this.workerTaskFunctionOperationExitListeners.get(workerNode) + if (listeners != null) { + listeners.delete(listener) + if (listeners.size === 0) { + this.workerTaskFunctionOperationExitListeners.delete(workerNode) + } + } + } + + private endTaskFunctionOperation (): void { + if (--this.pendingTaskFunctionOperations === 0) { + const resolve = this.resolveTaskFunctionOperations + delete this.taskFunctionOperationsFinished + delete this.resolveTaskFunctionOperations + resolve?.(undefined) + } + } + private enqueueTask (workerNodeKey: number, task: Task): number { this.updateTaskWorkerNodeKey(workerNodeKey, task) const tasksQueueSize = this.workerNodes[workerNodeKey].enqueueTask(task) @@ -1806,14 +1981,17 @@ export abstract class AbstractPool< } } - private readonly getAbortError = (taskName: string, taskId: UUID): Error => { - const abortError = this.promiseResponseMap.get(taskId)?.abortSignal - ?.reason as Error | string + private readonly getAbortError = ( + taskName: string, + taskId: undefined | UUID, + abortSignal?: AbortSignal + ): Error => { + const abortError = abortSignal?.reason as Error | string return abortError instanceof Error ? abortError : typeof abortError === 'string' ? new Error(abortError) - : new Error(`Task '${taskName}' id '${taskId}' aborted`) + : new Error(`Task '${taskName}' id '${taskId ?? randomUUID()}' aborted`) } private getQueuedTasks (): number { @@ -1832,6 +2010,18 @@ export abstract class AbstractPool< ) } + /** + * Builds the error for a task function operation acknowledgement without a UUID. + * @param workerId - The worker id that sent the acknowledgement. + * @returns The task function operation acknowledgement error. + */ + private readonly getTaskFunctionOperationAcknowledgementError = ( + workerId?: number + ): Error => + new Error( + `Task function operation acknowledgement from worker ${String(workerId)} is missing taskFunctionOperationId` + ) + /** * Gets task function worker choice strategy, if any. * @param name - The task function name. @@ -1876,31 +2066,35 @@ export abstract class AbstractPool< return undefined } - private getTasksQueuePriority (): boolean { - return this.listTaskFunctionsProperties().some( + private getTasksQueuePriority ( + taskFunctionsProperties: readonly TaskFunctionProperties[] = this.listTaskFunctionsProperties() + ): boolean { + return taskFunctionsProperties.some( taskFunctionProperties => taskFunctionProperties.priority != null ) } /** * Gets the worker choice strategies registered in this pool. + * @param taskFunctionsProperties - Task function properties used to collect worker choice strategies. * @returns The worker choice strategies. */ - private readonly getWorkerChoiceStrategies = - (): Set => { - return new Set([ - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.opts.workerChoiceStrategy!, - ...this.listTaskFunctionsProperties() - .map( - (taskFunctionProperties: TaskFunctionProperties) => - taskFunctionProperties.strategy - ) - .filter( - (strategy: undefined | WorkerChoiceStrategy) => strategy != null - ), - ]) - } + private readonly getWorkerChoiceStrategies = ( + taskFunctionsProperties: readonly TaskFunctionProperties[] = this.listTaskFunctionsProperties() + ): Set => { + return new Set([ + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.opts.workerChoiceStrategy!, + ...taskFunctionsProperties + .map( + (taskFunctionProperties: TaskFunctionProperties) => + taskFunctionProperties.strategy + ) + .filter( + (strategy: undefined | WorkerChoiceStrategy) => strategy != null + ), + ]) + } /** * Gets the worker node key given its worker id. @@ -1913,6 +2107,15 @@ export abstract class AbstractPool< ) } + private getWorkerNodeKeyForBootstrap ( + workerNode: IWorkerNode + ): number { + if (this.destroying || this.terminatingWorkerNodes.has(workerNode)) { + return -1 + } + return this.workerNodes.indexOf(workerNode) + } + /** * Gets worker node task function priority, if any. * @param workerNodeKey - The worker node key. @@ -1970,11 +2173,18 @@ export abstract class AbstractPool< } private handleTaskExecutionResponse (message: MessageValue): void { - const { data, taskId, workerError } = message + const { data, taskId } = message + let { workerError } = message // eslint-disable-next-line @typescript-eslint/no-non-null-assertion const promiseResponse = this.promiseResponseMap.get(taskId!) if (promiseResponse != null) { const { asyncResource, reject, resolve, workerNodeKey } = promiseResponse + if ( + workerError != null && + Object.hasOwn(promiseResponse, 'originalData') + ) { + workerError = { ...workerError, data: promiseResponse.originalData } + } const workerNode = this.workerNodes[workerNodeKey] let error: Error | undefined if (workerError != null) { @@ -2061,8 +2271,11 @@ export abstract class AbstractPool< ): Error => { const { aborted, error, message, name, stack } = workerError if (aborted) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - return this.getAbortError(name!, taskId) + return this.getAbortError( + name ?? DEFAULT_TASK_NAME, + taskId, + this.promiseResponseMap.get(taskId)?.abortSignal + ) } if (error != null) { return error @@ -2195,7 +2408,9 @@ export abstract class AbstractPool< } const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) const workerNode = this.workerNodes[workerNodeKey] - workerNode.info.ready = ready + const bootstrap = this.workerTaskFunctionBootstraps.get(workerNode) + if (bootstrap != null) bootstrap.ready = ready + workerNode.info.ready = bootstrap == null workerNode.info.taskFunctionsProperties = taskFunctionsProperties this.sendStatisticsMessageToWorker(workerNodeKey) this.setTasksQueuePriority(workerNodeKey) @@ -2247,9 +2462,22 @@ export abstract class AbstractPool< abortSignal?: AbortSignal, transferList?: readonly Transferable[] ): Promise { + this.throwIfTaskAborted(name, abortSignal) + const timestamp = performance.now() + let chosenWorkerNodeKey = this.chooseWorkerNode(name) + while (chosenWorkerNodeKey == null) { + await this.waitForWorkerTaskFunctionBootstraps(abortSignal, name, 'first') + this.throwIfTaskAborted(name, abortSignal) + if (!this.started) { + throw new Error('Cannot execute a task on not started pool') + } + if (this.destroying) { + throw new Error('Cannot execute a task on destroying pool') + } + chosenWorkerNodeKey = this.chooseWorkerNode(name) + } + const workerNodeKey = chosenWorkerNodeKey return await new Promise((resolve, reject) => { - const timestamp = performance.now() - const workerNodeKey = this.chooseWorkerNode(name) const task: Task = { abortable: abortSignal != null, data: data ?? ({} as Data), @@ -2275,6 +2503,8 @@ export abstract class AbstractPool< }), }), abortSignal, + ...(transferList != null && + transferList.length > 0 && { originalData: task.data }), } // eslint-disable-next-line @typescript-eslint/no-non-null-assertion this.promiseResponseMap.set(task.taskId!, promiseResponse) @@ -2375,6 +2605,19 @@ export abstract class AbstractPool< ) } + private queueTaskFunctionOperation ( + operation: () => Promise + ): Promise { + const result = (this.taskFunctionOperationsQueue ?? Promise.resolve()).then( + operation + ) + this.taskFunctionOperationsQueue = result.then( + EMPTY_FUNCTION, + EMPTY_FUNCTION + ) + return result + } + private redistributeQueuedTasks ( sourceWorkerNode: IWorkerNode ): void { @@ -2393,7 +2636,11 @@ export abstract class AbstractPool< this.rejectTask( taskId, promiseResponse, - this.getAbortError(task.name ?? DEFAULT_TASK_NAME, taskId) + this.getAbortError( + task.name ?? DEFAULT_TASK_NAME, + taskId, + promiseResponse.abortSignal + ) ) continue } @@ -2442,6 +2689,36 @@ export abstract class AbstractPool< } } + private registerTaskFunctionOperationExitListener ( + workerNode: IWorkerNode, + listener: () => void + ): void { + let listeners = + this.workerTaskFunctionOperationExitListeners.get(workerNode) + if (listeners == null) { + listeners = new Set() + this.workerTaskFunctionOperationExitListeners.set(workerNode, listeners) + } + listeners.add(listener) + } + + private registerWorkerLifecycleMessageListener ( + workerNodeKey: number, + listener: (message: MessageValue) => void + ): void { + const workerNode = this.workerNodes[workerNodeKey] + const cleanup = this.createWorkerMessageListenerCleanup( + workerNode, + listener + ) + // Retire lifecycle consumers without discarding buffered kill acknowledgements. + workerNode.worker.prependOnceListener('exit', () => { + workerNode.info.ready = false + cleanup() + }) + this.registerWorkerMessageListener(workerNodeKey, listener) + } + private rejectTask ( taskId: UUID, promiseResponse: PromiseResponseWrapper, @@ -2486,6 +2763,7 @@ export abstract class AbstractPool< * @param workerNode - The worker node. */ private removeWorkerNode (workerNode: IWorkerNode): void { + this.workerTaskFunctionBootstraps.delete(workerNode) const workerNodeKey = this.workerNodes.indexOf(workerNode) if (workerNodeKey !== -1) { this.workerNodes.splice(workerNodeKey, 1) @@ -2502,6 +2780,77 @@ export abstract class AbstractPool< } } + private async replayTaskFunctions (workerNodeKey: number): Promise { + if (this.taskFunctions.size === 0) return + const workerNode = this.workerNodes[workerNodeKey] + for (const message of this.taskFunctions.values()) { + const currentWorkerNodeKey = this.getWorkerNodeKeyForBootstrap(workerNode) + if (currentWorkerNodeKey === -1) return + await this.sendTaskFunctionOperationToWorker( + currentWorkerNodeKey, + message + ) + } + } + + private async replayTaskFunctionsAfterOperations ( + workerNode: IWorkerNode, + bootstrap: { ready: boolean } + ): Promise { + try { + // Publish the bootstrap promise before sending replay requests. + do { + await this.taskFunctionOperationsFinished + } while (this.taskFunctionOperationsFinished != null) + let workerNodeKey = this.getWorkerNodeKeyForBootstrap(workerNode) + if (workerNodeKey === -1) return + await this.replayTaskFunctions(workerNodeKey) + workerNodeKey = this.getWorkerNodeKeyForBootstrap(workerNode) + if (workerNodeKey === -1) return + if (this.defaultTaskFunctionName != null) { + await this.sendTaskFunctionOperationToWorker(workerNodeKey, { + taskFunctionOperation: 'default', + taskFunctionProperties: this.taskFunctions.get( + this.defaultTaskFunctionName + )?.taskFunctionProperties ?? { name: this.defaultTaskFunctionName }, + }) + workerNodeKey = this.getWorkerNodeKeyForBootstrap(workerNode) + if (workerNodeKey === -1) return + } + // Keep strategies needed by ready peers while this worker becomes selectable. + const workerChoiceStrategies = this.getWorkerChoiceStrategies( + workerNode.info.taskFunctionsProperties ?? [] + ) + for (const readyWorkerNode of this.workerNodes) { + if ( + readyWorkerNode === workerNode || + !readyWorkerNode.info.ready || + this.terminatingWorkerNodes.has(readyWorkerNode) + ) { + continue + } + const taskFunctionsProperties = + readyWorkerNode.info.taskFunctionsProperties + if (taskFunctionsProperties != null) { + for (const { strategy } of taskFunctionsProperties) { + if (strategy != null) workerChoiceStrategies.add(strategy) + } + } + } + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + workerChoiceStrategies + ) + this.sendStatisticsMessageToWorker(workerNodeKey) + workerNode.info.ready = bootstrap.ready + this.checkAndEmitReadyEvent() + } catch (error) { + // Retired bootstraps no longer block waiting submissions. + if (this.workerNodes.includes(workerNode)) throw error + } finally { + this.workerTaskFunctionBootstraps.delete(workerNode) + } + } + private resetTaskSequentiallyStolenStatisticsWorkerUsage ( workerNodeKey: number, taskName?: string @@ -2529,6 +2878,7 @@ export abstract class AbstractPool< workerNodeKey: number, timeout = 1000 ): Promise { + const workerId = this.getWorkerInfo(workerNodeKey)?.id let timeoutHandle: NodeJS.Timeout | undefined let killMessageListener: | ((message: MessageValue) => void) @@ -2542,12 +2892,7 @@ export abstract class AbstractPool< }, timeout) : undefined killMessageListener = (message: MessageValue): void => { - if ( - this.workerNodes.length === 0 || - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - this.workerNodes[workerNodeKey] == null - ) { - resolve() + if (message.workerId !== workerId) { return } if (message.kill === 'success') { @@ -2568,8 +2913,12 @@ export abstract class AbstractPool< if (timeoutHandle != null) { clearTimeout(timeoutHandle) } - if (killMessageListener != null) { - this.deregisterWorkerMessageListener(workerNodeKey, killMessageListener) + const currentWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + if (killMessageListener != null && currentWorkerNodeKey !== -1) { + this.deregisterWorkerMessageListener( + currentWorkerNodeKey, + killMessageListener + ) } } } @@ -2593,117 +2942,182 @@ export abstract class AbstractPool< workerNodeKey: number, message: MessageValue ): Promise { - let taskFunctionOperationListener: - | ((message: MessageValue) => void) - | undefined - try { - return await new Promise((resolve, reject) => { - taskFunctionOperationListener = ( - message: MessageValue - ): void => { - this.checkMessageWorkerId(message) - const workerId = this.getWorkerInfo(workerNodeKey)?.id - if ( - message.taskFunctionOperationStatus != null && - message.workerId === workerId - ) { - if (message.taskFunctionOperationStatus) { - resolve(true) - return - } - reject( - new Error( - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions - `Task function operation '${message.taskFunctionOperation?.toString()}' failed on worker ${message.workerId?.toString()} with error: '${ - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions - message.workerError?.message - }'` - ) - ) - } + const workerNode = this.workerNodes[workerNodeKey] + const workerId = this.getWorkerInfo(workerNodeKey)?.id + const taskFunctionOperationId = randomUUID() + const { promise, reject, resolve } = Promise.withResolvers() + const listener = (message: MessageValue): void => { + if ( + message.taskFunctionOperationStatus != null && + message.workerId === workerId && + (message.taskFunctionOperationId == null || + message.taskFunctionOperationId === taskFunctionOperationId) + ) { + this.checkMessageWorkerId(message) + if (message.taskFunctionOperationId == null) { + reject( + this.getTaskFunctionOperationAcknowledgementError(message.workerId) + ) + return } - this.registerWorkerMessageListener( - workerNodeKey, - taskFunctionOperationListener + if (message.taskFunctionOperationStatus) { + resolve(true) + return + } + reject( + new Error( + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + `Task function operation '${message.taskFunctionOperation?.toString()}' failed on worker ${message.workerId?.toString()} with error: '${ + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + message.workerError?.message + }'` + ) + ) + } + } + const cleanup = this.createWorkerMessageListenerCleanup( + workerNode, + listener + ) + const exitListener = (): void => { + cleanup() + reject( + new Error( + `Worker ${String(workerId)} exited before acknowledging task function operation '${message.taskFunctionOperation ?? 'unknown'}'` ) - this.sendToWorker(workerNodeKey, message) + ) + } + try { + this.registerTaskFunctionOperationExitListener(workerNode, exitListener) + this.registerWorkerMessageListener(workerNodeKey, listener) + this.sendToWorker(workerNodeKey, { + ...message, + taskFunctionOperationId, }) + } catch (error) { + reject(error) + } + try { + return await promise } finally { - if (taskFunctionOperationListener != null) { - this.deregisterWorkerMessageListener( - workerNodeKey, - taskFunctionOperationListener - ) - } + this.deregisterTaskFunctionOperationExitListener(workerNode, exitListener) + cleanup() } } private async sendTaskFunctionOperationToWorkers ( message: MessageValue ): Promise { - const targetWorkerNodeCount = this.workerNodes.length - if (targetWorkerNodeCount === 0) { + if (this.workerNodes.length === 0) { return true } - const responsesReceived: MessageValue[] = [] - const taskFunctionOperationsListener = ( - message: MessageValue, - resolve: (value: boolean | PromiseLike) => void, - reject: (reason?: unknown) => void - ): void => { - this.checkMessageWorkerId(message) - const workerNodeKey = this.getWorkerNodeKeyByWorkerId(message.workerId) + const taskFunctionOperationId = randomUUID() + const responsesReceived = new Map< + number | undefined, + MessageValue | undefined + >() + for (const workerNode of this.workerNodes) { + responsesReceived.set(workerNode.info.id, undefined) + } + let remainingResponses = responsesReceived.size + let errorResponse: MessageValue | undefined + let protocolError: Error | undefined + const { promise, reject, resolve } = Promise.withResolvers() + const listeners: { + cleanup: () => void + exitListener: () => void + workerNode: IWorkerNode + }[] = [] + const complete = (): void => { + if (remainingResponses !== 0) return + if (protocolError != null) { + reject(protocolError) + } else if (errorResponse != null) { + reject( + new Error( + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + `Task function operation '${message.taskFunctionOperation as string}' failed on worker ${errorResponse.workerId?.toString()} with error: '${errorResponse.workerError?.message ?? 'Unknown error'}'` + ) + ) + } else if (responsesReceived.size === 0) { + reject( + new Error( + `All targets exited before acknowledging task function operation '${message.taskFunctionOperation ?? 'unknown'}'` + ) + ) + } else { + resolve(true) + } + } + const listener = (message: MessageValue): void => { if ( message.taskFunctionOperationStatus != null && - workerNodeKey >= 0 && - workerNodeKey < targetWorkerNodeCount + (message.taskFunctionOperationId == null || + message.taskFunctionOperationId === taskFunctionOperationId) && + responsesReceived.has(message.workerId) && + responsesReceived.get(message.workerId) == null ) { - responsesReceived.push(message) - if (responsesReceived.length >= targetWorkerNodeCount) { + this.checkMessageWorkerId(message) + responsesReceived.set(message.workerId, message) + --remainingResponses + if (message.taskFunctionOperationId == null) { + protocolError ??= this.getTaskFunctionOperationAcknowledgementError( + message.workerId + ) + } else if (!message.taskFunctionOperationStatus) { + errorResponse ??= message + } + complete() + } + } + try { + for (const workerId of responsesReceived.keys()) { + const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + const workerNode = this.workerNodes[workerNodeKey] + const cleanup = this.createWorkerMessageListenerCleanup( + workerNode, + listener + ) + const exitListener = (): void => { + cleanup() if ( - responsesReceived.every( - msg => msg.taskFunctionOperationStatus === true - ) + responsesReceived.has(workerId) && + responsesReceived.get(workerId) == null ) { - resolve(true) - } else { - const errorResponse = responsesReceived.find( - msg => msg.taskFunctionOperationStatus === false - ) - reject( - new Error( - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions - `Task function operation '${message.taskFunctionOperation as string}' failed on worker ${errorResponse?.workerId?.toString()} with error: '${errorResponse?.workerError?.error?.message ?? 'Unknown error'}'` - ) - ) + responsesReceived.delete(workerId) + --remainingResponses + complete() } } + listeners.push({ cleanup, exitListener, workerNode }) + this.registerTaskFunctionOperationExitListener(workerNode, exitListener) + this.registerWorkerMessageListener(workerNodeKey, listener) + this.sendToWorker(workerNodeKey, { + ...message, + taskFunctionOperationId, + }) } + } catch (error) { + reject(error) } - let listener: ((message: MessageValue) => void) | undefined - const workerNodeKeys = [...this.workerNodes.keys()] try { - return await new Promise((resolve, reject) => { - listener = (message: MessageValue) => { - taskFunctionOperationsListener(message, resolve, reject) - } - for (const workerNodeKey of workerNodeKeys) { - this.registerWorkerMessageListener(workerNodeKey, listener) - this.sendToWorker(workerNodeKey, message) - } - }) + return await promise } finally { - if (listener != null) { - for (const workerNodeKey of workerNodeKeys) { - this.deregisterWorkerMessageListener(workerNodeKey, listener) - } + for (const { cleanup, exitListener, workerNode } of listeners) { + this.deregisterTaskFunctionOperationExitListener( + workerNode, + exitListener + ) + cleanup() } } } private setTasksQueuePriority (workerNodeKey: number): void { this.workerNodes[workerNodeKey].setTasksQueuePriority( - this.getTasksQueuePriority() + this.getTasksQueuePriority( + this.workerNodes[workerNodeKey].info.taskFunctionsProperties + ) ) } @@ -2829,6 +3243,16 @@ export abstract class AbstractPool< return workerNode.tasksQueueSize() } + private throwIfTaskAborted (name?: string, abortSignal?: AbortSignal): void { + if (abortSignal?.aborted === true) { + throw this.getAbortError( + name ?? DEFAULT_TASK_NAME, + undefined, + abortSignal + ) + } + } + private unsetTasksStealingOnBackPressure (): void { for (const workerNodeKey of this.workerNodes.keys()) { this.workerNodes[workerNodeKey].off( @@ -2917,6 +3341,83 @@ export abstract class AbstractPool< } } + private async waitForWorkerTaskFunctionBootstraps ( + abortSignal?: AbortSignal, + name?: string, + waitMode: 'all' | 'first' = 'all' + ): Promise { + let abortPromise: Promise | undefined + let abortListener: (() => void) | undefined + if (abortSignal != null) { + this.throwIfTaskAborted(name, abortSignal) + const { promise, reject } = Promise.withResolvers() + abortPromise = promise + abortListener = () => { + reject( + this.getAbortError(name ?? DEFAULT_TASK_NAME, undefined, abortSignal) + ) + } + abortSignal.addEventListener('abort', abortListener, { once: true }) + } + try { + while (this.workerTaskFunctionBootstraps.size > 0) { + if (waitMode === 'first') { + const promises: Promise[] = [] + for (const [workerNode, bootstrap] of this + .workerTaskFunctionBootstraps) { + if (bootstrap.promise != null) { + promises.push( + bootstrap.promise.catch((error: unknown) => { + const workerNodeKey = this.workerNodes.indexOf(workerNode) + if (workerNodeKey === -1) return + const taskFunctionsProperties = + this.listTaskFunctionsProperties() + const taskFunctionName = + name == null || name === DEFAULT_TASK_NAME + ? (this.defaultTaskFunctionName ?? + taskFunctionsProperties[1]?.name) + : name + const registration = this.taskFunctions.get(taskFunctionName) + const workerNodeKeys = + registration != null + ? registration.taskFunctionProperties.workerNodeKeys + : this.getTaskFunctionWorkerNodeKeys( + taskFunctionName, + taskFunctionsProperties + ) + // Ineligible failures still wake selection without rejecting the task. + if ( + workerNodeKeys == null || + workerNodeKeys.includes(workerNodeKey) + ) { + throw error + } + }) + ) + } + } + if (promises.length === 0) return + if (abortPromise != null) promises.push(abortPromise) + await Promise.race(promises) + return + } + for (const bootstrap of this.workerTaskFunctionBootstraps.values()) { + if (bootstrap.promise != null) { + if (abortPromise != null) { + await Promise.race([bootstrap.promise, abortPromise]) + } else { + await bootstrap.promise + } + } + } + } + } finally { + if (abortListener != null) { + abortSignal?.removeEventListener('abort', abortListener) + } + } + } + private readonly workerNodeStealTask = ( workerNodeKey: number ): Task | undefined => { diff --git a/src/pools/pool.ts b/src/pools/pool.ts index cef766180..2de13f2a5 100644 --- a/src/pools/pool.ts +++ b/src/pools/pool.ts @@ -132,7 +132,7 @@ export interface IPool< * Executes the specified function in the worker constructor with the task data input parameter. * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data. * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed. - * @param abortSignal - The optional AbortSignal to abort the task. + * @param abortSignal - The optional AbortSignal to abort the task. An already-aborted signal prevents submission. * @param transferList - The optional array of transferable objects to transfer ownership of. Ownership of the transferred objects is given to the chosen pool's worker_threads worker and they should not be used in the main thread afterwards. * @returns Promise with a task function response that will be fulfilled when the task is completed. */ @@ -161,7 +161,7 @@ export interface IPool< * Executes the specified function in the worker constructor with the tasks data iterable input parameter. * @param data - The tasks iterable input data for the specified task function. This can only be an iterable of structured-cloneable data. * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed. - * @param abortSignals - The optional iterable of AbortSignal to abort the tasks iterable. + * @param abortSignals - The optional AbortSignal iterable matching the input order and length. * @param transferList - The optional array of transferable objects to transfer ownership of. Ownership of the transferred objects is given to the chosen pool's worker_threads worker and they should not be used in the main thread afterwards. * @returns Promise with an array of task function responses that will be fulfilled when the tasks are completed. */ diff --git a/src/pools/selection-strategies/selection-strategies-utils.ts b/src/pools/selection-strategies/selection-strategies-utils.ts index 14f708a23..4e496c797 100644 --- a/src/pools/selection-strategies/selection-strategies-utils.ts +++ b/src/pools/selection-strategies/selection-strategies-utils.ts @@ -76,9 +76,10 @@ export const getWorkerChoiceStrategiesRetries = < pool: IPool, opts?: WorkerChoiceStrategyOptions ): number => { + const weights = opts?.weights return ( pool.info.maxSize + - Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length + (weights != null ? Object.keys(weights).length : pool.info.maxSize) ) } diff --git a/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts b/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts index 3ed8bc3c4..576ae22b9 100644 --- a/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts @@ -68,6 +68,8 @@ export class WeightedRoundRobinWorkerChoiceStrategy< return undefined } if (!this.isWorkerNodeEligible(chosenWorkerNodeKey, workerNodeKeysSet)) { + this.nextWorkerNodeKey = this.getRoundRobinNextWorkerNodeKey() + this.workerNodeVirtualTaskExecutionTime = 0 return undefined } return this.checkWorkerNodeKey(chosenWorkerNodeKey) diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index 8e9ad71ac..6c304c932 100644 --- a/src/pools/worker-node.ts +++ b/src/pools/worker-node.ts @@ -173,31 +173,35 @@ export class WorkerNode public async terminate (): Promise { this.closeMessageChannel() this.removeAllListeners() - switch (this.info.type) { - case WorkerTypes.cluster: { - if (this.worker.isDead?.() === true) { - break - } - const waitWorkerExit = new Promise(resolve => { - this.registerOnceWorkerEventHandler('exit', () => { - resolve() + try { + switch (this.info.type) { + case WorkerTypes.cluster: { + if (this.worker.isDead?.() === true) { + break + } + const waitWorkerExit = new Promise(resolve => { + this.registerOnceWorkerEventHandler('exit', () => { + resolve() + }) }) - }) - if (this.worker.isConnected?.() === true) { - this.registerOnceWorkerEventHandler('disconnect', () => { + if (this.worker.isConnected?.() === true) { + this.registerOnceWorkerEventHandler('disconnect', () => { + this.worker.kill?.() + }) + this.worker.disconnect?.() + } else { this.worker.kill?.() - }) - this.worker.disconnect?.() - } else { - this.worker.kill?.() + } + await waitWorkerExit + break } - await waitWorkerExit - break + case WorkerTypes.thread: + this.worker.unref?.() + await this.worker.terminate?.() + break } - case WorkerTypes.thread: - this.worker.unref?.() - await this.worker.terminate?.() - break + } finally { + delete this.messageChannel } this.worker.removeAllListeners() } @@ -208,7 +212,6 @@ export class WorkerNode this.messageChannel.port2.unref() this.messageChannel.port1.close() this.messageChannel.port2.close() - delete this.messageChannel } } diff --git a/src/utility-types.ts b/src/utility-types.ts index 52bec3bb2..e1bb84084 100644 --- a/src/utility-types.ts +++ b/src/utility-types.ts @@ -45,6 +45,10 @@ export interface MessageValue * - `'default'` - Set a task function as default. */ readonly taskFunctionOperation?: 'add' | 'default' | 'remove' + /** + * Task function operation UUID. + */ + readonly taskFunctionOperationId?: UUID /** * Whether the task function operation is successful or not. */ @@ -90,6 +94,10 @@ export interface PromiseResponseWrapper { * The asynchronous resource used to track the task execution. */ readonly asyncResource?: AsyncResource + /** + * Original input retained for errors from tasks with transferred objects. + */ + readonly originalData?: unknown /** * Reject callback to reject the promise. */ @@ -163,9 +171,11 @@ export interface TaskFunctionProperties { readonly strategy?: WorkerChoiceStrategy /** * Task function worker node keys affinity. - * Restricts task execution to specified worker nodes by their indices. + * Restricts initial task dispatch to specified worker nodes by their indices. * Must contain valid indices within [0, pool max size - 1]. * If undefined, task can execute on any worker node. + * Task stealing ignores affinity. For strict affinity with task queues enabled, + * disable both `taskStealing` and `tasksStealingOnBackPressure`. */ readonly workerNodeKeys?: number[] } diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 311161077..09972a417 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -2,7 +2,9 @@ import type { Worker } from 'node:cluster' import type { UUID } from 'node:crypto' import type { MessagePort } from 'node:worker_threads' +import { AsyncLocalStorage } from 'node:async_hooks' import { performance } from 'node:perf_hooks' +import { isPromise } from 'node:util/types' import type { MessageValue, @@ -86,11 +88,23 @@ export abstract class AbstractWorker< */ protected taskAbortFunctions: Map void> + /** + * Correlation of task function operations across protected handler overrides. + */ + protected readonly taskFunctionOperationContext = new AsyncLocalStorage< + undefined | UUID + >() + /** * Task function object(s) processed by the worker when the pool's `execute` method is invoked. */ protected taskFunctions!: Map> + private readonly taskFunctionOperationIds = new WeakMap< + TaskFunctionProperties, + UUID + >() + /** * Constructs a new poolifier worker. * @param isMain - Whether this is the main worker or not. @@ -269,6 +283,23 @@ export abstract class AbstractWorker< return this.mainWorker } + /** + * Gets the task function operation UUID for a request or response. + * @param message - The task function operation message. + * @returns The task function operation UUID if available. + */ + protected getTaskFunctionOperationId ( + message: MessageValue + ): undefined | UUID { + return ( + message.taskFunctionOperationId ?? + (message.taskFunctionProperties != null + ? this.taskFunctionOperationIds.get(message.taskFunctionProperties) + : undefined) ?? + this.taskFunctionOperationContext.getStore() + ) + } + /** * Handles a worker error. * @param error - The error raised by the worker. @@ -287,22 +318,32 @@ export abstract class AbstractWorker< */ protected handleKillMessage (message: MessageValue): void { this.stopCheckActive() + let promise: Promise | undefined try { - const result = this.opts.killHandler?.() - if (result instanceof Promise) { - result - .then(() => { + promise = this.getResultPromise(this.opts.killHandler?.()) + } catch { + this.sendToMainWorker({ kill: 'failure' }) + return + } + if (promise != null) { + promise + .then( + () => { this.sendToMainWorker({ kill: 'success' }) return undefined - }) - .catch(() => { + }, + () => { this.sendToMainWorker({ kill: 'failure' }) + return undefined + } + ) + .catch((error: unknown) => { + queueMicrotask(() => { + throw error }) - } else { - this.sendToMainWorker({ kill: 'success' }) - } - } catch { - this.sendToMainWorker({ kill: 'failure' }) + }) + } else { + this.sendToMainWorker({ kill: 'success' }) } } @@ -312,6 +353,13 @@ export abstract class AbstractWorker< */ protected abstract handleReadyMessage (message: MessageValue): void + /** + * Handles a task function operation. + * Deferred overrides must retain the task function properties reference or the + * operation UUID when forwarding outside the incoming asynchronous context. + * An acknowledgement without a recoverable UUID rejects the pool operation. + * @param message - The task function operation request. + */ protected handleTaskFunctionOperationMessage ( message: MessageValue ): void { @@ -365,6 +413,7 @@ export abstract class AbstractWorker< const { error, status } = response this.sendToMainWorker({ taskFunctionOperation, + taskFunctionOperationId: this.getTaskFunctionOperationId(message), taskFunctionOperationStatus: status, taskFunctionProperties, ...(!status && @@ -400,7 +449,21 @@ export abstract class AbstractWorker< checkActive ? this.startCheckActive() : this.stopCheckActive() } else if (taskFunctionOperation != null) { // Task function operation message received - this.handleTaskFunctionOperationMessage(message) + if ( + message.taskFunctionProperties != null && + typeof message.taskFunctionProperties === 'object' && + message.taskFunctionOperationId != null + ) { + this.taskFunctionOperationIds.set( + message.taskFunctionProperties, + message.taskFunctionOperationId + ) + } + this.taskFunctionOperationContext.run( + message.taskFunctionOperationId, + this.runTaskFunctionOperation, + message + ) } else if (taskId != null && data != null) { // Task message received this.run(message) @@ -420,19 +483,13 @@ export abstract class AbstractWorker< * @param task - The task to execute. */ protected readonly run = (task: Task): void => { - const { abortable, data, name, taskId } = task + const { abortable, name, taskId } = task const taskFunctionName = name ?? DEFAULT_TASK_NAME if (!this.taskFunctions.has(taskFunctionName)) { - this.sendToMainWorker({ - taskId, - workerError: { - data, - name, - ...this.handleError( - new Error(`Task function '${taskFunctionName}' not found`) - ), - }, - }) + this.sendTaskError( + task, + new Error(`Task function '${taskFunctionName}' not found`) + ) return } let fn: TaskFunction @@ -459,36 +516,21 @@ export abstract class AbstractWorker< fn: TaskAsyncFunction, task: Task ): void => { - const { abortable, data, name, taskId } = task - let taskPerformance = this.beginTaskPerformance(name) - fn(data) - .then(res => { - taskPerformance = this.endTaskPerformance(taskPerformance) - this.sendToMainWorker({ - data: res, - taskId, - taskPerformance, - }) - return undefined - }) - .catch((error: unknown) => { - this.sendToMainWorker({ - taskId, - workerError: { - data, - name, - ...this.handleError(error as Error), - }, - }) - }) - .finally(() => { - this.updateLastTaskTimestamp() - if (abortable === true) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.taskAbortFunctions.delete(taskId!) - } - }) - .catch(EMPTY_FUNCTION) + try { + const taskPerformance = this.beginTaskPerformance(task.name) + const result = fn(task.data) + this.completeAsyncTask( + this.getResultPromise(result) ?? Promise.resolve(result), + task, + taskPerformance + ) + } catch (error) { + try { + this.sendTaskError(task, error as Error) + } finally { + this.finishTask(task) + } + } } /** @@ -500,10 +542,17 @@ export abstract class AbstractWorker< fn: TaskSyncFunction, task: Task ): void => { - const { abortable, data, name, taskId } = task + const { data, name, taskId } = task + let asynchronous = false try { let taskPerformance = this.beginTaskPerformance(name) const res = fn(data) + const promise = this.getResultPromise(res) + if (promise != null) { + this.completeAsyncTask(promise, task, taskPerformance) + asynchronous = true + return + } taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, @@ -511,20 +560,9 @@ export abstract class AbstractWorker< taskPerformance, }) } catch (error) { - this.sendToMainWorker({ - taskId, - workerError: { - data, - name, - ...this.handleError(error as Error), - }, - }) + this.sendTaskError(task, error as Error) } finally { - this.updateLastTaskTimestamp() - if (abortable === true) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.taskAbortFunctions.delete(taskId!) - } + if (!asynchronous) this.finishTask(task) } } @@ -644,6 +682,35 @@ export abstract class AbstractWorker< this.opts = { ...DEFAULT_WORKER_OPTIONS, ...opts } } + private completeAsyncTask ( + promise: Promise, + task: Task, + taskPerformance: TaskPerformance + ): void { + const { taskId } = task + promise + .then(res => { + taskPerformance = this.endTaskPerformance(taskPerformance) + this.sendToMainWorker({ + data: res, + taskId, + taskPerformance, + }) + return undefined + }) + .catch((error: unknown) => { + this.sendTaskError(task, error as Error) + }) + .finally(() => { + this.finishTask(task) + }) + .catch((error: unknown) => { + queueMicrotask(() => { + throw error + }) + }) + } + private endTaskPerformance ( taskPerformance: TaskPerformance ): TaskPerformance { @@ -661,6 +728,13 @@ export abstract class AbstractWorker< } } + private finishTask (task: Task): void { + this.updateLastTaskTimestamp() + if (task.abortable === true && task.taskId != null) { + this.taskAbortFunctions.delete(task.taskId) + } + } + /** * Gets abortable task function. * An abortable promise is built to permit the task to be aborted. @@ -672,23 +746,93 @@ export abstract class AbstractWorker< name: string, taskId: UUID ): TaskAsyncFunction { - return async (data?: Data): Promise => - await new Promise( - (resolve, reject: (reason?: unknown) => void) => { - this.taskAbortFunctions.set(taskId, () => { - reject(new AbortError(`Task '${name}' id '${taskId}' aborted`)) - }) - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const taskFunction = this.taskFunctions.get(name)!.taskFunction - if (isAsyncFunction(taskFunction)) { - ;(taskFunction as TaskAsyncFunction)(data) - .then(resolve) - .catch(reject) - } else { - resolve((taskFunction as TaskSyncFunction)(data)) - } + return async (data?: Data): Promise => { + let result: PromiseLike | Response | undefined + // Keep cancellation active while awaiting the task. Resolve only this gate, + // not the task result, to avoid an extra promise assimilation. + await new Promise((resolve, reject: (reason?: unknown) => void) => { + this.taskAbortFunctions.set(taskId, () => { + reject(new AbortError(`Task '${name}' id '${taskId}' aborted`)) + }) + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const taskFunction = this.taskFunctions.get(name)!.taskFunction + const value = taskFunction(data) + const promise = this.getResultPromise(value) + if (promise != null) { + // eslint-disable-next-line promise/catch-or-return -- The second callback handles rejection. + promise.then(value => { + result = value + resolve() + return undefined + }, reject) + } else { + result = value + resolve() } - ) + }) + return result as Response + } + } + + private getResultPromise( + result: PromiseLike | Result + ): Promise | undefined { + if ( + result == null || + (typeof result !== 'object' && typeof result !== 'function') + ) { + return + } + if ( + isPromise(result) && + Object.getPrototypeOf(result) === Promise.prototype && + !Object.hasOwn(result, 'then') + ) { + return result as Promise + } + const then: unknown = Reflect.get(result, 'then') + if (typeof then === 'function') { + // The declared result type determines the callable's fulfillment type. + const promiseThen = then as PromiseLike['then'] + return Promise.resolve({ + then (...args: Parameters): void { + Reflect.apply(promiseThen, result, args) + }, + }) + } + return undefined + } + + private readonly runTaskFunctionOperation = ( + message: MessageValue + ): void => { + this.handleTaskFunctionOperationMessage(message) + } + + private sendTaskError (task: Task, error: Error): void { + const message = { + taskId: task.taskId, + workerError: { + ...((task.transferList == null || task.transferList.length === 0) && { + data: task.data, + }), + name: task.name, + ...this.handleError(error), + }, + } satisfies MessageValue + try { + this.sendToMainWorker(message) + } catch (sendError) { + if ( + !(sendError instanceof Error) || + sendError.name !== 'DataCloneError' || + !Object.hasOwn(message.workerError, 'data') + ) { + throw sendError + } + delete message.workerError.data + this.sendToMainWorker(message) + } } /** diff --git a/src/worker/cluster-worker.ts b/src/worker/cluster-worker.ts index a2b626102..76aac8828 100644 --- a/src/worker/cluster-worker.ts +++ b/src/worker/cluster-worker.ts @@ -78,9 +78,11 @@ export class ClusterWorker< protected readonly sendToMainWorker = ( message: MessageValue ): void => { - this.getMainWorker().send({ - ...message, - workerId: this.id, - } satisfies MessageValue) + const response = { ...message, workerId: this.id } + if (response.taskFunctionOperationStatus != null) { + response.taskFunctionOperationId ??= + this.getTaskFunctionOperationId(response) + } + this.getMainWorker().send(response) } } diff --git a/src/worker/task-functions.ts b/src/worker/task-functions.ts index cce383074..32b0cd5e5 100644 --- a/src/worker/task-functions.ts +++ b/src/worker/task-functions.ts @@ -14,13 +14,13 @@ export type TaskAsyncFunction = ( /** * Task function that can be executed. - * This function can be synchronous or asynchronous. + * This function can return a value, a promise or another thenable. * @template Data - Type of data sent to the worker. This can only be structured-cloneable data. * @template Response - Type of execution response. This can only be structured-cloneable data. */ -export type TaskFunction = - | TaskAsyncFunction - | TaskSyncFunction +export type TaskFunction = ( + data?: Data +) => PromiseLike | Response /** * Task function object. @@ -42,9 +42,11 @@ export interface TaskFunctionObject { taskFunction: TaskFunction /** * Task function worker node keys affinity. - * Restricts task execution to specified worker nodes by their indices. + * Restricts initial task dispatch to specified worker nodes by their indices. * Must contain valid indices within [0, pool max size - 1]. * If undefined, task can execute on any worker node. + * Task stealing ignores affinity. For strict affinity with task queues enabled, + * disable both `taskStealing` and `tasksStealingOnBackPressure`. */ workerNodeKeys?: number[] } diff --git a/src/worker/thread-worker.ts b/src/worker/thread-worker.ts index 34bfd5bbb..af1c0917a 100644 --- a/src/worker/thread-worker.ts +++ b/src/worker/thread-worker.ts @@ -68,13 +68,6 @@ export class ThreadWorker< } } - /** @inheritDoc */ - protected override handleKillMessage (message: MessageValue): void { - super.handleKillMessage(message) - this.port?.unref() - this.port?.close() - } - /** @inheritDoc */ protected handleReadyMessage (message: MessageValue): void { if ( @@ -102,9 +95,15 @@ export class ThreadWorker< protected readonly sendToMainWorker = ( message: MessageValue ): void => { - this.port?.postMessage({ - ...message, - workerId: this.id, - } satisfies MessageValue) + const response = { ...message, workerId: this.id } + if (response.taskFunctionOperationStatus != null) { + response.taskFunctionOperationId ??= + this.getTaskFunctionOperationId(response) + } + this.port?.postMessage(response) + if (message.kill === 'success' || message.kill === 'failure') { + this.port?.unref() + this.port?.close() + } } } diff --git a/src/worker/worker-options.ts b/src/worker/worker-options.ts index d2b8b93a2..11b1e454f 100644 --- a/src/worker/worker-options.ts +++ b/src/worker/worker-options.ts @@ -19,9 +19,10 @@ export const KillBehaviors: Readonly<{ HARD: 'HARD'; SOFT: 'SOFT' }> = export type KillBehavior = keyof typeof KillBehaviors /** - * Handler called when a worker is killed. + * Cleanup handler called when the pool requests worker termination. + * Returned promises and other thenable values are awaited. */ -export type KillHandler = () => Promise | void +export type KillHandler = () => PromiseLike | void /** * Options for workers. diff --git a/tests/declarations.test.mjs b/tests/declarations.test.mjs new file mode 100644 index 000000000..ea9f78c68 --- /dev/null +++ b/tests/declarations.test.mjs @@ -0,0 +1,76 @@ +import { resolve } from 'node:path' +import ts from 'typescript' +import { expect, it } from 'vitest' + +const typeOnlyClasses = [ + 'AbstractPool', + 'AbstractWorker', + 'CircularBuffer', + 'PriorityQueue', + 'WorkerChoiceStrategiesContext', +] + +it('preserves public class and task function contracts', () => { + const file = resolve('tests/declaration-consumer.mts').replaceAll('\\', '/') + const source = ` + import type { Worker, MessagePort } from 'node:worker_threads' + import type { AbstractPool, AbstractWorker, CircularBuffer, PriorityQueue, Task, TaskAsyncFunction, TaskFunction, TaskFunctions, TaskSyncFunction, WorkerChoiceStrategiesContext, WorkerOptions } from 'poolifier' + import { FixedThreadPool, DynamicThreadPool, FixedClusterPool, DynamicClusterPool, ThreadWorker, ClusterWorker } from 'poolifier' + export type Instances = [AbstractPool, AbstractWorker, CircularBuffer, PriorityQueue, WorkerChoiceStrategiesContext] + export type Constructors = [typeof AbstractPool, typeof AbstractWorker, typeof CircularBuffer, typeof PriorityQueue, typeof WorkerChoiceStrategiesContext] + export const publicConstructors = [FixedThreadPool, DynamicThreadPool, FixedClusterPool, DynamicClusterPool, ThreadWorker, ClusterWorker] + export abstract class CompatibleWorker extends ThreadWorker { + protected abstract override readonly runAsync: (fn: TaskAsyncFunction, task: Task) => void + protected abstract override readonly runSync: (fn: TaskSyncFunction, task: Task) => void + } + declare const result: PromiseLike + declare const cleanup: PromiseLike + export const task: TaskFunction = () => result + export const mixedTask: TaskFunction = data => data == null ? result : data + export const functions: TaskFunctions = { task, wrapped: { taskFunction: mixedTask } } + export const options: WorkerOptions = { killHandler: () => cleanup } + export const threadWorker = new ThreadWorker(task, options) + export const clusterWorker = new ClusterWorker(functions, options) + declare const threadPool: FixedThreadPool + declare const clusterPool: FixedClusterPool + void threadPool.addTaskFunction('thenable', task) + void clusterPool.addTaskFunction('mixed', { taskFunction: mixedTask }) + declare const wrongResult: PromiseLike + // @ts-expect-error A thenable must fulfill with the declared response type. + export const wrongTask: TaskFunction = () => wrongResult + // @ts-expect-error Protected async dispatch retains its native Promise contract. + export const protectedAsync: TaskAsyncFunction = () => result + ${typeOnlyClasses + .map( + name => ` + // @ts-expect-error A type-only class export cannot be imported as a value. + import { ${name} as ${name}Value } from 'poolifier' + // @ts-expect-error A type-only class export cannot be used as a value. + void ${name}Value + ` + ) + .join('\n')} + ` + const options = { + module: ts.ModuleKind.Node16, + moduleResolution: ts.ModuleResolutionKind.Node16, + noEmit: true, + strict: true, + target: ts.ScriptTarget.ES2022, + types: ['node'], + verbatimModuleSyntax: true, + } + const host = ts.createCompilerHost(options) + const getSourceFile = host.getSourceFile.bind(host) + host.getSourceFile = (name, ...args) => + name === file + ? ts.createSourceFile(name, source, options.target, true) + : getSourceFile(name, ...args) + const program = ts.createProgram([file], options, host) + const diagnostics = ts.getPreEmitDiagnostics(program) + expect( + diagnostics.map(diagnostic => + ts.flattenDiagnosticMessageText(diagnostic.messageText, '\n') + ) + ).toEqual([]) +}) diff --git a/tests/pools/abstract-pool.test.mjs b/tests/pools/abstract-pool.test.mjs index 16a107e5f..4dde4fb14 100644 --- a/tests/pools/abstract-pool.test.mjs +++ b/tests/pools/abstract-pool.test.mjs @@ -1739,11 +1739,6 @@ describe('Abstract pool test suite', () => { taskFunction: echoTaskFunction, }) ).resolves.toBe(true) - expect(dynamicThreadPool.taskFunctions.size).toBe(1) - expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual({ - strategy: WorkerChoiceStrategies.LEAST_ELU, - taskFunction: echoTaskFunction, - }) expect([ ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(), ]).toStrictEqual([ @@ -1857,10 +1852,6 @@ describe('Abstract pool test suite', () => { workerNodeKeys: [poolWorkerNodeKeys[0]], }) ).resolves.toBe(true) - expect(dynamicThreadPool.taskFunctions.get('affinityEcho')).toStrictEqual({ - taskFunction: echoTaskFunction, - workerNodeKeys: [poolWorkerNodeKeys[0]], - }) // Test with invalid workerNodeKeys (out of range) await expect( @@ -2034,11 +2025,6 @@ describe('Abstract pool test suite', () => { strategy: WorkerChoiceStrategies.LEAST_ELU, taskFunction: echoTaskFunction, }) - expect(dynamicThreadPool.taskFunctions.size).toBe(1) - expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual({ - strategy: WorkerChoiceStrategies.LEAST_ELU, - taskFunction: echoTaskFunction, - }) expect([ ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(), ]).toStrictEqual([ @@ -2053,8 +2039,6 @@ describe('Abstract pool test suite', () => { await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe( true ) - expect(dynamicThreadPool.taskFunctions.size).toBe(0) - expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined() expect([ ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(), ]).toStrictEqual([WorkerChoiceStrategies.LEAST_USED]) diff --git a/tests/pools/task-function-replay.test.mjs b/tests/pools/task-function-replay.test.mjs new file mode 100644 index 000000000..a7dc579d5 --- /dev/null +++ b/tests/pools/task-function-replay.test.mjs @@ -0,0 +1,1773 @@ +import { execFile } from 'node:child_process' +import { getEventListeners, once } from 'node:events' +import { promisify } from 'node:util' +import { describe, expect, it, vi } from 'vitest' + +import { + DynamicThreadPool, + FixedClusterPool, + FixedThreadPool, + PoolEvents, + WorkerChoiceStrategies, +} from '../../lib/index.mjs' +import { bounded, waitForMessage } from '../test-utils.cjs' + +const release = gate => { + Atomics.store(new Int32Array(gate), 0, 1) + Atomics.notify(new Int32Array(gate), 0) +} + +const deferWorkerMessagesUntilExit = workerNode => { + const port = workerNode.messageChannel.port1 + const listeners = getEventListeners(port, 'message') + for (const listener of listeners) port.off('message', listener) + let restored = false + const restore = () => { + if (restored) return + restored = true + for (const listener of listeners) port.on('message', listener) + } + workerNode.worker.prependOnceListener('exit', restore) + return restore +} + +class WrappedThreadPool extends FixedThreadPool { + deregisterWorkerMessageListener (workerNodeKey, listener) { + const wrappers = this.listenerWrappers.get(this.workerNodes[workerNodeKey]) + const wrapped = wrappers?.get(listener) + wrappers?.delete(listener) + super.deregisterWorkerMessageListener(workerNodeKey, wrapped ?? listener) + this.onListenerRemoval?.(workerNodeKey) + } + + registerWorkerMessageListener (workerNodeKey, listener) { + const workerNode = this.workerNodes[workerNodeKey] + let wrappers = this.listenerWrappers.get(workerNode) + if (wrappers == null) { + wrappers = new Map() + this.listenerWrappers.set(workerNode, wrappers) + } + const wrapped = listener.bind(this) + wrappers.set(listener, wrapped) + super.registerWorkerMessageListener(workerNodeKey, wrapped) + } + + setupHook () { + this.listenerWrappers = new WeakMap() + } +} + +describe('Runtime task function replay', () => { + it.each([ + ['thread', FixedThreadPool], + ['cluster', FixedClusterPool], + ])( + 'rejects an uncorrelated %s bootstrap acknowledgement', + async (workerType, Pool) => { + const pool = new Pool( + 1, + workerType === 'thread' + ? './tests/worker-files/thread/taskLifecycleWorker.mjs' + : './tests/worker-files/cluster/legacyTaskFunctionWorker.mjs', + { + startWorkers: false, + ...(workerType === 'thread' + ? { + workerOptions: { + workerData: { + legacyTaskFunctionMessages: true, + queuedTaskFunctionOperations: true, + statusOnlyTaskFunctionOperation: 'add', + }, + }, + } + : { env: { POOLIFIER_TEST_STATUS_ONLY_OPERATION: 'add' } }), + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + let execution + try { + await bounded(pool.addTaskFunction('runtime', () => 42)) + pool.start() + const node = pool.workerNodes[0] + execution = pool.execute({}, 'runtime').then( + value => ({ value }), + error => ({ error }) + ) + const result = await bounded(execution) + expect(result.error).toBeInstanceOf(Error) + expect(errors).toEqual([result.error]) + expect(pool.workerNodes[0]).toBe(node) + expect(node.info.ready).toBe(false) + } finally { + await bounded(pool.destroy()) + await bounded(Promise.allSettled([execution])) + } + } + ) + + it('preserves a ready survivor strategy when a replacement restores its functions', async () => { + const bootstrapStatus = new SharedArrayBuffer(8) + const roles = [ + { strategy: WorkerChoiceStrategies.FAIR_SHARE, workerNodeKeys: [0] }, + {}, + {}, + ] + let created = 0 + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/bootstrapAffinityWorker.mjs', + { + workerOptions: { + get workerData () { + return { bootstrapRole: created++, bootstrapStatus, roles } + }, + }, + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', data => data)) + const survivor = pool.workerNodes[0] + expect(await bounded(pool.execute({ id: 'before' }, 'pinned'))).toEqual({ + id: 'before', + taskFunction: 'pinned', + workerId: survivor.info.id, + }) + await pool.workerNodes[1].worker.terminate() + await vi.waitFor(() => { + expect(pool.workerNodes.every(node => node.info.ready)).toBe(true) + }) + expect(pool.workerNodes[0]).toBe(survivor) + expect(await bounded(pool.execute({ id: 'after' }, 'pinned'))).toEqual({ + id: 'after', + taskFunction: 'pinned', + workerId: survivor.info.id, + }) + expect(errors).toEqual([]) + } finally { + await bounded(pool.destroy()) + } + }) + + it('registers the first replayed strategy while an earlier worker still initializes', async () => { + const replayGates = [new SharedArrayBuffer(4), new SharedArrayBuffer(4)] + const replayCounter = new SharedArrayBuffer(4) + const blocked = new Map() + const allBlocked = Promise.withResolvers() + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) { + blocked.set(message.workerId, message.replayGeneration) + if (blocked.size === 2) allBlocked.resolve() + } + }, + startWorkers: false, + workerOptions: { + workerData: { asyncReplay: true, replayCounter, replayGates }, + }, + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + await bounded( + pool.addTaskFunction('runtime', { + strategy: WorkerChoiceStrategies.FAIR_SHARE, + taskFunction: () => 42, + workerNodeKeys: [1], + }) + ) + pool.start() + const nodes = [...pool.workerNodes] + await bounded( + Promise.all([ + allBlocked.promise, + ...nodes.map(node => + waitForMessage( + node.messageChannel.port1, + message => message.ready != null + ) + ), + ]) + ) + expect(nodes[0].info.taskFunctionsProperties).toBeDefined() + expect( + nodes[0].info.taskFunctionsProperties.some( + properties => properties.name === 'runtime' + ) + ).toBe(false) + const bootstrap = pool.workerTaskFunctionBootstraps.get(nodes[1]).promise + release(replayGates[blocked.get(nodes[1].info.id)]) + await bounded(bootstrap) + expect(nodes[0].info.ready).toBe(false) + expect(nodes[1].info.ready).toBe(true) + expect(await bounded(pool.execute({}, 'runtime'))).toBe(42) + expect(errors).toEqual([]) + } finally { + replayGates.forEach(release) + await bounded(pool.destroy()) + } + }) + + it.each(['remove', 'default'])( + 'waits for workers created while resuming a %s operation', + async operation => { + const gate = new SharedArrayBuffer(4) + const tasks = [] + const pool = new DynamicThreadPool( + 1, + 3, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', data => data)) + await bounded( + pool.addTaskFunction('secondHold', { + taskFunction: async ({ gate }) => { + await Atomics.waitAsync(new Int32Array(gate), 0, 0).value + return 'released' + }, + workerNodeKeys: [1], + }) + ) + tasks.push(pool.execute({ gate }, 'hold')) + tasks.push(pool.execute({}, 'echo')) + const bootstrap = pool.workerTaskFunctionBootstraps.get( + pool.workerNodes[1] + ).promise + const change = + operation === 'remove' + ? pool.removeTaskFunction('runtime') + : pool.setDefaultTaskFunction('runtime') + await Promise.resolve() + const growth = bootstrap.then(() => { + tasks.push(pool.execute({ gate }, 'secondHold')) + tasks.push(pool.execute({}, 'echo')) + return undefined + }) + await expect(bounded(change)).resolves.toBe(true) + await bounded(growth) + expect(pool.workerNodes).toHaveLength(3) + expect(pool.hasTaskFunction('runtime')).toBe(operation !== 'remove') + if (operation === 'default') { + expect(await bounded(pool.execute({ selected: true }))).toEqual({ + selected: true, + }) + } + } finally { + release(gate) + await bounded(Promise.allSettled(tasks)) + await bounded(pool.destroy()) + } + } + ) + + it('keeps a HARD-terminating dynamic worker unavailable after replay', async () => { + const replayGate = new SharedArrayBuffer(4) + const cleanupGate = new SharedArrayBuffer(4) + const taskGate = new SharedArrayBuffer(4) + const started = new SharedArrayBuffer(4) + const cleanup = Promise.withResolvers() + const pool = new DynamicThreadPool( + 0, + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.cleanupStarted) cleanup.resolve() + }, + workerOptions: { + workerData: { + asyncReplay: true, + cleanupGate, + hardReplayKill: true, + replayGate, + }, + }, + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + let execution + let settled = false + try { + await pool.addTaskFunction('runtime', async data => { + Atomics.store(new Int32Array(data.started), 0, 1) + await Atomics.waitAsync(new Int32Array(data.taskGate), 0, 0).value + return data.buffer.byteLength + }) + const data = { buffer: new ArrayBuffer(8), started, taskGate } + execution = pool.execute(data, 'runtime', undefined, [data.buffer]).then( + value => { + settled = true + return { value } + }, + error => { + settled = true + return { error } + } + ) + await bounded(cleanup.promise) + const node = pool.workerNodes[0] + const exited = once(node.worker, 'exit') + expect(node.info.ready).toBe(false) + release(replayGate) + await vi.waitFor(() => { + expect(settled || Atomics.load(new Int32Array(started), 0) === 1).toBe( + true + ) + }) + expect(node.info.ready).toBe(false) + expect(Atomics.load(new Int32Array(started), 0)).toBe(0) + expect(data.buffer.byteLength).toBe(8) + expect((await bounded(execution)).error).toBeInstanceOf(Error) + expect(node.worker.threadId).toBeGreaterThan(0) + expect(pool.info.executingTasks).toBe(0) + release(cleanupGate) + await bounded(exited) + expect(pool.workerNodes).toEqual([]) + expect(errors).toEqual([]) + } finally { + release(replayGate) + release(taskGate) + if (Atomics.load(new Int32Array(started), 0) === 1 && execution != null) { + await bounded(execution) + } + release(cleanupGate) + await bounded(pool.destroy()) + } + }) + + it.each(['replacement', 'restart'])( + 'replays the captured source after %s', + async transition => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + const taskFunction = () => 42 + await bounded(pool.addTaskFunction('captured', taskFunction)) + await bounded(pool.setDefaultTaskFunction('captured')) + taskFunction.toString = () => '() => 99' + const original = pool.workerNodes[0] + if (transition === 'replacement') { + await original.worker.terminate() + } else { + await bounded(pool.destroy()) + pool.start() + } + await vi.waitFor(() => { + expect(pool.workerNodes).toHaveLength(1) + expect(pool.workerNodes[0]).not.toBe(original) + expect(pool.workerNodes[0].info.ready).toBe(true) + }) + expect(await bounded(pool.execute({}))).toBe(42) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it('keeps execution and registration waiting across replacement replay exits', async () => { + const replayGates = Array.from( + { length: 3 }, + () => new SharedArrayBuffer(4) + ) + const blocked = Array.from({ length: 3 }, () => Promise.withResolvers()) + release(replayGates[0]) + const errors = [] + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) { + blocked[message.replayGeneration].resolve() + } + }, + workerOptions: { + workerData: { replayCounter: new SharedArrayBuffer(4), replayGates }, + }, + } + ) + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + const tasks = [] + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded( + pool.addTaskFunction('runtime', data => data.buffer.byteLength) + ) + await pool.workerNodes[0].worker.terminate() + await bounded(blocked[1].promise) + const data = { buffer: new ArrayBuffer(8) } + const abortedData = { buffer: new ArrayBuffer(4) } + const controller = new AbortController() + const reason = new Error('Replay wait aborted') + let executed = false + let registered = false + const execution = pool + .execute(data, 'runtime', undefined, [data.buffer]) + .then( + value => { + executed = true + return { value } + }, + error => { + executed = true + return { error } + } + ) + const registration = pool + .addTaskFunction('later', () => 9) + .then( + value => { + registered = true + return { value } + }, + error => { + registered = true + return { error } + } + ) + const aborted = pool + .execute(abortedData, 'runtime', controller.signal, [ + abortedData.buffer, + ]) + .catch(error => error) + tasks.push(execution, registration, aborted) + const retiring = pool.workerNodes[0].worker + retiring.once('exit', () => controller.abort(reason)) + await retiring.terminate() + await bounded(blocked[2].promise) + expect(await bounded(aborted)).toBe(reason) + expect(executed).toBe(false) + expect(registered).toBe(false) + expect(data.buffer.byteLength).toBe(8) + expect(abortedData.buffer.byteLength).toBe(4) + release(replayGates[2]) + expect(await bounded(execution)).toEqual({ value: 8 }) + expect(await bounded(registration)).toEqual({ value: true }) + expect(await bounded(pool.execute({}, 'later'))).toBe(9) + expect(errors).toEqual([]) + } finally { + replayGates.forEach(release) + await bounded(Promise.allSettled(tasks)) + await bounded(pool.destroy()) + } + }) + + it.each([ + ['worker-defined', 1], + ['runtime', 1], + ['default', 1], + ['worker-defined', 0], + ['unrestricted', 1], + ])( + 'scopes replay failures for %s submissions when worker %i fails', + async (kind, failedWorkerNodeKey) => { + const replayGates = Array.from( + { length: kind === 'default' ? 4 : 2 }, + () => new SharedArrayBuffer(4) + ) + const replayCounter = new SharedArrayBuffer(4) + const decisionCounter = new SharedArrayBuffer(4) + const taskGate = new SharedArrayBuffer(4) + release(taskGate) + const blocked = new Map() + const allBlocked = Promise.withResolvers() + const failed = Promise.withResolvers() + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) { + blocked.set(message.workerId, message.replayGeneration) + if (blocked.size === 2) allBlocked.resolve() + } + }, + startWorkers: false, + workerOptions: { + workerData: { + asyncReplay: true, + blockWorkerKeys: [kind === 'default' ? 1 : 0], + decisionCounter, + replayCounter, + replayGates, + }, + }, + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => { + errors.push(error) + failed.resolve(error) + }) + let execution + try { + const runtime = () => 42 + // Only the first released worker fails source evaluation. + runtime.toString = () => + '(Atomics.add(new Int32Array(process.getBuiltinModule("node:worker_threads").workerData.decisionCounter), 0, 1) === 0 ? null : (() => 42))' + const registration = { + taskFunction: runtime, + ...((kind === 'runtime' || kind === 'default') && { + workerNodeKeys: [0], + }), + } + if (kind === 'default') { + replayGates.slice(0, 2).forEach(release) + Atomics.store(new Int32Array(decisionCounter), 0, 2) + const ready = once(pool.emitter, PoolEvents.ready) + pool.start() + await bounded(ready) + await bounded(pool.addTaskFunction('runtime', registration)) + await bounded(pool.setDefaultTaskFunction('runtime')) + await bounded(pool.destroy()) + Atomics.store(new Int32Array(decisionCounter), 0, 0) + } else { + await bounded( + pool.addTaskFunction( + kind === 'unrestricted' ? 'block' : 'runtime', + registration + ) + ) + } + pool.start() + const nodes = [...pool.workerNodes] + await bounded( + Promise.all([ + allBlocked.promise, + ...nodes.map(node => + waitForMessage( + node.messageChannel.port1, + message => message.ready != null + ) + ), + ]) + ) + const failedNode = nodes[failedWorkerNodeKey] + const healthyNode = nodes[1 - failedWorkerNodeKey] + const healthyBootstrap = + pool.workerTaskFunctionBootstraps.get(healthyNode).promise + const name = + kind === 'default' + ? undefined + : kind === 'runtime' + ? 'runtime' + : 'block' + execution = pool.execute({ gate: taskGate }, name).then( + value => ({ value }), + error => ({ error }) + ) + release(replayGates[blocked.get(failedNode.info.id)]) + const error = await bounded(failed.promise) + expect(error).toBeInstanceOf(Error) + expect(failedNode.worker.threadId).toBeGreaterThan(0) + expect(failedNode.info.ready).toBe(false) + release(replayGates[blocked.get(healthyNode.info.id)]) + await bounded(healthyBootstrap) + const result = await bounded(execution) + if (failedWorkerNodeKey === 0 || kind === 'unrestricted') { + expect(result.error).toBe(error) + } else { + expect(result).toStrictEqual({ + value: kind === 'worker-defined' ? 'released' : 42, + }) + } + expect(errors).toEqual([error]) + expect(failedNode.info.ready).toBe(false) + } finally { + replayGates.forEach(release) + await bounded(Promise.allSettled([execution])) + await bounded(pool.destroy()) + } + } + ) + + it('propagates replay failure from a worker that is still alive', async () => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { startWorkers: false } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + const invalid = () => 42 + invalid.toString = () => 'null' + await bounded(pool.addTaskFunction('invalid', invalid)) + pool.start() + const worker = pool.workerNodes[0].worker + const data = { buffer: new ArrayBuffer(8) } + const error = await bounded( + pool + .execute(data, 'invalid', undefined, [data.buffer]) + .catch(error => error) + ) + expect(error).toBeInstanceOf(Error) + expect(errors).toEqual([error]) + expect(pool.workerNodes[0].worker).toBe(worker) + expect(worker.threadId).toBeGreaterThan(0) + expect(data.buffer.byteLength).toBe(8) + expect(pool.info.executingTasks).toBe(0) + } finally { + await bounded(pool.destroy()) + } + }) + + it('retires wrapped message listeners through protected hooks during CRUD and replay', async () => { + const warnings = [] + const onWarning = warning => { + if (warning.name === 'MaxListenersExceededWarning') { + warnings.push(warning.message) + } + } + process.on('warning', onWarning) + const pool = new WrappedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + for (let i = 0; i < 12; ++i) { + await bounded(pool.addTaskFunction('runtime', () => 42)) + } + await bounded(pool.setDefaultTaskFunction('runtime')) + await expect( + bounded(pool.removeTaskFunction('runtime')) + ).rejects.toBeInstanceOf(Error) + const original = pool.workerNodes[0] + let reentered + pool.onListenerRemoval = workerNodeKey => { + if (pool.workerNodes[workerNodeKey].worker.threadId === -1) { + pool.onListenerRemoval = undefined + reentered = pool.execute({}).then( + value => ({ value }), + error => ({ error }) + ) + } + } + await original.worker.terminate() + await vi.waitFor(() => { + expect(pool.workerNodes).toHaveLength(2) + expect(pool.workerNodes.includes(original)).toBe(false) + expect(pool.workerNodes.every(node => node.info.ready)).toBe(true) + }) + expect(await bounded(reentered)).toEqual({ value: 42 }) + expect(await bounded(pool.execute({}))).toBe(42) + expect(warnings).toEqual([]) + } finally { + await bounded(pool.destroy()) + process.off('warning', onWarning) + } + }) + + it.each([false, true])( + 'selects a ready weighted worker during replay (function strategy: %s)', + async perFunction => { + const size = perFunction ? 3 : 2 + const replayGates = Array.from( + { length: size }, + () => new SharedArrayBuffer(4) + ) + const replayBlocked = Array.from({ length: size }, () => + Promise.withResolvers() + ) + const pool = new FixedThreadPool( + size, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) { + const key = pool.workerNodes.findIndex( + node => node.info.id === message.workerId + ) + replayBlocked[key].resolve(message.replayGeneration) + } + }, + startWorkers: false, + workerChoiceStrategy: perFunction + ? WorkerChoiceStrategies.ROUND_ROBIN + : WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN, + workerChoiceStrategyOptions: { + weights: perFunction ? { 0: 1, 1: 1, 2: 1 } : { 0: 1, 1: 1 }, + }, + workerOptions: { + workerData: { + replayCounter: new SharedArrayBuffer(4), + replayGates, + }, + }, + } + ) + const tasks = [] + try { + await pool.addTaskFunction('runtime', { + taskFunction: data => data.buffer.byteLength, + ...(perFunction && { + strategy: WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN, + workerNodeKeys: [0, 1], + }), + }) + pool.start() + const data = { buffer: new ArrayBuffer(8) } + let settled = false + const execution = pool + .execute(data, 'runtime', undefined, [data.buffer]) + .then( + value => { + settled = true + return { value } + }, + error => { + settled = true + return { error } + } + ) + tasks.push(execution) + if (perFunction) { + release(replayGates[await bounded(replayBlocked[2].promise)]) + await vi.waitFor(() => + expect(pool.workerNodes[2].info.ready).toBe(true) + ) + expect(settled).toBe(false) + expect(data.buffer.byteLength).toBe(8) + } + const eligibleGate = await bounded(replayBlocked[1].promise) + expect(data.buffer.byteLength).toBe(8) + release(replayGates[eligibleGate]) + expect(await bounded(execution)).toEqual({ value: 8 }) + expect(data.buffer.byteLength).toBe(0) + const heldGate = await bounded(replayBlocked[0].promise) + expect(Atomics.load(new Int32Array(replayGates[heldGate]), 0)).toBe(0) + expect( + await bounded(pool.execute({ buffer: new ArrayBuffer(8) }, 'runtime')) + ).toBe(8) + } finally { + replayGates.forEach(release) + await bounded(Promise.allSettled(tasks)) + await bounded(pool.destroy()) + } + } + ) + + it('wakes already-waiting executions when their eligible worker finishes replay', async () => { + const replayGates = [new SharedArrayBuffer(4), new SharedArrayBuffer(4)] + const replayBlocked = Promise.withResolvers() + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if ( + message.replayBlocked && + message.workerId === pool.workerNodes[1].info.id + ) { + replayBlocked.resolve(message.replayGeneration) + } + }, + startWorkers: false, + workerOptions: { + workerData: { replayCounter: new SharedArrayBuffer(4), replayGates }, + }, + } + ) + const tasks = [] + try { + await bounded( + pool.addTaskFunction('runtime', { + taskFunction: data => data.buffer.byteLength, + workerNodeKeys: [1], + }) + ) + pool.start() + const data = { buffer: new ArrayBuffer(8) } + tasks.push(pool.execute(data, 'runtime', undefined, [data.buffer])) + const eligibleGate = await bounded(replayBlocked.promise) + expect(data.buffer.byteLength).toBe(8) + release(replayGates[eligibleGate]) + await vi.waitFor(() => expect(pool.workerNodes[1].info.ready).toBe(true)) + expect(await bounded(tasks[0])).toBe(8) + expect(data.buffer.byteLength).toBe(0) + tasks.push(pool.execute({ buffer: new ArrayBuffer(8) }, 'runtime')) + expect(await bounded(tasks[1])).toBe(8) + } finally { + for (const gate of replayGates) release(gate) + await bounded(Promise.allSettled(tasks)) + await bounded(pool.destroy()) + } + }) + + it('uses ready metadata while the first worker replays without blocking eligible work', async () => { + const replayGates = Array.from( + { length: 4 }, + () => new SharedArrayBuffer(4) + ) + for (const i of [0, 1, 3]) release(replayGates[i]) + const replayBlocked = Promise.withResolvers() + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + enableTasksQueue: true, + messageHandler: message => { + if (message.replayBlocked) replayBlocked.resolve(message.workerId) + }, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: false, + }, + workerOptions: { + workerData: { replayCounter: new SharedArrayBuffer(4), replayGates }, + }, + } + ) + const holdGate = new SharedArrayBuffer(4) + const tasks = [] + let blocked + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded( + pool.addTaskFunction('runtime', { + priority: -10, + taskFunction: () => 'priority', + workerNodeKeys: [1], + }) + ) + await bounded( + pool.addTaskFunction('blockedOnly', { + taskFunction: data => data.buffer.byteLength, + workerNodeKeys: [0], + }) + ) + await bounded(pool.setDefaultTaskFunction('runtime')) + const [first, second] = pool.workerNodes + await second.worker.terminate() + const blockedWorkerId = await bounded(replayBlocked.promise) + await first.worker.terminate() + await vi.waitFor(() => expect(pool.workerNodes[1].info.ready).toBe(true)) + expect(pool.workerNodes[0].info.id).toBe(blockedWorkerId) + const data = { buffer: new ArrayBuffer(8) } + blocked = pool.execute(data, 'blockedOnly', undefined, [data.buffer]) + expect(data.buffer.byteLength).toBe(8) + const entered = waitForMessage( + pool.workerNodes[1].worker, + message => message.started + ) + tasks.push(pool.execute({ gate: holdGate }, 'hold')) + await bounded(entered) + const order = [] + tasks.push(pool.execute({}, 'echo').then(() => order.push('ordinary'))) + tasks.push(pool.execute({}).then(() => order.push('priority'))) + release(holdGate) + await bounded(Promise.all(tasks)) + expect(order).toEqual(['priority', 'ordinary']) + expect(data.buffer.byteLength).toBe(8) + release(replayGates[2]) + expect(await bounded(blocked)).toBe(8) + expect(data.buffer.byteLength).toBe(0) + } finally { + for (const gate of replayGates) release(gate) + release(holdGate) + await bounded(Promise.allSettled([...tasks, blocked])) + await bounded(pool.destroy()) + } + }) + + it.each(['remove', 'update', 'failed removal', 'late default'])( + 'does not replay stale registrations during %s', + async operation => { + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + workerOptions: { workerData: { blockWorkerKeys: [1] } }, + } + ) + const gate = new SharedArrayBuffer(4) + let blocked + let change + let defaultChange + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'old')) + if (operation === 'failed removal') { + await bounded(pool.setDefaultTaskFunction('runtime')) + } + const original = pool.workerNodes[0] + const blocking = waitForMessage( + pool.workerNodes[1].worker, + message => message.blocked + ) + blocked = pool.execute({ gate }, 'block') + await bounded(blocking) + const acknowledged = waitForMessage( + original.messageChannel.port1, + message => + message.taskFunctionOperationStatus != null && + message.taskFunctionProperties?.name === 'runtime' && + message.taskFunctionOperation === + (operation === 'remove' || operation === 'failed removal' + ? 'remove' + : 'add') + ) + change = ( + operation === 'remove' || operation === 'failed removal' + ? pool.removeTaskFunction('runtime') + : pool.addTaskFunction('runtime', () => 'new') + ).then( + value => ({ value }), + error => ({ error }) + ) + await bounded(acknowledged) + await original.worker.terminate() + await vi.waitFor(() => { + expect(pool.workerNodes).toHaveLength(2) + expect( + pool.workerNodes.some( + node => node !== original && node !== pool.workerNodes[0] + ) + ).toBe(true) + }) + if (operation === 'late default') { + defaultChange = pool.setDefaultTaskFunction('runtime') + } + release(gate) + const result = await bounded(change) + if (operation === 'failed removal') { + expect(result.error.message).toContain( + "Task function operation 'remove' failed" + ) + } else { + expect(result.value).toBe(true) + } + if (defaultChange != null) { + expect(await bounded(defaultChange)).toBe(true) + } + await bounded(blocked) + await bounded(pool.addTaskFunction('barrier', data => data)) + for (let i = 0; i < 2; ++i) { + if (operation === 'remove') { + await expect(bounded(pool.execute({}, 'runtime'))).rejects.toThrow( + "Task function 'runtime' not found" + ) + } else { + const name = operation === 'late default' ? undefined : 'runtime' + expect(await bounded(pool.execute({}, name))).toBe( + operation === 'failed removal' ? 'old' : 'new' + ) + } + } + } finally { + release(gate) + await bounded(Promise.allSettled([blocked, change, defaultChange])) + await bounded(pool.destroy()) + } + } + ) + + it.each(['add', 'remove', 'default'])( + 'preserves a buffered %s operation result after worker exit', + async operation => { + let created = 0 + const pool = new WrappedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + workerOptions: { + get workerData () { + return created++ === 0 + ? { exitAfterTaskFunctionOperation: operation } + : {} + }, + }, + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + let original + let change + let restore + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + if (operation !== 'add') { + await bounded(pool.addTaskFunction('runtime', () => 42)) + } + original = pool.workerNodes[0] + const exit = once(original.worker, 'exit') + change = ( + operation === 'add' + ? pool.addTaskFunction('runtime', () => 42) + : operation === 'remove' + ? pool.removeTaskFunction('runtime') + : pool.setDefaultTaskFunction('runtime') + ).then( + value => ({ value }), + error => ({ error }) + ) + await Promise.resolve() + restore = deferWorkerMessagesUntilExit(original) + expect(await bounded(change)).toEqual({ value: true }) + await bounded(exit) + await vi.waitFor(() => { + expect(pool.workerNodes[0]).not.toBe(original) + expect(pool.workerNodes[0].info.ready).toBe(true) + }) + if (operation === 'remove') { + const result = await bounded( + pool.execute({}, 'runtime').then( + value => ({ value }), + error => ({ error }) + ) + ) + expect(result).toEqual({ error: expect.any(Error) }) + } else { + expect( + await bounded( + pool.execute({}, operation === 'default' ? undefined : 'runtime') + ) + ).toBe(42) + } + expect(errors).toEqual([]) + } finally { + restore?.() + await bounded(pool.destroy()) + await bounded(Promise.allSettled([change])) + original?.messageChannel?.port1.close() + original?.messageChannel?.port2.close() + } + } + ) + + it('rejects a buffered operation failure when its worker exits before a peer succeeds', async () => { + let created = 0 + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + workerOptions: { + get workerData () { + return created++ === 0 + ? { exitAfterTaskFunctionOperation: 'remove' } + : {} + }, + }, + } + ) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + let original + let change + let restore + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 42)) + await bounded( + pool.sendTaskFunctionOperationToWorker(0, { + taskFunctionOperation: 'default', + taskFunctionProperties: { name: 'runtime' }, + }) + ) + original = pool.workerNodes[0] + const exit = once(original.worker, 'exit') + change = pool.removeTaskFunction('runtime').then( + value => ({ value }), + error => ({ error }) + ) + await Promise.resolve() + restore = deferWorkerMessagesUntilExit(original) + expect(await bounded(change)).toEqual({ error: expect.any(Error) }) + await bounded(exit) + await vi.waitFor(() => { + expect(pool.workerNodes.every(node => node.info.ready)).toBe(true) + }) + expect(await bounded(pool.execute({ usable: true }, 'echo'))).toEqual({ + usable: true, + }) + expect(errors).toEqual([]) + } finally { + restore?.() + await bounded(pool.destroy()) + await bounded(Promise.allSettled([change])) + original?.messageChannel?.port1.close() + original?.messageChannel?.port2.close() + } + }) + + it.each(['add', 'remove', 'default'])( + 'releases replacement bootstrap after a target exits without an ACK (%s)', + async operation => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + const gate = new SharedArrayBuffer(4) + let blocked + let change + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', data => data)) + const original = pool.workerNodes[0] + const entered = waitForMessage( + original.worker, + message => message.blocked + ) + blocked = pool.execute({ gate }, 'block').catch(error => error) + await bounded(entered) + const pending = + operation === 'add' + ? pool.addTaskFunction('newRuntime', data => data) + : operation === 'remove' + ? pool.removeTaskFunction('runtime') + : pool.setDefaultTaskFunction('echo') + change = pending.then( + value => ({ value }), + error => ({ error }) + ) + await original.worker.terminate() + const result = await bounded( + change, + 'Operation remained pending after target exit' + ) + expect(result.error.message).toContain( + 'exited before acknowledging task function operation' + ) + await vi.waitFor( + () => { + expect(pool.workerNodes).toHaveLength(1) + expect(pool.workerNodes[0].worker).not.toBe(original.worker) + expect(pool.workerNodes[0].info.ready).toBe(true) + }, + { timeout: 5000 } + ) + expect( + await bounded(pool.execute({ recovered: true }, 'echo')) + ).toEqual({ recovered: true }) + expect( + await bounded(pool.execute({ retained: true }, 'runtime')) + ).toEqual({ retained: true }) + expect(pool.hasTaskFunction('newRuntime')).toBe(false) + expect(await bounded(pool.addTaskFunction('after', data => data))).toBe( + true + ) + expect(await bounded(pool.execute({ usable: true }, 'after'))).toEqual({ + usable: true, + }) + } finally { + release(gate) + await bounded(Promise.allSettled([blocked])) + await bounded(pool.destroy()) + } + } + ) + + it.each(['add', 'update', 'remove'])( + 'keeps survivors and replacements consistent after a peer exits without an ACK (%s)', + async operation => { + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + workerOptions: { workerData: { blockWorkerKeys: [1] } }, + } + ) + const gate = new SharedArrayBuffer(4) + let blocked + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + if (operation !== 'add') { + await bounded(pool.addTaskFunction('runtime', () => 'old')) + } + const survivor = pool.workerNodes[0] + const departed = pool.workerNodes[1] + const entered = waitForMessage( + departed.worker, + message => message.blocked + ) + blocked = pool.execute({ gate }, 'block').catch(error => error) + await bounded(entered) + const acknowledged = waitForMessage( + survivor.messageChannel.port1, + message => + message.taskFunctionOperationStatus != null && + message.taskFunctionProperties?.name === 'runtime' + ) + const pending = + operation === 'remove' + ? pool.removeTaskFunction('runtime') + : pool.addTaskFunction('runtime', () => 'new') + const change = pending.then( + value => ({ value }), + error => ({ error }) + ) + await bounded(acknowledged) + await departed.worker.terminate() + expect(await bounded(change)).toStrictEqual({ value: true }) + await vi.waitFor( + () => { + expect(pool.workerNodes).toHaveLength(2) + expect(pool.workerNodes[1]).not.toBe(departed) + expect(pool.workerNodes[1].info.ready).toBe(true) + }, + { timeout: 5000 } + ) + await bounded(pool.addTaskFunction('barrier', data => data)) + for (let i = 0; i < 2; i++) { + if (operation === 'remove') { + await expect(pool.execute({}, 'runtime')).rejects.toThrow( + "Task function 'runtime' not found" + ) + } else { + expect(await bounded(pool.execute({}, 'runtime'))).toBe('new') + } + } + } finally { + release(gate) + await bounded(Promise.allSettled([blocked])) + await bounded(pool.destroy()) + } + } + ) + + it('owns inherited registration options and their affinity array', async () => { + const pool = new DynamicThreadPool( + 0, + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + try { + const workerNodeKeys = [0] + const definition = Object.create({ + priority: 0, + taskFunction: () => + process.getBuiltinModule('node:worker_threads').threadId, + workerNodeKeys, + }) + const registration = pool.addTaskFunction('runtime', definition) + workerNodeKeys[0] = 1 + await bounded(registration) + expect(await bounded(pool.execute({}, 'runtime'))).toBe( + pool.workerNodes[0].info.id + ) + } finally { + await bounded(pool.destroy()) + } + }) + + it('orders add and remove calls and continues after a failed mutation', async () => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + const addition = pool.addTaskFunction('runtime', () => 'temporary') + const removal = pool.removeTaskFunction('runtime') + expect(await bounded(Promise.all([addition, removal]))).toEqual([ + true, + true, + ]) + await expect(bounded(pool.execute({}, 'runtime'))).rejects.toBeInstanceOf( + Error + ) + const failure = pool.removeTaskFunction('missing').catch(error => error) + const recovery = pool.addTaskFunction('runtime', () => 'recovered') + expect(await bounded(failure)).toBeInstanceOf(Error) + await bounded(recovery) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('recovered') + } finally { + await bounded(pool.destroy()) + } + }) + + it('captures registration definitions before the caller reuses them', async () => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + const definition = { taskFunction: () => 'original' } + const registration = pool.addTaskFunction('runtime', definition) + definition.taskFunction = () => 'mutated' + await bounded(registration) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('original') + const original = pool.workerNodes[0] + await original.worker.terminate() + await vi.waitFor(() => { + expect(pool.workerNodes).toHaveLength(1) + expect(pool.workerNodes[0]).not.toBe(original) + expect(pool.workerNodes[0].info.ready).toBe(true) + }) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('original') + } finally { + await bounded(pool.destroy()) + } + }) + + it('preserves call order across a reentrant ready callback', async () => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { startWorkers: false } + ) + let newer + try { + await bounded(pool.addTaskFunction('seed', () => 0)) + pool.emitter.once(PoolEvents.ready, () => { + queueMicrotask(() => { + newer = pool.addTaskFunction('runtime', () => 'newer') + }) + }) + pool.start() + const older = pool.addTaskFunction('runtime', () => 'older') + await bounded(older) + await bounded(newer) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('newer') + } finally { + await bounded(pool.destroy()) + } + }) + + it('commits concurrent updates in dispatch order when their unanswered peer exits', async () => { + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + workerOptions: { workerData: { blockWorkerKeys: [1] } }, + } + ) + const gate = new SharedArrayBuffer(4) + let blocked + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'old')) + const survivor = pool.workerNodes[0] + const departed = pool.workerNodes[1] + const entered = waitForMessage( + departed.worker, + message => message.blocked + ) + blocked = pool.execute({ gate }, 'block').catch(error => error) + await bounded(entered) + const acknowledged = waitForMessage( + survivor.messageChannel.port1, + message => + message.taskFunctionOperationStatus != null && + message.taskFunctionProperties?.name === 'runtime' + ) + const changes = [ + pool.addTaskFunction('runtime', () => 'older'), + pool.addTaskFunction('runtime', () => 'newer'), + ].map(promise => + promise.then( + value => ({ value }), + error => ({ error }) + ) + ) + await bounded(acknowledged) + await departed.worker.terminate() + expect(await bounded(Promise.all(changes))).toStrictEqual([ + { value: true }, + { value: true }, + ]) + await bounded(pool.addTaskFunction('barrier', data => data)) + for (let i = 0; i < 2; i++) { + expect(await bounded(pool.execute({}, 'runtime'))).toBe('newer') + } + } finally { + release(gate) + await bounded(Promise.allSettled([blocked])) + await bounded(pool.destroy()) + } + }) + + it('rejects an unacknowledged operation even when replacement construction throws', async () => { + const { stdout } = await promisify(execFile)( + process.execPath, + [ + '--eval', + ` + const { once } = require('node:events') + const { MessageChannel } = require('node:worker_threads') + const { FixedThreadPool, PoolEvents } = require('./lib/index.cjs') + const { bounded, waitForMessage } = require('./tests/test-utils.cjs') + + const channel = new MessageChannel() + const gate = new SharedArrayBuffer(4) + const exception = Promise.withResolvers() + process.once('uncaughtException', exception.resolve) + let pool + ;(async () => { + try { + pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + workerOptions: { + workerData: { port: channel.port1 }, + transferList: [channel.port1], + }, + } + ) + await bounded(once(pool.emitter, PoolEvents.ready)) + const original = pool.workerNodes[0].worker + const entered = waitForMessage(original, message => message.blocked) + pool.execute({ gate }, 'block').catch(error => error) + await bounded(entered) + const change = pool.setDefaultTaskFunction('echo').then( + value => ({ status: 'fulfilled', value }), + error => ({ status: 'rejected', message: error.message }) + ) + original.terminate().catch(error => { + console.error(error) + process.exitCode = 1 + }) + const error = await bounded(exception.promise) + console.log(JSON.stringify({ + constructionError: error.name, + operation: await bounded(change), + })) + } finally { + Atomics.store(new Int32Array(gate), 0, 1) + Atomics.notify(new Int32Array(gate), 0) + try { + if (pool != null) await bounded(pool.destroy()) + } finally { + channel.port1.close() + channel.port2.close() + process.off('uncaughtException', exception.resolve) + } + } + })().catch(error => { + console.error(error) + process.exitCode = 1 + }) + `, + ], + { timeout: 10000 } + ) + const result = JSON.parse(stdout) + expect(result.constructionError).toBe('DataCloneError') + expect(result.operation.status).toBe('rejected') + expect(result.operation.message).toContain( + "exited before acknowledging task function operation 'default'" + ) + }) + + it('keeps a replacement unready until replayed scheduling metadata is available', async () => { + const replayGate = new SharedArrayBuffer(4) + Atomics.store(new Int32Array(replayGate), 0, 1) + const replayBlocked = Promise.withResolvers() + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + enableTasksQueue: true, + messageHandler: message => { + if (message.replayBlocked) replayBlocked.resolve(message.workerId) + }, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: false, + }, + workerOptions: { workerData: { replayGate } }, + } + ) + const errors = [] + pool.emitter.on('error', error => errors.push(error)) + const holdGate = new SharedArrayBuffer(4) + const tasks = [] + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded( + pool.addTaskFunction('runtime', { + priority: -10, + strategy: WorkerChoiceStrategies.ROUND_ROBIN, + taskFunction: data => data, + }) + ) + const original = pool.workerNodes[0] + Atomics.store(new Int32Array(replayGate), 0, 0) + await original.worker.terminate() + const workerId = await bounded(replayBlocked.promise) + const replacement = pool.workerNodes.find( + node => node.info.id === workerId + ) + await vi.waitFor(() => { + expect(replacement.info.taskFunctionsProperties).toBeDefined() + }) + expect(replacement.info.ready).toBe(false) + release(replayGate) + await vi.waitFor(() => expect(replacement.info.ready).toBe(true)) + expect(pool.hasTaskFunction('runtime')).toBe(true) + expect(pool.listTaskFunctionsProperties()).toContainEqual({ + name: 'runtime', + priority: -10, + strategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + const entered = waitForMessage( + replacement.worker, + message => message.started + ) + const held = pool.execute({ gate: holdGate }, 'hold') + tasks.push(held) + await bounded(entered) + const order = [] + const ordinary = pool.execute({ id: 'ordinary' }, 'echo').then(data => { + order.push(data.id) + return data + }) + const priority = pool + .execute({ id: 'priority' }, 'runtime') + .then(data => { + order.push(data.id) + return data + }) + tasks.push(ordinary, priority) + release(holdGate) + await bounded(Promise.all(tasks)) + expect(order).toStrictEqual(['priority', 'ordinary']) + expect(errors).toStrictEqual([]) + } finally { + release(replayGate) + release(holdGate) + await bounded(Promise.allSettled(tasks)) + await bounded(pool.destroy()) + } + }) + + it('cancels a first-use task without transferring its input while dynamic replay waits', async () => { + const replayGate = new SharedArrayBuffer(4) + const replayBlocked = Promise.withResolvers() + const pool = new DynamicThreadPool( + 0, + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) replayBlocked.resolve() + }, + workerOptions: { workerData: { replayGate } }, + } + ) + const errors = [] + pool.emitter.on('error', error => errors.push(error)) + const tasks = [] + try { + await bounded( + pool.addTaskFunction('runtime', { + priority: -10, + strategy: WorkerChoiceStrategies.ROUND_ROBIN, + taskFunction: data => data, + }) + ) + const controller = new AbortController() + const reason = new Error('Cancelled during runtime replay') + const data = { buffer: new ArrayBuffer(8) } + const cancelled = pool + .execute(data, 'runtime', controller.signal, [data.buffer]) + .then( + value => ({ value }), + error => ({ error }) + ) + tasks.push(cancelled) + const live = pool.execute({ id: 'live' }, 'runtime') + tasks.push(live) + await bounded(replayBlocked.promise) + expect(data.buffer.byteLength).toBe(8) + controller.abort(reason) + expect((await bounded(cancelled)).error).toBe(reason) + expect(data.buffer.byteLength).toBe(8) + release(replayGate) + expect(await bounded(live)).toStrictEqual({ id: 'live' }) + expect(pool.hasTaskFunction('runtime')).toBe(true) + expect(errors).toStrictEqual([]) + } finally { + release(replayGate) + await bounded(Promise.allSettled(tasks)) + await bounded(pool.destroy()) + } + }) + + it('settles first-use execution when destruction interrupts dynamic replay', async () => { + const replayGate = new SharedArrayBuffer(4) + const replayBlocked = Promise.withResolvers() + const pool = new DynamicThreadPool( + 0, + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) replayBlocked.resolve() + }, + workerOptions: { workerData: { replayGate } }, + } + ) + const errors = [] + pool.emitter.on('error', error => errors.push(error)) + let execution + try { + await bounded(pool.addTaskFunction('runtime', data => data)) + const data = { buffer: new ArrayBuffer(8) } + execution = pool.execute(data, 'runtime', undefined, [data.buffer]).then( + value => ({ value }), + error => ({ error }) + ) + await bounded(replayBlocked.promise) + await bounded(pool.destroy()) + expect((await bounded(execution)).error).toBeInstanceOf(Error) + expect(data.buffer.byteLength).toBe(8) + } finally { + release(replayGate) + if (pool.info.started) await bounded(pool.destroy()) + } + }) + + it.each([ + ['runtime', 'replacement'], + ['runtime', 'restart'], + ['echo', 'replacement'], + ['echo', 'restart'], + ])('preserves the %s default after %s', async (name, transition) => { + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN } + ) + const gate = new SharedArrayBuffer(4) + release(gate) + const data = { gate, value: 'selected' } + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + if (name === 'runtime') { + await bounded(pool.addTaskFunction(name, () => 'selected')) + } + await bounded(pool.setDefaultTaskFunction(name)) + await expect( + bounded(pool.setDefaultTaskFunction('missing')) + ).rejects.toBeInstanceOf(Error) + if (transition === 'replacement') { + const original = pool.workerNodes[0] + await original.worker.terminate() + await vi.waitFor(() => { + expect(pool.workerNodes).toHaveLength(2) + expect(pool.workerNodes.includes(original)).toBe(false) + expect(pool.workerNodes.every(node => node.info.ready)).toBe(true) + }) + } else { + await bounded(pool.destroy()) + await expect( + bounded(pool.setDefaultTaskFunction('missing')) + ).resolves.toBe(true) + const ready = once(pool.emitter, PoolEvents.ready) + pool.start() + await bounded(ready) + } + for (let i = 0; i < 2; ++i) { + expect(await bounded(pool.execute(data))).toStrictEqual( + name === 'runtime' ? 'selected' : data + ) + } + } finally { + await bounded(pool.destroy()) + } + }) + + it('clears a selected runtime default only after its successful removal while empty', async () => { + const pool = new FixedThreadPool( + 1, + './tests/worker-files/thread/taskLifecycleWorker.mjs' + ) + const gate = new SharedArrayBuffer(4) + release(gate) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'selected')) + await bounded(pool.setDefaultTaskFunction('runtime')) + await expect( + bounded(pool.removeTaskFunction('runtime')) + ).rejects.toBeInstanceOf(Error) + expect(await bounded(pool.execute({ gate }))).toBe('selected') + await bounded(pool.destroy()) + expect(await bounded(pool.removeTaskFunction('runtime'))).toBe(true) + const ready = once(pool.emitter, PoolEvents.ready) + pool.start() + await bounded(ready) + expect(await bounded(pool.execute({ gate }))).toBe('released') + } finally { + await bounded(pool.destroy()) + } + }) + + it('replays many functions without warnings when another exit shifts its index', async () => { + const replayGate = new SharedArrayBuffer(4) + release(replayGate) + const replayBlocked = Promise.withResolvers() + const warnings = [] + const onWarning = warning => { + if (warning.name === 'MaxListenersExceededWarning') { + warnings.push(warning.message) + } + } + process.on('warning', onWarning) + const pool = new FixedThreadPool( + 2, + './tests/worker-files/thread/taskLifecycleWorker.mjs', + { + messageHandler: message => { + if (message.replayBlocked) replayBlocked.resolve(message.workerId) + }, + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + workerOptions: { workerData: { replayGate } }, + } + ) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + for (let i = 0; i < 12; i++) { + await bounded(pool.addTaskFunction(`runtime${i}`, () => 'restored')) + } + const [first, second] = pool.workerNodes + Atomics.store(new Int32Array(replayGate), 0, 0) + await second.worker.terminate() + const replayingWorkerId = await bounded(replayBlocked.promise) + const replaying = pool.workerNodes.find( + node => node.info.id === replayingWorkerId + ) + await first.worker.terminate() + expect(pool.workerNodes[0]).toBe(replaying) + release(replayGate) + await vi.waitFor( + () => { + expect(pool.workerNodes).toHaveLength(2) + expect(pool.workerNodes.every(node => node.info.ready)).toBe(true) + }, + { timeout: 5000 } + ) + for (let i = 0; i < 2; i++) { + expect(await bounded(pool.execute({}, 'runtime11'))).toBe('restored') + } + expect(warnings).toStrictEqual([]) + } finally { + release(replayGate) + await bounded(pool.destroy()) + process.off('warning', onWarning) + } + }) +}) diff --git a/tests/pools/task-lifecycle.test.mjs b/tests/pools/task-lifecycle.test.mjs new file mode 100644 index 000000000..5292a781a --- /dev/null +++ b/tests/pools/task-lifecycle.test.mjs @@ -0,0 +1,1172 @@ +import { once } from 'node:events' +import { MessageChannel } from 'node:worker_threads' +import { describe, expect, it, vi } from 'vitest' + +import { + FixedClusterPool, + FixedThreadPool, + PoolEvents, +} from '../../lib/index.mjs' +import { bounded, waitForMessage } from '../test-utils.cjs' + +const workerFile = './tests/worker-files/thread/taskLifecycleWorker.mjs' +const ready = pool => bounded(once(pool.emitter, PoolEvents.ready)) + +const release = gate => { + Atomics.store(new Int32Array(gate), 0, 1) + Atomics.notify(new Int32Array(gate), 0) +} + +const settle = promise => + promise.then( + value => ({ value }), + error => ({ error }) + ) + +const queueOptions = { + enableTasksQueue: true, + tasksQueueOptions: { + concurrency: 1, + tasksStealingOnBackPressure: false, + taskStealing: false, + }, +} + +describe('Task lifecycle regressions', () => { + it.each([ + ['thread', false, FixedThreadPool], + ['thread', true, FixedThreadPool], + ['cluster', true, FixedClusterPool], + ])( + 'preserves operations through an existing protected override (%s, queued: %s)', + async (workerType, queued, Pool) => { + const pool = new Pool( + 1, + workerType === 'thread' + ? workerFile + : './tests/worker-files/cluster/legacyTaskFunctionWorker.mjs', + workerType === 'thread' + ? { + workerOptions: { + workerData: { + legacyTaskFunctionMessages: true, + queuedTaskFunctionOperations: queued, + }, + }, + } + : undefined + ) + try { + await ready(pool) + await expect( + bounded(pool.addTaskFunction('runtime', data => ({ runtime: data }))) + ).resolves.toBe(true) + await expect( + bounded(pool.setDefaultTaskFunction('runtime')) + ).resolves.toBe(true) + expect(await bounded(pool.execute('selected'))).toEqual({ + runtime: 'selected', + }) + await expect( + bounded(pool.setDefaultTaskFunction('missing')) + ).rejects.toBeInstanceOf(Error) + await bounded(pool.addTaskFunction('fallback', data => data)) + await bounded(pool.setDefaultTaskFunction('fallback')) + await expect(bounded(pool.removeTaskFunction('runtime'))).resolves.toBe( + true + ) + expect(pool.hasTaskFunction('runtime')).toBe(false) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([ + ['thread', 'add', FixedThreadPool], + ['thread', 'default', FixedThreadPool], + ['thread', 'remove', FixedThreadPool], + ['cluster', 'add', FixedClusterPool], + ['cluster', 'default', FixedClusterPool], + ['cluster', 'remove', FixedClusterPool], + ])( + 'rejects an uncorrelated %s/%s acknowledgement without blocking later changes', + async (workerType, operation, Pool) => { + const pool = new Pool( + 1, + workerType === 'thread' + ? workerFile + : './tests/worker-files/cluster/legacyTaskFunctionWorker.mjs', + workerType === 'thread' + ? { + workerOptions: { + workerData: { + legacyTaskFunctionMessages: true, + queuedTaskFunctionOperations: true, + statusOnlyTaskFunctionOperation: operation, + }, + }, + } + : { env: { POOLIFIER_TEST_STATUS_ONLY_OPERATION: operation } } + ) + const pending = [] + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + await ready(pool) + const worker = pool.workerNodes[0].worker + if (operation !== 'add') { + await bounded(pool.addTaskFunction('runtime', () => 42)) + await bounded(pool.addTaskFunction('healthy', data => data)) + } + const change = + operation === 'add' + ? pool.addTaskFunction('runtime', () => 42) + : operation === 'default' + ? pool.setDefaultTaskFunction('runtime') + : pool.removeTaskFunction('runtime') + const failed = settle(change) + pending.push(failed) + const following = settle( + operation === 'add' + ? pool.addTaskFunction('next', () => 43) + : operation === 'default' + ? pool.setDefaultTaskFunction('healthy') + : pool.removeTaskFunction('healthy') + ) + pending.push(following) + expect((await bounded(failed)).error).toBeInstanceOf(Error) + expect(await bounded(following)).toEqual({ value: true }) + expect(pool.workerNodes[0].worker).toBe(worker) + if (operation === 'add') { + expect(await bounded(pool.execute({}, 'runtime'))).toBe(42) + expect(await bounded(pool.execute({}, 'next'))).toBe(43) + } else if (operation === 'default') { + expect(await bounded(pool.execute({ selected: true }))).toEqual({ + selected: true, + }) + } else { + expect(pool.hasTaskFunction('runtime')).toBe(false) + expect(pool.hasTaskFunction('healthy')).toBe(false) + await bounded(pool.addTaskFunction('next', () => 43)) + expect(await bounded(pool.execute({}, 'next'))).toBe(43) + } + expect(errors).toEqual([]) + } finally { + await bounded(pool.destroy()) + await bounded(Promise.allSettled(pending)) + } + } + ) + + it('waits for every cohort target after an uncorrelated acknowledgement', async () => { + const replayGates = [new SharedArrayBuffer(4), new SharedArrayBuffer(4)] + const replayCounter = new SharedArrayBuffer(4) + const blocked = new Map() + const allBlocked = Promise.withResolvers() + let created = 0 + const pool = new FixedThreadPool(2, workerFile, { + messageHandler: message => { + if (message.replayBlocked) { + blocked.set(message.workerId, message.replayGeneration) + if (blocked.size === 2) allBlocked.resolve() + } + }, + workerOptions: { + get workerData () { + return { + asyncReplay: true, + legacyTaskFunctionMessages: true, + queuedTaskFunctionOperations: true, + replayCounter, + replayGates, + ...(created++ === 0 && { + statusOnlyTaskFunctionOperation: 'add', + }), + } + }, + }, + }) + let change + try { + await ready(pool) + const [first, second] = pool.workerNodes + const acknowledged = waitForMessage( + first.messageChannel.port1, + message => message.taskFunctionOperationStatus != null + ) + let completed = false + change = settle(pool.addTaskFunction('runtime', () => 42)).then( + result => { + completed = true + return result + } + ) + await bounded(allBlocked.promise) + release(replayGates[blocked.get(first.info.id)]) + await bounded(acknowledged) + expect(await bounded(pool.execute({ barrier: true }, 'echo'))).toEqual({ + barrier: true, + }) + expect(completed).toBe(false) + release(replayGates[blocked.get(second.info.id)]) + expect((await bounded(change)).error).toBeInstanceOf(Error) + expect(await bounded(pool.execute({}, 'runtime'))).toBe(42) + await bounded(pool.addTaskFunction('next', () => 43)) + expect(await bounded(pool.execute({}, 'next'))).toBe(43) + } finally { + replayGates.forEach(release) + await bounded(pool.destroy()) + await bounded(Promise.allSettled([change])) + } + }) + + it.each([false, true])( + 'does not let a delayed override acknowledgement complete a later operation (queued: %s)', + async queued => { + const replayGate = new SharedArrayBuffer(4) + const duplicateAckGate = new SharedArrayBuffer(4) + release(replayGate) + const blocked = Promise.withResolvers() + const duplicate = Promise.withResolvers() + const pool = new FixedThreadPool(1, workerFile, { + messageHandler: message => { + if (message.replayBlocked) blocked.resolve() + if (message.duplicateSent) duplicate.resolve() + }, + workerOptions: { + workerData: { + asyncReplay: true, + duplicateAckGate, + legacyTaskFunctionMessages: true, + queuedTaskFunctionOperations: queued, + replayGate, + }, + }, + }) + let change + try { + await ready(pool) + await bounded(pool.addTaskFunction('first', () => 'first')) + Atomics.store(new Int32Array(replayGate), 0, 0) + let completed = false + change = pool + .addTaskFunction('second', () => 'second') + .then(result => { + completed = true + return result + }) + await bounded(blocked.promise) + release(duplicateAckGate) + await bounded(duplicate.promise) + expect(await bounded(pool.execute({ barrier: true }, 'echo'))).toEqual({ + barrier: true, + }) + expect(completed).toBe(false) + expect(pool.hasTaskFunction('second')).toBe(false) + release(replayGate) + expect(await bounded(change)).toBe(true) + expect(await bounded(pool.execute({}, 'second'))).toBe('second') + } finally { + release(duplicateAckGate) + release(replayGate) + if (change != null) await bounded(settle(change)) + await bounded(pool.destroy()) + } + } + ) + + it('preserves a non-thenable result from an abortable synchronous task', async () => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + await bounded( + pool.addTaskFunction('guardedResult', () => { + let reads = 0 + return Object.defineProperty({ ok: true }, 'then', { + get () { + if (++reads > 2) throw new Error('Repeated result assimilation') + return undefined + }, + }) + }) + ) + expect( + await bounded( + pool.execute({}, 'guardedResult', new AbortController().signal) + ) + ).toEqual({ ok: true }) + expect(await bounded(pool.execute({ recovered: true }, 'echo'))).toEqual({ + recovered: true, + }) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each([false, true])( + 'does not reassimilate an ordinary native promise result (async: %s)', + async asyncCallable => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + const task = async () => { + let reads = 0 + return Object.defineProperty({ ok: true }, 'then', { + get () { + if (++reads > 1) throw new Error('Repeated result assimilation') + return undefined + }, + }) + } + await bounded( + pool.addTaskFunction( + 'native', + asyncCallable + ? task + : () => { + let reads = 0 + return Promise.resolve( + Object.defineProperty({ ok: true }, 'then', { + get () { + if (++reads > 1) { + throw new Error('Repeated result assimilation') + } + return undefined + }, + }) + ) + } + ) + ) + expect(await bounded(pool.execute({}, 'native'))).toEqual({ ok: true }) + expect( + await bounded(pool.execute({ recovered: true }, 'echo')) + ).toEqual({ + recovered: true, + }) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it('reads a decorated native then getter once even when it returns the native method', async () => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + await bounded( + pool.addTaskFunction('decoratedNative', () => { + let reads = 0 + return Object.defineProperty(Promise.resolve(42), 'then', { + get () { + if (++reads > 1) throw new Error('Repeated native then access') + return Promise.prototype.then + }, + }) + }) + ) + expect(await bounded(pool.execute({}, 'decoratedNative'))).toBe(42) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each([false, true])( + 'reports synchronous invocation failure from an async callable (abortable: %s)', + async abortable => { + const pool = new FixedThreadPool(1, workerFile, { + restartWorkerOnError: false, + }) + const unexpected = [] + const failures = [] + pool.emitter.on(PoolEvents.error, error => unexpected.push(error)) + pool.emitter.on(PoolEvents.taskError, error => failures.push(error)) + try { + await ready(pool) + const worker = pool.workerNodes[0].worker + const signal = abortable ? new AbortController().signal : undefined + await expect( + bounded(pool.execute({}, 'proxyFailure', signal)) + ).rejects.toThrow('Proxy task failure') + expect(unexpected).toEqual([]) + expect(failures).toEqual([ + expect.objectContaining({ message: 'Proxy task failure' }), + ]) + expect(await bounded(pool.execute({}, 'proxyFailure', signal))).toBe(2) + expect(pool.workerNodes[0].worker).toBe(worker) + expect(pool.info.executingTasks).toBe(0) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([ + [false, false], + [false, true], + [true, false], + [true, true], + ])( + 'forwards ACK transport errors (sync: %s, cleanup failure: %s)', + async (syncCleanup, failCleanup) => { + const pool = new FixedThreadPool(1, workerFile, { + workerOptions: { + execArgv: ['--unhandled-rejections=warn'], + workerData: { + failAckSend: failCleanup ? 'failure' : 'success', + failCleanup, + syncCleanup, + }, + }, + }) + const errors = [] + const failed = Promise.withResolvers() + pool.emitter.on(PoolEvents.error, error => { + errors.push(error) + failed.resolve(error) + }) + let destruction + try { + await ready(pool) + const worker = pool.workerNodes[0].worker + const exited = new Promise(resolve => worker.once('exit', resolve)) + destruction = pool.destroy() + expect(await bounded(failed.promise)).toMatchObject({ + message: 'Kill ACK transport failed', + }) + await bounded(exited) + await bounded(destruction) + expect(errors).toEqual([ + expect.objectContaining({ message: 'Kill ACK transport failed' }), + ]) + } finally { + if (destruction != null) await bounded(destruction) + if (pool.started) await bounded(pool.destroy()) + } + } + ) + + it.each([ + [ + 'sync', + data => { + structuredClone(data.buffer, { transfer: [data.buffer] }) + throw new Error('Detached input failure') + }, + ], + [ + 'promise', + data => { + structuredClone(data.buffer, { transfer: [data.buffer] }) + return Promise.reject(new Error('Detached input failure')) + }, + ], + [ + 'async', + async data => { + structuredClone(data.buffer, { transfer: [data.buffer] }) + throw new Error('Detached input failure') + }, + ], + ])( + 'reports %s failure after input is detached inside the worker', + async (_kind, fn) => { + const pool = new FixedThreadPool(1, workerFile, { + restartWorkerOnError: false, + }) + const failures = [] + pool.emitter.on(PoolEvents.taskError, error => failures.push(error)) + const unexpected = [] + pool.emitter.on(PoolEvents.error, error => unexpected.push(error)) + try { + await ready(pool) + await bounded(pool.addTaskFunction('detachedFailure', fn)) + const data = { buffer: new ArrayBuffer(8) } + await expect( + bounded(pool.execute(data, 'detachedFailure')) + ).rejects.toThrow('Detached input failure') + expect(failures).toEqual([ + expect.objectContaining({ + error: expect.objectContaining({ + message: 'Detached input failure', + }), + }), + ]) + expect(failures[0].data).toBeUndefined() + expect(unexpected).toEqual([]) + expect(data.buffer.byteLength).toBe(8) + expect(pool.info.executingTasks).toBe(0) + expect( + await bounded(pool.execute({ recovered: true }, 'echo')) + ).toEqual({ + recovered: true, + }) + } finally { + await Promise.all(pool.workerNodes.map(node => node.worker.terminate())) + await bounded(pool.destroy()) + } + } + ) + + it('settles every dispatched task when error reporting itself cannot be cloned', async () => { + const pool = new FixedThreadPool(1, workerFile, { + restartWorkerOnError: false, + workerOptions: { execArgv: ['--unhandled-rejections=warn'] }, + }) + const gate = new SharedArrayBuffer(4) + const tasks = [] + try { + await ready(pool) + await bounded( + pool.addTaskFunction('reportFailure', () => + Promise.reject( + new Error('Error reporting failure', { cause: () => 0 }) + ) + ) + ) + const worker = pool.workerNodes[0].worker + const entered = waitForMessage(worker, message => message.started) + tasks.push(settle(pool.execute({ gate }, 'hold'))) + await bounded(entered) + const failed = new Promise(resolve => + pool.emitter.once(PoolEvents.error, resolve) + ) + const exited = new Promise(resolve => worker.once('exit', resolve)) + tasks.push(settle(pool.execute({}, 'reportFailure'))) + await bounded(failed) + const results = await bounded(Promise.all(tasks)) + for (const result of results) expect(result.error).toBeInstanceOf(Error) + await bounded(exited) + expect(pool.info.executingTasks).toBe(0) + } finally { + release(gate) + await bounded(pool.destroy()) + } + }) + + it.each([new Error('Original queued cancellation'), 'Queued cancellation'])( + 'preserves the queued abort reason: %s', + async reason => { + const pool = new FixedThreadPool(1, workerFile, queueOptions) + const gate = new SharedArrayBuffer(4) + const tasks = [] + try { + await ready(pool) + const started = waitForMessage( + pool.workerNodes[0].worker, + message => message.started + ) + tasks.push(settle(pool.execute({ gate }, 'hold'))) + await bounded(started) + const controller = new AbortController() + const cancelled = settle(pool.execute({}, 'echo', controller.signal)) + tasks.push(cancelled) + controller.abort(reason) + const { error } = await bounded(cancelled) + expect(error.message).toBe( + reason instanceof Error ? reason.message : reason + ) + if (reason instanceof Error) { + expect(error).toBe(reason) + } + expect(pool.info.queuedTasks).toBe(0) + expect(pool.promiseResponseMap.size).toBe(1) + } finally { + release(gate) + await bounded(Promise.all(tasks)) + await bounded(pool.destroy()) + } + } + ) + + it('rejects an already aborted task without dispatching it', async () => { + const messages = [] + const pool = new FixedThreadPool(1, workerFile, { + messageHandler: message => messages.push(message), + }) + try { + await ready(pool) + const controller = new AbortController() + const reason = new Error('Already aborted') + controller.abort(reason) + const result = await bounded( + settle(pool.execute({ ran: true }, 'echo', controller.signal)) + ) + expect(result.error).toBe(reason) + expect(await bounded(pool.execute({ barrier: true }, 'echo'))).toEqual({ + barrier: true, + }) + expect(messages.filter(message => message.started)).toHaveLength(1) + expect(pool.info.executedTasks).toBe(1) + expect(pool.promiseResponseMap.size).toBe(0) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each(['hold', 'promiseHold'])( + 'can abort a running promise task (%s)', + async name => { + const pool = new FixedThreadPool(1, workerFile) + const gate = new SharedArrayBuffer(4) + let running + try { + await ready(pool) + const started = waitForMessage( + pool.workerNodes[0].worker, + message => message.started + ) + const controller = new AbortController() + running = settle(pool.execute({ gate }, name, controller.signal)) + await bounded(started) + const reason = new Error('Cancel running task') + controller.abort(reason) + expect((await bounded(running)).error).toBe(reason) + expect(pool.promiseResponseMap.size).toBe(0) + expect(pool.info.executingTasks).toBe(0) + } finally { + release(gate) + await bounded(running) + await bounded(pool.destroy()) + } + } + ) + + it('checks initial cancellation before cloning task data or selecting a worker', async () => { + const pool = new FixedThreadPool(1, workerFile) + const controller = new AbortController() + const reason = new Error('Cancelled before submission') + controller.abort(reason) + const data = (async function * () { + yield 'not cloneable' + })() + try { + await ready(pool) + const result = await settle(pool.execute(data, 'echo', controller.signal)) + expect(result.error).toBe(reason) + await expect( + pool.mapExecute([{}], 'echo', [controller.signal]) + ).rejects.toBe(reason) + expect(pool.info.executedTasks).toBe(0) + expect(pool.promiseResponseMap.size).toBe(0) + } finally { + await data.return() + await bounded(pool.destroy()) + } + }) + + it('consumes a one-shot abort signal iterable exactly once', async () => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + const controller = new AbortController() + const signals = (function * () { + yield new AbortController().signal + yield controller.signal + })() + expect( + await bounded(pool.mapExecute([{ id: 1 }, { id: 2 }], 'echo', signals)) + ).toEqual([{ id: 1 }, { id: 2 }]) + } finally { + await bounded(pool.destroy()) + } + }) + + it('validates a one-shot signal iterable before dispatching any tasks', async () => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + const signals = (function * () { + yield new AbortController().signal + yield undefined + })() + await expect(pool.mapExecute([{}, {}], 'echo', signals)).rejects.toThrow( + 'abortSignals argument must be an iterable of AbortSignal' + ) + expect(pool.info.executedTasks).toBe(0) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each([false, true])( + 'settles promise-returning functions (abortable: %s)', + async abortable => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + const signal = abortable ? new AbortController().signal : undefined + expect( + await bounded(pool.execute({ ok: true }, 'promiseEcho', signal)) + ).toEqual({ + data: { ok: true }, + executions: 1, + }) + await expect( + bounded(pool.execute({}, 'promiseFailure', signal)) + ).rejects.toThrow('Promise task failure') + expect(pool.promiseResponseMap.size).toBe(0) + expect(pool.info.executingTasks).toBe(0) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([false, true])( + 'settles foreign promise results and rejections (abortable: %s)', + async abortable => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + await pool.addTaskFunction('foreign', data => + process + .getBuiltinModule('node:vm') + .runInNewContext('Promise.resolve(data)', { data }) + ) + await pool.addTaskFunction('foreignFailure', () => + process + .getBuiltinModule('node:vm') + .runInNewContext( + 'Promise.reject(new Error("Foreign task failure"))' + ) + ) + const signal = abortable ? new AbortController().signal : undefined + expect( + await bounded(pool.execute({ foreign: true }, 'foreign', signal)) + ).toEqual({ + foreign: true, + }) + await expect( + bounded(pool.execute({}, 'foreignFailure', signal)) + ).rejects.toThrow('Foreign task failure') + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([ + [false, false], + [false, true], + [true, false], + [true, true], + ])( + 'assimilates decorated promises independently of then return values (async: %s, abortable: %s)', + async (asyncCallable, abortable) => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + if (!asyncCallable) { + await pool.addTaskFunction('decorated', data => { + const promise = data.fail + ? Promise.reject(new Error('Decorated task failure')) + : Promise.resolve(42) + const originalThen = promise.then + let reads = 0 + Object.defineProperty(promise, 'then', { + get () { + if (++reads > 1) throw new Error('Repeated then access') + return function (resolve, reject) { + originalThen.call(this, resolve, reject) + } + }, + }) + return promise + }) + } + const name = asyncCallable ? 'proxyPromise' : 'decorated' + const signal = abortable ? new AbortController().signal : undefined + expect(await bounded(pool.execute({}, name, signal))).toBe(42) + await expect( + bounded(pool.execute({ fail: true }, name, signal)) + ).rejects.toThrow('Decorated task failure') + expect( + await bounded(pool.execute({ recovered: true }, 'echo')) + ).toEqual({ + recovered: true, + }) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([false, true])( + 'uses thenable receivers and reads callable then getters once (abortable: %s)', + async abortable => { + const pool = new FixedThreadPool(1, workerFile) + try { + await ready(pool) + await pool.addTaskFunction('thenable', () => { + const result = { + reads: 0, + get then () { + if (++this.reads > 1) throw new Error('Then getter read twice') + const then = function (resolve) { + resolve({ reads: this.reads, sameReceiver: this === result }) + } + Object.setPrototypeOf(then, null) + for (const name of ['apply', 'bind', 'call']) { + Object.defineProperty(then, name, { + get () { + throw new Error('Unexpected callable property access') + }, + }) + } + return then + }, + } + return result + }) + await pool.addTaskFunction('badThen', () => ({ + get then () { + throw new Error('Then getter failure') + }, + })) + const signal = abortable ? new AbortController().signal : undefined + expect(await bounded(pool.execute({}, 'thenable', signal))).toEqual({ + reads: 1, + sameReceiver: true, + }) + await expect( + bounded(pool.execute({}, 'badThen', signal)) + ).rejects.toThrow('Then getter failure') + expect(await bounded(pool.execute({ next: true }, 'echo'))).toEqual({ + next: true, + }) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it('preserves active cancellation while a foreign promise remains pending', async () => { + const pool = new FixedThreadPool(1, workerFile) + const gate = new SharedArrayBuffer(4) + let running + try { + await ready(pool) + await pool.addTaskFunction('foreignHold', ({ gate }) => + process + .getBuiltinModule('node:vm') + .runInNewContext( + 'new Promise(resolve => { port.postMessage({ started: true }); Atomics.waitAsync(new Int32Array(gate), 0, 0).value.then(() => resolve("released")) })', + { + gate, + port: process.getBuiltinModule('node:worker_threads').parentPort, + } + ) + ) + const entered = waitForMessage( + pool.workerNodes[0].worker, + message => message.started + ) + const controller = new AbortController() + running = settle(pool.execute({ gate }, 'foreignHold', controller.signal)) + await bounded(entered) + const reason = new Error('Cancel foreign promise') + controller.abort(reason) + expect((await bounded(running)).error).toBe(reason) + expect(pool.info.executingTasks).toBe(0) + } finally { + release(gate) + await bounded(running) + await bounded(pool.destroy()) + } + }) + + it.each([ + ['failure', false, 'Expected task failure'], + ['failure', true, 'Expected task failure'], + ['syncFailure', false, 'Expected task failure'], + ['closedFailure', false, 'Expected task failure'], + ['absent', false, "Task function 'absent' not found"], + ])( + 'reports transferred-port errors (%s, abortable: %s)', + async (name, abortable, message) => { + const pool = new FixedThreadPool(1, workerFile) + const channel = new MessageChannel() + const data = { port: channel.port2 } + const errors = [] + pool.emitter.on(PoolEvents.taskError, error => errors.push(error)) + try { + await ready(pool) + const signal = abortable ? new AbortController().signal : undefined + await expect( + bounded(pool.execute(data, name, signal, [channel.port2])) + ).rejects.toThrow(message) + expect(errors).toHaveLength(1) + expect(errors[0].data).toBe(data) + expect(errors[0].name).toBe(name) + expect(errors[0].message).toBe(message) + expect(await bounded(pool.execute({ usable: true }, 'echo'))).toEqual({ + usable: true, + }) + expect(pool.promiseResponseMap.size).toBe(0) + expect(pool.info.executingTasks).toBe(0) + } finally { + channel.port1.close() + channel.port2.close() + await bounded(pool.destroy()) + } + } + ) + + it('preserves worker-mutated error data for tasks without transfers', async () => { + const pool = new FixedThreadPool(1, workerFile) + const data = { input: true } + const errors = [] + pool.emitter.on(PoolEvents.taskError, error => errors.push(error)) + try { + await ready(pool) + await expect( + bounded(pool.execute(data, 'mutatedFailure')) + ).rejects.toThrow('Expected task failure') + expect(errors[0].data).toEqual({ input: true, workerMutation: true }) + expect(data).toEqual({ input: true }) + } finally { + await bounded(pool.destroy()) + } + }) + + it('ignores inherited original input when reporting an ordinary task error', async () => { + const pool = new FixedThreadPool(1, workerFile) + const errors = [] + pool.emitter.on(PoolEvents.taskError, error => errors.push(error)) + try { + await ready(pool) + // eslint-disable-next-line no-extend-native + Object.defineProperty(Object.prototype, 'originalData', { + configurable: true, + value: 'inherited sentinel', + }) + await expect( + bounded(pool.execute({ input: true }, 'mutatedFailure')) + ).rejects.toThrow('Expected task failure') + expect(errors[0].data).toEqual({ input: true, workerMutation: true }) + expect(pool.promiseResponseMap.size).toBe(0) + } finally { + Reflect.deleteProperty(Object.prototype, 'originalData') + await bounded(pool.destroy()) + } + }) + + it.each([ + [false, false], + [false, true], + [true, false], + [true, true], + ])( + 'snapshots a queued transfer list (batch: %s, initially nonempty: %s)', + async (batch, initiallyNonempty) => { + const pool = new FixedThreadPool(1, workerFile, queueOptions) + const gate = new SharedArrayBuffer(4) + const data = { input: true } + const buffer = new ArrayBuffer(8) + const transferList = initiallyNonempty ? [buffer] : [] + const errors = [] + pool.emitter.on(PoolEvents.taskError, error => errors.push(error)) + let held + let failure + try { + await ready(pool) + const started = waitForMessage( + pool.workerNodes[0].worker, + message => message.started + ) + held = settle(pool.execute({ gate }, 'hold')) + await bounded(started) + failure = settle( + batch + ? pool.mapExecute([data], 'mutatedFailure', undefined, transferList) + : pool.execute(data, 'mutatedFailure', undefined, transferList) + ) + if (initiallyNonempty) transferList.length = 0 + else transferList.push(buffer) + release(gate) + expect((await bounded(failure)).error.message).toBe( + 'Expected task failure' + ) + expect(buffer.byteLength).toBe(initiallyNonempty ? 0 : 8) + expect(errors[0].data).toEqual( + initiallyNonempty ? data : { input: true, workerMutation: true } + ) + } finally { + release(gate) + await bounded(Promise.all([held, failure])) + await bounded(pool.destroy()) + } + } + ) + + it.each([ + ['local', false], + ['local', true], + ['foreign', false], + ['foreign', true], + ['decorated', false], + ['decorated', true], + ])( + 'receives the async kill acknowledgement (mode: %s, failure: %s)', + async (mode, failCleanup) => { + const pool = new FixedThreadPool(1, workerFile, { + workerOptions: { + workerData: { + decoratedCleanup: mode === 'decorated', + failCleanup, + foreignCleanup: mode === 'foreign', + }, + }, + }) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + await ready(pool) + const acknowledgements = [] + pool.workerNodes[0].messageChannel.port1.on('message', message => { + if (message.kill != null) acknowledgements.push(message.kill) + }) + await bounded(pool.destroy()) + if (failCleanup) { + expect(errors).toEqual([expect.any(Error)]) + } else { + expect(errors).toEqual([]) + } + expect(acknowledgements).toEqual([failCleanup ? 'failure' : 'success']) + } finally { + if (pool.started) await bounded(pool.destroy()) + } + } + ) + + it.each([ + ['thread', FixedThreadPool], + ['cluster', FixedClusterPool], + ])( + 'correlates concurrent task-function operations (%s)', + async (_workerType, Pool) => { + const pool = new Pool( + 2, + Pool === FixedThreadPool + ? workerFile + : './tests/worker-files/cluster/echoWorker.mjs' + ) + try { + await ready(pool) + const [added, absent] = await bounded( + Promise.all([ + settle(pool.addTaskFunction('added', data => data)), + settle(pool.setDefaultTaskFunction('absent')), + ]) + ) + expect(added.value).toBe(true) + expect(absent.error?.message).toContain( + "Task function operation 'default' failed" + ) + expect( + await bounded(pool.execute({ registered: true }, 'added')) + ).toEqual({ registered: true }) + expect(await bounded(pool.setDefaultTaskFunction('added'))).toBe(true) + expect(await bounded(pool.execute({ default: true }))).toEqual({ + default: true, + }) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([ + ['thread', FixedThreadPool], + ['cluster', FixedClusterPool], + ])( + 'separates concurrent removals of the same function (%s)', + async (_workerType, Pool) => { + const pool = new Pool( + 2, + Pool === FixedThreadPool + ? workerFile + : './tests/worker-files/cluster/echoWorker.mjs' + ) + try { + await ready(pool) + await bounded(pool.addTaskFunction('removable', data => data)) + const [first, second] = await bounded( + Promise.all([ + settle(pool.removeTaskFunction('removable')), + settle(pool.removeTaskFunction('removable')), + ]) + ) + expect(first.value).toBe(true) + expect(second.error).toBeInstanceOf(Error) + expect(pool.hasTaskFunction('removable')).toBe(false) + expect(await bounded(pool.addTaskFunction('next', data => data))).toBe( + true + ) + expect(await bounded(pool.execute({ next: true }, 'next'))).toEqual({ + next: true, + }) + } finally { + await bounded(pool.destroy()) + } + } + ) + + it.each([ + ['thread', FixedThreadPool], + ['cluster', FixedClusterPool], + ])( + 'replays runtime functions after native fixed-worker exit (%s)', + async (_workerType, Pool) => { + const pool = new Pool( + 1, + Pool === FixedThreadPool + ? workerFile + : './tests/worker-files/cluster/echoWorker.mjs' + ) + try { + await ready(pool) + await bounded( + pool.addTaskFunction('runtime', data => ({ runtime: data })) + ) + await bounded( + pool.addTaskFunction('secondRuntime', data => ({ second: data })) + ) + expect(await bounded(pool.execute('before', 'runtime'))).toEqual({ + runtime: 'before', + }) + const previousWorker = pool.workerNodes[0].worker + if (Pool === FixedThreadPool) { + await previousWorker.terminate() + } else { + const exited = new Promise(resolve => + previousWorker.once('exit', resolve) + ) + previousWorker.kill() + await bounded(exited) + } + await vi.waitFor( + () => { + expect(pool.workerNodes).toHaveLength(1) + expect(pool.workerNodes[0].worker).not.toBe(previousWorker) + expect(pool.workerNodes[0].info.ready).toBe(true) + }, + { timeout: 5000 } + ) + expect(pool.workerNodes[0].worker).not.toBe(previousWorker) + expect(await bounded(pool.execute('after', 'runtime'))).toEqual({ + runtime: 'after', + }) + expect(await bounded(pool.execute('after', 'secondRuntime'))).toEqual({ + second: 'after', + }) + expect(pool.hasTaskFunction('runtime')).toBe(true) + expect(pool.hasTaskFunction('secondRuntime')).toBe(true) + } finally { + await bounded(pool.destroy()) + } + } + ) +}) diff --git a/tests/pools/worker-exit.test.mjs b/tests/pools/worker-exit.test.mjs index 07bbbe48c..c016df226 100644 --- a/tests/pools/worker-exit.test.mjs +++ b/tests/pools/worker-exit.test.mjs @@ -649,7 +649,7 @@ for (const [workerType, PoolClass, workerFile] of [ await bounded(exit) const reason = new Error('Cancel redistributed task') controller.abort(reason) - expect(await bounded(queued)).toBeInstanceOf(Error) + expect(await bounded(queued)).toBe(reason) send(destination, { release: 'destination' }) expect(await bounded(destinationResult)).toStrictEqual(destinationData) expect(destination.tasksQueue.size).toBe(0) @@ -1041,6 +1041,108 @@ for (const [workerType, PoolClass, workerFile] of [ }) }) + it.each([ + [false, false], + [false, true], + [true, false], + ])( + 'settles queued cleanup ACKs after owner exit (dynamic: %s, failure: %s)', + async (dynamic, failCleanup) => { + const { stdout } = await promisify(execFile)( + process.execPath, + [ + '--eval', + ` + const { getEventListeners } = require('node:events') + const { FixedThreadPool, DynamicThreadPool } = require('./lib/index.cjs') + const { bounded, waitForMessage } = require('./tests/test-utils.cjs') + const dynamic = ${JSON.stringify(dynamic)} + const failCleanup = ${JSON.stringify(failCleanup)} + const startupGate = new SharedArrayBuffer(4) + const PoolClass = dynamic ? DynamicThreadPool : FixedThreadPool + class DelayedDeliveryPool extends PoolClass { + sendToWorker (workerNodeKey, message) { + super.sendToWorker(workerNodeKey, message) + const node = this.workerNodes[workerNodeKey] + if (message.kill === true && node === this.pauseTarget) { + this.pauseTarget = undefined + const port = node.messageChannel.port1 + const listeners = getEventListeners(port, 'message') + for (const listener of listeners) port.off('message', listener) + node.worker.prependOnceListener('exit', () => { + for (const listener of listeners) port.on('message', listener) + }) + Atomics.store(new Int32Array(startupGate), 0, 1) + Atomics.notify(new Int32Array(startupGate), 0) + } + } + } + const options = { + workerOptions: { + execArgv: [], + workerData: { startupGate, failCleanup }, + }, + } + const workerFile = './tests/worker-files/thread/taskLifecycleWorker.mjs' + let pool + ;(async () => { + try { + pool = dynamic + ? new DelayedDeliveryPool(0, 1, workerFile, options) + : new DelayedDeliveryPool(1, workerFile, options) + if (dynamic) pool.createAndSetupDynamicWorkerNode() + const original = pool.workerNodes[0] + await bounded(waitForMessage(original.worker, message => message.startupBlocked)) + pool.pauseTarget = original + const operation = await bounded(pool.destroyWorkerNode(0)).then( + () => ({ status: 'fulfilled' }), + error => ({ status: 'rejected', message: error.message }) + ) + if (!dynamic && !pool.workerNodes[0].info.ready) { + await bounded(waitForMessage( + pool.workerNodes[0].messageChannel.port1, + message => message.ready === true + )) + } + const response = await bounded(pool.execute({ recovered: true }, 'echo')) + console.log(JSON.stringify({ + operation, + response, + replacement: pool.workerNodes[0] !== original, + })) + } finally { + Atomics.store(new Int32Array(startupGate), 0, 1) + Atomics.notify(new Int32Array(startupGate), 0) + if (pool != null) { + try { + await bounded(pool.destroy()) + } catch (error) { + if (!failCleanup || !error.message.startsWith('Kill message handling failed')) throw error + } + } + } + })().catch(error => { + console.error(error) + process.exitCode = 1 + }) + `, + ], + { timeout: 10000 } + ) + const result = JSON.parse(stdout) + expect(result.operation.status).toBe( + failCleanup ? 'rejected' : 'fulfilled' + ) + if (failCleanup) { + expect(result.operation.message).toContain( + 'Kill message handling failed' + ) + } + expect(result.replacement).toBe(true) + expect(result.response).toStrictEqual({ recovered: true }) + } + ) + for (const [description, message] of [ ['number', { crashWith: 42 }], ['string', { crashWith: 'Worker threw a string' }], diff --git a/tests/worker-files/cluster/legacyTaskFunctionWorker.mjs b/tests/worker-files/cluster/legacyTaskFunctionWorker.mjs new file mode 100644 index 000000000..38f941826 --- /dev/null +++ b/tests/worker-files/cluster/legacyTaskFunctionWorker.mjs @@ -0,0 +1,60 @@ +import { ClusterWorker } from '../../../lib/index.mjs' + +class LegacyTaskFunctionWorker extends ClusterWorker { + sendToMainWorker = (() => { + const send = this.sendToMainWorker + let statusOnlySent = false + return message => { + const response = { ...message } + delete response.taskFunctionOperationId + if ( + !statusOnlySent && + response.taskFunctionOperationStatus != null && + response.taskFunctionOperation === + process.env.POOLIFIER_TEST_STATUS_ONLY_OPERATION + ) { + statusOnlySent = true + delete response.taskFunctionProperties + } + send(response) + } + })() + + constructor (...args) { + super(...args) + this.operationMessages = [] + this.operationAvailable = Promise.withResolvers() + this.consumeTaskFunctionOperations().catch(error => { + queueMicrotask(() => { + throw error + }) + }) + } + + async consumeTaskFunctionOperations () { + for (;;) { + await this.operationAvailable.promise + this.operationAvailable = Promise.withResolvers() + while (this.operationMessages.length > 0) { + super.handleTaskFunctionOperationMessage(this.operationMessages.shift()) + } + } + } + + handleTaskFunctionOperationMessage ({ + taskFunction, + taskFunctionOperation, + taskFunctionProperties, + workerId, + }) { + this.operationMessages.push({ + taskFunction, + taskFunctionOperation, + taskFunctionProperties, + workerId, + }) + this.operationAvailable.resolve() + } +} + +export default new LegacyTaskFunctionWorker(data => data) diff --git a/tests/worker-files/thread/bootstrapAffinityWorker.mjs b/tests/worker-files/thread/bootstrapAffinityWorker.mjs index d1c263b9c..c844a3ae6 100644 --- a/tests/worker-files/thread/bootstrapAffinityWorker.mjs +++ b/tests/worker-files/thread/bootstrapAffinityWorker.mjs @@ -3,7 +3,8 @@ import { parentPort, threadId, workerData } from 'node:worker_threads' import { ThreadWorker } from '../../../lib/index.mjs' const status = new Int32Array(workerData.bootstrapStatus) -const role = Atomics.add(status, 0, 1) +const claimedRole = Atomics.add(status, 0, 1) +const role = workerData.bootstrapRole ?? claimedRole const configuration = workerData.roles[role] const blocked = role === workerData.blockedRole parentPort.postMessage({ blocked, bootstrapRole: role, workerId: threadId }) @@ -33,6 +34,9 @@ const run = async (data, taskFunction) => { const echo = async data => await run(data, 'echo') const pinned = { taskFunction: async data => await run(data, 'pinned'), + ...(configuration.strategy != null && { + strategy: configuration.strategy, + }), ...(configuration.workerNodeKeys != null && { workerNodeKeys: configuration.workerNodeKeys, }), diff --git a/tests/worker-files/thread/taskLifecycleWorker.mjs b/tests/worker-files/thread/taskLifecycleWorker.mjs new file mode 100644 index 000000000..7031b734b --- /dev/null +++ b/tests/worker-files/thread/taskLifecycleWorker.mjs @@ -0,0 +1,285 @@ +import { runInNewContext } from 'node:vm' +import { parentPort, workerData } from 'node:worker_threads' + +import { KillBehaviors, ThreadWorker } from '../../../lib/index.mjs' + +if (workerData?.startupGate != null) { + parentPort.postMessage({ startupBlocked: true }) + Atomics.wait(new Int32Array(workerData.startupGate), 0, 0) +} + +let promiseExecutions = 0 +let proxyInvocations = 0 +const replayGeneration = + workerData?.replayGates == null + ? undefined + : Atomics.add(new Int32Array(workerData.replayCounter), 0, 1) +const replayGate = + replayGeneration == null + ? workerData?.replayGate + : workerData.replayGates[replayGeneration] +const replayState = replayGate == null ? undefined : new Int32Array(replayGate) + +class TaskLifecycleWorker extends ThreadWorker { + sendToMainWorker = (() => { + const send = this.sendToMainWorker + let duplicateScheduled = false + let statusOnlySent = false + return message => { + if (message.kill != null && workerData?.failAckSend === message.kill) { + throw new Error('Kill ACK transport failed') + } + if ( + message.taskFunctionOperationStatus != null && + workerData?.legacyTaskFunctionMessages === true + ) { + message = { + taskFunctionOperation: message.taskFunctionOperation, + taskFunctionOperationStatus: message.taskFunctionOperationStatus, + taskFunctionProperties: message.taskFunctionProperties, + workerError: message.workerError, + } + } + if ( + !statusOnlySent && + message.taskFunctionOperationStatus != null && + message.taskFunctionOperation === + workerData?.statusOnlyTaskFunctionOperation + ) { + statusOnlySent = true + message = { + taskFunctionOperation: message.taskFunctionOperation, + taskFunctionOperationStatus: message.taskFunctionOperationStatus, + workerError: message.workerError, + } + } + send(message) + if ( + message.taskFunctionOperationStatus != null && + message.taskFunctionOperation === + workerData?.exitAfterTaskFunctionOperation + ) { + // eslint-disable-next-line n/no-process-exit + process.exit(0) + } + if ( + message.taskFunctionOperationStatus != null && + workerData?.duplicateAckGate != null && + !duplicateScheduled + ) { + duplicateScheduled = true + Promise.resolve( + Atomics.waitAsync(new Int32Array(workerData.duplicateAckGate), 0, 0) + .value + ) + .then(() => { + send(message) + parentPort.postMessage({ duplicateSent: true }) + return undefined + }) + .catch(error => { + queueMicrotask(() => { + throw error + }) + }) + } + } + })() + + constructor (...args) { + super(...args) + if (workerData?.queuedTaskFunctionOperations === true) { + this.operationMessages = [] + this.operationAvailable = Promise.withResolvers() + this.consumeTaskFunctionOperations().catch(error => { + queueMicrotask(() => { + throw error + }) + }) + } + } + + async consumeTaskFunctionOperations () { + for (;;) { + await this.operationAvailable.promise + this.operationAvailable = Promise.withResolvers() + while (this.operationMessages.length > 0) { + this.processTaskFunctionOperation(this.operationMessages.shift()) + } + } + } + + handleTaskFunctionOperationMessage (message) { + if (workerData?.legacyTaskFunctionMessages === true) { + const { + taskFunction, + taskFunctionOperation, + taskFunctionProperties, + workerId, + } = message + message = { + taskFunction, + taskFunctionOperation, + taskFunctionProperties, + workerId, + } + } + if (workerData?.queuedTaskFunctionOperations === true) { + this.operationMessages.push(message) + this.operationAvailable.resolve() + return + } + this.processTaskFunctionOperation(message) + } + + processTaskFunctionOperation (message) { + if (replayState != null && Atomics.load(replayState, 0) === 0) { + parentPort.postMessage({ + replayBlocked: true, + replayGeneration, + workerId: this.id, + }) + if (workerData?.asyncReplay === true) { + Promise.resolve(Atomics.waitAsync(replayState, 0, 0).value) + .then(() => { + super.handleTaskFunctionOperationMessage(message) + return undefined + }) + .catch(error => { + queueMicrotask(() => { + throw error + }) + }) + return + } + Atomics.wait(replayState, 0, 0) + } + super.handleTaskFunctionOperationMessage(message) + } +} + +export default new TaskLifecycleWorker( + { + block: { + taskFunction: ({ gate }) => { + parentPort.postMessage({ blocked: true }) + Atomics.wait(new Int32Array(gate), 0, 0) + return 'released' + }, + workerNodeKeys: workerData?.blockWorkerKeys, + }, + closedFailure: async ({ port }) => { + port.close() + throw new Error('Expected task failure') + }, + echo: data => { + parentPort.postMessage({ started: true }) + return data + }, + failure: async () => { + throw new Error('Expected task failure') + }, + hold: async ({ gate }) => { + parentPort.postMessage({ started: true }) + await Atomics.waitAsync(new Int32Array(gate), 0, 0).value + return 'released' + }, + mutatedFailure: data => { + data.workerMutation = true + throw new Error('Expected task failure') + }, + promiseEcho: data => + Promise.resolve({ data, executions: ++promiseExecutions }), + promiseFailure: () => Promise.reject(new Error('Promise task failure')), + promiseHold: ({ gate }) => { + parentPort.postMessage({ started: true }) + return Promise.resolve( + Atomics.waitAsync(new Int32Array(gate), 0, 0).value + ).then(() => 'released') + }, + proxyFailure: new Proxy(async () => proxyInvocations, { + apply (target, receiver, args) { + if (++proxyInvocations === 1) throw new Error('Proxy task failure') + return Reflect.apply(target, receiver, args) + }, + }), + proxyPromise: new Proxy( + async data => { + if (data.fail) throw new Error('Decorated task failure') + return 42 + }, + { + apply (target, receiver, args) { + const promise = Reflect.apply(target, receiver, args) + const originalThen = promise.then + let reads = 0 + Object.defineProperty(promise, 'then', { + get () { + if (++reads > 1) throw new Error('Repeated then access') + return function (resolve, reject) { + Reflect.apply(originalThen, this, [resolve, reject]) + } + }, + }) + return promise + }, + } + ), + syncFailure: () => { + throw new Error('Expected task failure') + }, + }, + { + ...(workerData?.hardReplayKill === true && { + killBehavior: KillBehaviors.HARD, + maxInactiveTime: 50, + }), + killHandler: + workerData?.syncCleanup === true + ? () => { + if (workerData?.failCleanup === true) { + throw new Error('Cleanup failure') + } + } + : workerData?.foreignCleanup === true + ? () => + runInNewContext( + 'new Promise((resolve, reject) => setImmediate(() => failCleanup ? reject(new Error("Cleanup failure")) : resolve()))', + { failCleanup: workerData?.failCleanup === true, setImmediate } + ) + : workerData?.decoratedCleanup === true + ? () => { + const promise = new Promise((resolve, reject) => { + setImmediate(() => { + if (workerData?.failCleanup === true) { + reject(new Error('Cleanup failure')) + } else { + resolve() + } + }) + }) + const originalThen = promise.then + const then = function (resolve, reject) { + Reflect.apply(originalThen, this, [resolve, reject]) + } + Object.setPrototypeOf(then, null) + Object.defineProperty(promise, 'then', { value: then }) + return promise + } + : async () => { + if (workerData?.cleanupGate != null) { + parentPort.postMessage({ cleanupStarted: true }) + await Atomics.waitAsync( + new Int32Array(workerData.cleanupGate), + 0, + 0 + ).value + } else { + await new Promise(resolve => setImmediate(resolve)) + } + if (workerData?.failCleanup === true) { + throw new Error('Cleanup failure') + } + }, + } +) diff --git a/tsconfig.build.json b/tsconfig.build.json index c34b38c20..574dca656 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -1,6 +1,7 @@ { "extends": "./tsconfig.json", "compilerOptions": { + "noEmitOnError": true, "rootDir": "./src" }, "exclude": [ diff --git a/tsconfig.json b/tsconfig.json index 6564c98b0..24d974b59 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,7 +1,8 @@ { "$schema": "https://json.schemastore.org/tsconfig", "compilerOptions": { - "target": "ES2022", + "target": "ES2024", + "lib": ["ES2024"], "module": "Node16", "moduleResolution": "Node16", "types": ["node"], -- 2.53.0