src/pools/pool.ts

   1 import { EventEmitter } from 'node:events'
   2 import { type TransferListItem } from 'node:worker_threads'
   3 import type { TaskFunction } from '../worker/task-functions'
   4 import type {
   5   ErrorHandler,
   6   ExitHandler,
   7   IWorker,
   8   IWorkerNode,
   9   MessageHandler,
  10   OnlineHandler,
  11   WorkerType
  12 } from './worker'
  13 import type {
  14   WorkerChoiceStrategy,
  15   WorkerChoiceStrategyOptions
  16 } from './selection-strategies/selection-strategies-types'
  17
  18 /**
  19  * Enumeration of pool types.
  20  */
  21 export const PoolTypes = Object.freeze({
  22   /**
  23    * Fixed pool type.
  24    */
  25   fixed: 'fixed',
  26   /**
  27    * Dynamic pool type.
  28    */
  29   dynamic: 'dynamic'
  30 } as const)
  31
  32 /**
  33  * Pool type.
  34  */
  35 export type PoolType = keyof typeof PoolTypes
  36
  37 /**
  38  * Pool events emitter.
  39  */
  40 export class PoolEmitter extends EventEmitter {}
  41
  42 /**
  43  * Enumeration of pool events.
  44  */
  45 export const PoolEvents = Object.freeze({
  46   ready: 'ready',
  47   busy: 'busy',
  48   full: 'full',
  49   destroy: 'destroy',
  50   error: 'error',
  51   taskError: 'taskError',
  52   backPressure: 'backPressure'
  53 } as const)
  54
  55 /**
  56  * Pool event.
  57  */
  58 export type PoolEvent = keyof typeof PoolEvents
  59
  60 /**
  61  * Pool information.
  62  */
  63 export interface PoolInfo {
  64   readonly version: string
  65   readonly type: PoolType
  66   readonly worker: WorkerType
  67   readonly started: boolean
  68   readonly ready: boolean
  69   readonly strategy: WorkerChoiceStrategy
  70   readonly minSize: number
  71   readonly maxSize: number
  72   /** Pool utilization. */
  73   readonly utilization?: number
  74   /** Pool total worker nodes. */
  75   readonly workerNodes: 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.
 119    *
 120    * @defaultValue true
 121    */
 122   readonly taskStealing?: boolean
 123   /**
 124    * Whether to enable tasks stealing on back pressure.
 125    *
 126    * @defaultValue true
 127    */
 128   readonly tasksStealingOnBackPressure?: boolean
 129 }
 130
 131 /**
 132  * Options for a poolifier pool.
 133  *
 134  * @typeParam Worker - Type of worker.
 135  */
 136 export interface PoolOptions<Worker extends IWorker> {
 137   /**
 138    * A function that will listen for online event on each worker.
 139    *
 140    * @defaultValue `() => {}`
 141    */
 142   onlineHandler?: OnlineHandler<Worker>
 143   /**
 144    * A function that will listen for message event on each worker.
 145    *
 146    * @defaultValue `() => {}`
 147    */
 148   messageHandler?: MessageHandler<Worker>
 149   /**
 150    * A function that will listen for error event on each worker.
 151    *
 152    * @defaultValue `() => {}`
 153    */
 154   errorHandler?: ErrorHandler<Worker>
 155   /**
 156    * A function that will listen for exit event on each worker.
 157    *
 158    * @defaultValue `() => {}`
 159    */
 160   exitHandler?: ExitHandler<Worker>
 161   /**
 162    * Whether to start the minimum number of workers at pool initialization.
 163    *
 164    * @defaultValue true
 165    */
 166   startWorkers?: boolean
 167   /**
 168    * The worker choice strategy to use in this pool.
 169    *
 170    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 171    */
 172   workerChoiceStrategy?: WorkerChoiceStrategy
 173   /**
 174    * The worker choice strategy options.
 175    */
 176   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 177   /**
 178    * Restart worker on error.
 179    */
 180   restartWorkerOnError?: boolean
 181   /**
 182    * Pool events emission.
 183    *
 184    * @defaultValue true
 185    */
 186   enableEvents?: boolean
 187   /**
 188    * Pool worker node tasks queue.
 189    *
 190    * @defaultValue false
 191    */
 192   enableTasksQueue?: boolean
 193   /**
 194    * Pool worker node tasks queue options.
 195    */
 196   tasksQueueOptions?: TasksQueueOptions
 197 }
 198
 199 /**
 200  * Contract definition for a poolifier pool.
 201  *
 202  * @typeParam Worker - Type of worker which manages this pool.
 203  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 204  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 205  */
 206 export interface IPool<
 207   Worker extends IWorker,
 208   Data = unknown,
 209   Response = unknown
 210 > {
 211   /**
 212    * Pool information.
 213    */
 214   readonly info: PoolInfo
 215   /**
 216    * Pool worker nodes.
 217    *
 218    * @internal
 219    */
 220   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 221   /**
 222    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 223    *
 224    * @param workerNodeKey - The worker node key.
 225    * @returns `true` if the worker node has back pressure, `false` otherwise.
 226    * @internal
 227    */
 228   readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
 229   /**
 230    * Emitter on which events can be listened to.
 231    *
 232    * Events that can currently be listened to:
 233    *
 234    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
 235    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 236    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 237    * - `'destroy'`: Emitted when the pool is destroyed.
 238    * - `'error'`: Emitted when an uncaught error occurs.
 239    * - `'taskError'`: Emitted when an error occurs while executing a task.
 240    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 241    */
 242   readonly emitter?: PoolEmitter
 243   /**
 244    * Executes the specified function in the worker constructor with the task data input parameter.
 245    *
 246    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 247    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 248    * @param transferList - An optional array of transferable objects to transfer ownership of. Ownership of the transferred objects is given to the pool's worker_threads worker and they should not be used in the main thread afterwards.
 249    * @returns Promise that will be fulfilled when the task is completed.
 250    */
 251   readonly execute: (
 252     data?: Data,
 253     name?: string,
 254     transferList?: TransferListItem[]
 255   ) => Promise<Response>
 256   /**
 257    * Starts the minimum number of workers in this pool.
 258    */
 259   readonly start: () => void
 260   /**
 261    * Terminates all workers in this pool.
 262    */
 263   readonly destroy: () => Promise<void>
 264   /**
 265    * Whether the specified task function exists in this pool.
 266    *
 267    * @param name - The name of the task function.
 268    * @returns `true` if the task function exists, `false` otherwise.
 269    */
 270   readonly hasTaskFunction: (name: string) => boolean
 271   /**
 272    * Adds a task function to this pool.
 273    * If a task function with the same name already exists, it will be overwritten.
 274    *
 275    * @param name - The name of the task function.
 276    * @param fn - The task function.
 277    * @returns `true` if the task function was added, `false` otherwise.
 278    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string or an empty string.
 279    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function.
 280    */
 281   readonly addTaskFunction: (
 282     name: string,
 283     fn: TaskFunction<Data, Response>
 284   ) => Promise<boolean>
 285   /**
 286    * Removes a task function from this pool.
 287    *
 288    * @param name - The name of the task function.
 289    * @returns `true` if the task function was removed, `false` otherwise.
 290    */
 291   readonly removeTaskFunction: (name: string) => Promise<boolean>
 292   /**
 293    * Lists the names of task function available in this pool.
 294    *
 295    * @returns The names of task function available in this pool.
 296    */
 297   readonly listTaskFunctionNames: () => string[]
 298   /**
 299    * Sets the default task function in this pool.
 300    *
 301    * @param name - The name of the task function.
 302    * @returns `true` if the default task function was set, `false` otherwise.
 303    */
 304   readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
 305   /**
 306    * Sets the worker choice strategy in this pool.
 307    *
 308    * @param workerChoiceStrategy - The worker choice strategy.
 309    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 310    */
 311   readonly setWorkerChoiceStrategy: (
 312     workerChoiceStrategy: WorkerChoiceStrategy,
 313     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 314   ) => void
 315   /**
 316    * Sets the worker choice strategy options in this pool.
 317    *
 318    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 319    */
 320   readonly setWorkerChoiceStrategyOptions: (
 321     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 322   ) => void
 323   /**
 324    * Enables/disables the worker node tasks queue in this pool.
 325    *
 326    * @param enable - Whether to enable or disable the worker node tasks queue.
 327    * @param tasksQueueOptions - The worker node tasks queue options.
 328    */
 329   readonly enableTasksQueue: (
 330     enable: boolean,
 331     tasksQueueOptions?: TasksQueueOptions
 332   ) => void
 333   /**
 334    * Sets the worker node tasks queue options in this pool.
 335    *
 336    * @param tasksQueueOptions - The worker node tasks queue options.
 337    */
 338   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 339 }