feat!: add abortable task support (#81)

* feat!: add abortable task support

Signed-off-by: Jérôme Benoit <[email protected]>

* fix: properly handle task abortion worker side

Signed-off-by: Jérôme Benoit <[email protected]>

* fix: do not leak task at abortion

Signed-off-by: Jérôme Benoit <[email protected]>

---------

Signed-off-by: Jérôme Benoit <[email protected]>

Author: jerome-benoit

.vscode/settings.json

@@ -29,6 +29,7 @@
     "inheritDoc",
     "IWRR",
     "jaywcjlove",
+    "Jérôme",
     "lcov",
     "libuv",
     "loglevel",

README.md

@@ -55,6 +55,7 @@ Please consult our [general guidelines](#general-guidelines).
   - Tasks stealing under back pressure ✔
   - Tasks redistribution on worker error ✔
 - Support for sync and async task function ✔
+- Support for abortable task function ✔
 - Support for multiple task functions with per task function queuing priority
   and tasks distribution strategy ✔
 - Support for task functions

benchmarks/benchmarks-utils.mjs

@@ -258,14 +258,13 @@ const fibonacci = (n) => {
 const factorial = (n) => {
   if (n === 0 || n === 1) {
     return 1n
-  } else {
-    n = BigInt(n)
-    let factorial = 1n
-    for (let i = 1n; i <= n; i++) {
-      factorial *= i
-    }
-    return factorial
   }
+  n = BigInt(n)
+  let factorial = 1n
+  for (let i = 1n; i <= n; i++) {
+    factorial *= i
+  }
+  return factorial
 }
 
 const readWriteFiles = (

deno.json

@@ -3,9 +3,7 @@
   "version": "0.4.31",
   "exports": "./src/mod.ts",
   "compilerOptions": {
-    "lib": [
-      "deno.worker"
-    ],
+    "lib": ["deno.worker"],
     "strict": true
   },
   "tasks": {
@@ -26,9 +24,7 @@
     "documentation": "deno doc ./src/mod.ts"
   },
   "test": {
-    "include": [
-      "./tests/**/*.test.mjs"
-    ]
+    "include": ["./tests/**/*.test.mjs"]
   },
   "fmt": {
     "semiColons": false,
@@ -41,18 +37,8 @@
     "@std/testing": "jsr:@std/testing@^1.0.14"
   },
   "publish": {
-    "include": [
-      "LICENSE",
-      "README.md",
-      "deno.json",
-      "src/**/*.ts"
-    ]
+    "include": ["LICENSE", "README.md", "deno.json", "src/**/*.ts"]
   },
   "lock": false,
-  "exclude": [
-    "./coverage",
-    "./dist/browser",
-    "./dist/esm",
-    "./npm"
-  ]
+  "exclude": ["./coverage", "./dist/browser", "./dist/esm", "./npm"]
 }

docs/api.md

@@ -5,8 +5,8 @@
 - [Pool](#pool)
   - [`pool = new FixedThreadPool(numberOfThreads, fileURL, opts)`](#pool--new-fixedthreadpoolnumberofthreads-fileurl-opts)
   - [`pool = new DynamicThreadPool(min, max, fileURL, opts)`](#pool--new-dynamicthreadpoolmin-max-fileurl-opts)
-  - [`pool.execute(data, name, transferList)`](#poolexecutedata-name-transferlist)
-  - [`pool.mapExecute(data, name, transferList)`](#poolmapexecutedata-name-transferlist)
+  - [`pool.execute(data, name, abortSignal, transferList)`](#poolexecutedata-name-abortsignal-transferlist)
+  - [`pool.mapExecute(data, name, abortSignals, transferList)`](#poolmapexecutedata-name-abortsignals-transferlist)
   - [`pool.start()`](#poolstart)
   - [`pool.destroy()`](#pooldestroy)
   - [`pool.hasTaskFunction(name)`](#poolhastaskfunctionname)
@@ -41,25 +41,28 @@ override it in your worker implementation).\
 `fileURL` (mandatory) URL to a file with a worker implementation.\
 `opts` (optional) An object with the pool options properties described below.
 
-### `pool.execute(data, name, transferList)`
+### `pool.execute(data, name, abortSignal, transferList)`
 
 `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-threadworker)
 implementation.
 
 This method is available on both pool implementations and returns a promise with
 the task function execution response.
 
-### `pool.mapExecute(data, name, transferList)`
+### `pool.mapExecute(data, name, abortSignals, transferList)`
 
-`data` Iterable objects that you want to pass to your worker task function
+`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-threadworker))
 worker implementation.
@@ -129,7 +132,6 @@ An object with these properties:
 
 - `workerChoiceStrategy` (optional) - The default worker choice strategy to use
   in this pool:
-
   - `WorkerChoiceStrategies.ROUND_ROBIN`: Submit tasks to worker in a round
     robin fashion
   - `WorkerChoiceStrategies.LEAST_USED`: Submit tasks to the worker with the
@@ -159,7 +161,6 @@ An object with these properties:
 - `workerChoiceStrategyOptions` (optional) - The worker choice strategy options
   object to use in this pool.\
   Properties:
-
   - `measurement` (optional) - The measurement to use in worker choice
     strategies: `runTime`, `waitTime` <!-- or `elu`. -->
   - `runTime` (optional) - Use the tasks
@@ -195,7 +196,6 @@ An object with these properties:
 - `tasksQueueOptions` (optional) - The worker tasks queue options object to use
   in this pool.\
   Properties:
-
   - `size` (optional) - The maximum number of tasks that can be queued on a
     worker before flagging it as back pressured. It must be a positive integer.
   - `concurrency` (optional) - The maximum number of tasks that can be executed

src/mod.ts

@@ -1,5 +1,5 @@
+export type { CircularBuffer } from './circular-buffer.ts'
 export type { AbstractPool } from './pools/abstract-pool.ts'
-export { PoolEvents, PoolTypes } from './pools/pool.ts'
 export type {
   IPool,
   PoolEvent,
@@ -8,7 +8,25 @@ export type {
   PoolType,
   TasksQueueOptions,
 } from './pools/pool.ts'
-export { WorkerTypes } from './pools/worker.ts'
+export { PoolEvents, PoolTypes } from './pools/pool.ts'
+export type {
+  IWorkerChoiceStrategy,
+  Measurement,
+  MeasurementOptions,
+  MeasurementStatisticsRequirements,
+  StrategyPolicy,
+  TaskStatisticsRequirements,
+  WorkerChoiceStrategy,
+  WorkerChoiceStrategyOptions,
+} from './pools/selection-strategies/selection-strategies-types.ts'
+export {
+  Measurements,
+  WorkerChoiceStrategies,
+} from './pools/selection-strategies/selection-strategies-types.ts'
+export type { WorkerChoiceStrategiesContext } from './pools/selection-strategies/worker-choice-strategies-context.ts'
+export { DynamicThreadPool } from './pools/thread/dynamic.ts'
+export type { ThreadPoolOptions } from './pools/thread/fixed.ts'
+export { FixedThreadPool } from './pools/thread/fixed.ts'
 export type {
   ErrorEventHandler,
   EventLoopUtilizationMeasurementStatistics,
@@ -25,40 +43,9 @@ export type {
   WorkerType,
   WorkerUsage,
 } from './pools/worker.ts'
-export {
-  Measurements,
-  WorkerChoiceStrategies,
-} from './pools/selection-strategies/selection-strategies-types.ts'
-export type {
-  IWorkerChoiceStrategy,
-  Measurement,
-  MeasurementOptions,
-  MeasurementStatisticsRequirements,
-  StrategyPolicy,
-  TaskStatisticsRequirements,
-  WorkerChoiceStrategy,
-  WorkerChoiceStrategyOptions,
-} from './pools/selection-strategies/selection-strategies-types.ts'
-export type { WorkerChoiceStrategiesContext } from './pools/selection-strategies/worker-choice-strategies-context.ts'
-export { DynamicThreadPool } from './pools/thread/dynamic.ts'
-export { FixedThreadPool } from './pools/thread/fixed.ts'
-export type { ThreadPoolOptions } from './pools/thread/fixed.ts'
-export type { AbstractWorker } from './worker/abstract-worker.ts'
-export { ThreadWorker } from './worker/thread-worker.ts'
-export { KillBehaviors } from './worker/worker-options.ts'
-export type {
-  KillBehavior,
-  KillHandler,
-  WorkerOptions,
-} from './worker/worker-options.ts'
-export type {
-  TaskAsyncFunction,
-  TaskFunction,
-  TaskFunctionObject,
-  TaskFunctionOperationResult,
-  TaskFunctions,
-  TaskSyncFunction,
-} from './worker/task-functions.ts'
+export { WorkerTypes } from './pools/worker.ts'
+export type { PriorityQueue } from './queues/priority-queue.ts'
+export type { FixedQueueNode, IFixedQueue } from './queues/queue-types.ts'
 export type {
   MessageValue,
   PromiseResponseWrapper,
@@ -69,7 +56,20 @@ export type {
   WorkerStatistics,
   Writable,
 } from './utility-types.ts'
-export type { CircularBuffer } from './circular-buffer.ts'
-export type { PriorityQueue } from './queues/priority-queue.ts'
-export type { FixedQueueNode, IFixedQueue } from './queues/queue-types.ts'
 export { availableParallelism } from './utils.ts'
+export type { AbstractWorker } from './worker/abstract-worker.ts'
+export type {
+  TaskAsyncFunction,
+  TaskFunction,
+  TaskFunctionObject,
+  TaskFunctionOperationResult,
+  TaskFunctions,
+  TaskSyncFunction,
+} from './worker/task-functions.ts'
+export { ThreadWorker } from './worker/thread-worker.ts'
+export type {
+  KillBehavior,
+  KillHandler,
+  WorkerOptions,
+} from './worker/worker-options.ts'
+export { KillBehaviors } from './worker/worker-options.ts'