src/pools/pool.ts

   1 import type { TransferListItem, WorkerOptions } from 'node:worker_threads'
   2 import type { EventEmitterAsyncResource } from 'node:events'
   3 import type { ClusterSettings } from 'node:cluster'
   4 import type { TaskFunction } from '../worker/task-functions.js'
   5 import type {
   6   ErrorHandler,
   7   ExitHandler,
   8   IWorker,
   9   IWorkerNode,
  10   MessageHandler,
  11   OnlineHandler,
  12   WorkerType
  13 } from './worker.js'
  14 import type {
  15   WorkerChoiceStrategy,
  16   WorkerChoiceStrategyOptions
  17 } from './selection-strategies/selection-strategies-types.js'
  18
  19 /**
  20  * Enumeration of pool types.
  21  */
  22 export const PoolTypes = Object.freeze({
  23   /**
  24    * Fixed pool type.
  25    */
  26   fixed: 'fixed',
  27   /**
  28    * Dynamic pool type.
  29    */
  30   dynamic: 'dynamic'
  31 } as const)
  32
  33 /**
  34  * Pool type.
  35  */
  36 export type PoolType = keyof typeof PoolTypes
  37
  38 /**
  39  * Enumeration of pool events.
  40  */
  41 export const PoolEvents = Object.freeze({
  42   ready: 'ready',
  43   busy: 'busy',
  44   full: 'full',
  45   empty: 'empty',
  46   destroy: 'destroy',
  47   error: 'error',
  48   taskError: 'taskError',
  49   backPressure: 'backPressure'
  50 } as const)
  51
  52 /**
  53  * Pool event.
  54  */
  55 export type PoolEvent = keyof typeof PoolEvents
  56
  57 /**
  58  * Pool information.
  59  */
  60 export interface PoolInfo {
  61   readonly version: string
  62   readonly type: PoolType
  63   readonly worker: WorkerType
  64   readonly started: boolean
  65   readonly ready: boolean
  66   readonly strategy: WorkerChoiceStrategy
  67   readonly strategyRetries: number
  68   readonly minSize: number
  69   readonly maxSize: number
  70   /** Pool utilization. */
  71   readonly utilization?: number
  72   /** Pool total worker nodes. */
  73   readonly workerNodes: number
  74   /** Pool stealing worker nodes. */
  75   readonly stealingWorkerNodes?: number
  76   /** Pool idle worker nodes. */
  77   readonly idleWorkerNodes: number
  78   /** Pool busy worker nodes. */
  79   readonly busyWorkerNodes: number
  80   readonly executedTasks: number
  81   readonly executingTasks: number
  82   readonly queuedTasks?: number
  83   readonly maxQueuedTasks?: number
  84   readonly backPressure?: boolean
  85   readonly stolenTasks?: number
  86   readonly failedTasks: number
  87   readonly runTime?: {
  88     readonly minimum: number
  89     readonly maximum: number
  90     readonly average?: number
  91     readonly median?: number
  92   }
  93   readonly waitTime?: {
  94     readonly minimum: number
  95     readonly maximum: number
  96     readonly average?: number
  97     readonly median?: number
  98   }
  99 }
 100
 101 /**
 102  * Worker node tasks queue options.
 103  */
 104 export interface TasksQueueOptions {
 105   /**
 106    * Maximum tasks queue size per worker node flagging it as back pressured.
 107    *
 108    * @defaultValue (pool maximum size)^2
 109    */
 110   readonly size?: number
 111   /**
 112    * Maximum number of tasks that can be executed concurrently on a worker node.
 113    *
 114    * @defaultValue 1
 115    */
 116   readonly concurrency?: number
 117   /**
 118    * Whether to enable task stealing on idle.
 119    *
 120    * @defaultValue true
 121    */
 122   readonly taskStealing?: boolean
 123   /**
 124    * Whether to enable tasks stealing under back pressure.
 125    *
 126    * @defaultValue true
 127    */
 128   readonly tasksStealingOnBackPressure?: boolean
 129   /**
 130    * Queued tasks finished timeout in milliseconds at worker node termination.
 131    *
 132    * @defaultValue 2000
 133    */
 134   readonly tasksFinishedTimeout?: number
 135 }
 136
 137 /**
 138  * Options for a poolifier pool.
 139  *
 140  * @typeParam Worker - Type of worker.
 141  */
 142 export interface PoolOptions<Worker extends IWorker> {
 143   /**
 144    * A function that will listen for online event on each worker.
 145    *
 146    * @defaultValue `() => {}`
 147    */
 148   onlineHandler?: OnlineHandler<Worker>
 149   /**
 150    * A function that will listen for message event on each worker.
 151    *
 152    * @defaultValue `() => {}`
 153    */
 154   messageHandler?: MessageHandler<Worker>
 155   /**
 156    * A function that will listen for error event on each worker.
 157    *
 158    * @defaultValue `() => {}`
 159    */
 160   errorHandler?: ErrorHandler<Worker>
 161   /**
 162    * A function that will listen for exit event on each worker.
 163    *
 164    * @defaultValue `() => {}`
 165    */
 166   exitHandler?: ExitHandler<Worker>
 167   /**
 168    * Whether to start the minimum number of workers at pool initialization.
 169    *
 170    * @defaultValue true
 171    */
 172   startWorkers?: boolean
 173   /**
 174    * The worker choice strategy to use in this pool.
 175    *
 176    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 177    */
 178   workerChoiceStrategy?: WorkerChoiceStrategy
 179   /**
 180    * The worker choice strategy options.
 181    */
 182   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 183   /**
 184    * Restart worker on error.
 185    */
 186   restartWorkerOnError?: boolean
 187   /**
 188    * Pool events integrated with async resource emission.
 189    *
 190    * @defaultValue true
 191    */
 192   enableEvents?: boolean
 193   /**
 194    * Pool worker node tasks queue.
 195    *
 196    * @defaultValue false
 197    */
 198   enableTasksQueue?: boolean
 199   /**
 200    * Pool worker node tasks queue options.
 201    */
 202   tasksQueueOptions?: TasksQueueOptions
 203   /**
 204    * Worker options.
 205    *
 206    * @see https://nodejs.org/api/worker_threads.html#new-workerfilename-options
 207    */
 208   workerOptions?: WorkerOptions
 209   /**
 210    * Key/value pairs to add to worker process environment.
 211    *
 212    * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
 213    */
 214   env?: Record<string, unknown>
 215   /**
 216    * Cluster settings.
 217    *
 218    * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
 219    */
 220   settings?: ClusterSettings
 221 }
 222
 223 /**
 224  * Contract definition for a poolifier pool.
 225  *
 226  * @typeParam Worker - Type of worker which manages this pool.
 227  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 228  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 229  */
 230 export interface IPool<
 231   Worker extends IWorker,
 232   Data = unknown,
 233   Response = unknown
 234 > {
 235   /**
 236    * Pool information.
 237    */
 238   readonly info: PoolInfo
 239   /**
 240    * Pool worker nodes.
 241    *
 242    * @internal
 243    */
 244   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 245   /**
 246    * Event emitter integrated with async resource on which events can be listened to.
 247    * The async tracking tooling identifier is `poolifier:<PoolType>-<WorkerType>-pool`.
 248    *
 249    * Events that can currently be listened to:
 250    *
 251    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready. If the pool is dynamic with a minimum number of workers is set to zero, this event is emitted when at least one dynamic worker is ready.
 252    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 253    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 254    * - `'empty'`: Emitted when the pool is dynamic with a minimum number of workers set to zero and the number of workers has reached the minimum size expected.
 255    * - `'destroy'`: Emitted when the pool is destroyed.
 256    * - `'error'`: Emitted when an uncaught error occurs.
 257    * - `'taskError'`: Emitted when an error occurs while executing a task.
 258    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 259    */
 260   readonly emitter?: EventEmitterAsyncResource
 261   /**
 262    * Executes the specified function in the worker constructor with the task data input parameter.
 263    *
 264    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 265    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 266    * @param transferList - An 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.
 267    * @returns Promise that will be fulfilled when the task is completed.
 268    */
 269   readonly execute: (
 270     data?: Data,
 271     name?: string,
 272     transferList?: TransferListItem[]
 273   ) => Promise<Response>
 274   /**
 275    * Starts the minimum number of workers in this pool.
 276    */
 277   readonly start: () => void
 278   /**
 279    * Terminates all workers in this pool.
 280    */
 281   readonly destroy: () => Promise<void>
 282   /**
 283    * Whether the specified task function exists in this pool.
 284    *
 285    * @param name - The name of the task function.
 286    * @returns `true` if the task function exists, `false` otherwise.
 287    */
 288   readonly hasTaskFunction: (name: string) => boolean
 289   /**
 290    * Adds a task function to this pool.
 291    * If a task function with the same name already exists, it will be overwritten.
 292    *
 293    * @param name - The name of the task function.
 294    * @param fn - The task function.
 295    * @returns `true` if the task function was added, `false` otherwise.
 296    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string or an empty string.
 297    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function.
 298    */
 299   readonly addTaskFunction: (
 300     name: string,
 301     fn: TaskFunction<Data, Response>
 302   ) => Promise<boolean>
 303   /**
 304    * Removes a task function from this pool.
 305    *
 306    * @param name - The name of the task function.
 307    * @returns `true` if the task function was removed, `false` otherwise.
 308    */
 309   readonly removeTaskFunction: (name: string) => Promise<boolean>
 310   /**
 311    * Lists the names of task function available in this pool.
 312    *
 313    * @returns The names of task function available in this pool.
 314    */
 315   readonly listTaskFunctionNames: () => string[]
 316   /**
 317    * Sets the default task function in this pool.
 318    *
 319    * @param name - The name of the task function.
 320    * @returns `true` if the default task function was set, `false` otherwise.
 321    */
 322   readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
 323   /**
 324    * Sets the worker choice strategy in this pool.
 325    *
 326    * @param workerChoiceStrategy - The worker choice strategy.
 327    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 328    */
 329   readonly setWorkerChoiceStrategy: (
 330     workerChoiceStrategy: WorkerChoiceStrategy,
 331     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 332   ) => void
 333   /**
 334    * Sets the worker choice strategy options in this pool.
 335    *
 336    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 337    */
 338   readonly setWorkerChoiceStrategyOptions: (
 339     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 340   ) => void
 341   /**
 342    * Enables/disables the worker node tasks queue in this pool.
 343    *
 344    * @param enable - Whether to enable or disable the worker node tasks queue.
 345    * @param tasksQueueOptions - The worker node tasks queue options.
 346    */
 347   readonly enableTasksQueue: (
 348     enable: boolean,
 349     tasksQueueOptions?: TasksQueueOptions
 350   ) => void
 351   /**
 352    * Sets the worker node tasks queue options in this pool.
 353    *
 354    * @param tasksQueueOptions - The worker node tasks queue options.
 355    */
 356   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 357 }