src/pools/pool.ts

   1 import { EventEmitter } from 'node:events'
   2 import { type TransferListItem } from 'node:worker_threads'
   3 import type {
   4   ErrorHandler,
   5   ExitHandler,
   6   IWorker,
   7   IWorkerNode,
   8   MessageHandler,
   9   OnlineHandler,
  10   WorkerType
  11 } from './worker'
  12 import type {
  13   WorkerChoiceStrategy,
  14   WorkerChoiceStrategyOptions
  15 } from './selection-strategies/selection-strategies-types'
  16
  17 /**
  18  * Enumeration of pool types.
  19  */
  20 export const PoolTypes = Object.freeze({
  21   /**
  22    * Fixed pool type.
  23    */
  24   fixed: 'fixed',
  25   /**
  26    * Dynamic pool type.
  27    */
  28   dynamic: 'dynamic'
  29 } as const)
  30
  31 /**
  32  * Pool type.
  33  */
  34 export type PoolType = keyof typeof PoolTypes
  35
  36 /**
  37  * Pool events emitter.
  38  */
  39 export class PoolEmitter extends EventEmitter {}
  40
  41 /**
  42  * Enumeration of pool events.
  43  */
  44 export const PoolEvents = Object.freeze({
  45   ready: 'ready',
  46   busy: 'busy',
  47   full: 'full',
  48   destroy: 'destroy',
  49   error: 'error',
  50   taskError: 'taskError',
  51   backPressure: 'backPressure'
  52 } as const)
  53
  54 /**
  55  * Pool event.
  56  */
  57 export type PoolEvent = keyof typeof PoolEvents
  58
  59 /**
  60  * Pool information.
  61  */
  62 export interface PoolInfo {
  63   readonly version: string
  64   readonly type: PoolType
  65   readonly worker: WorkerType
  66   readonly started: boolean
  67   readonly ready: boolean
  68   readonly strategy: WorkerChoiceStrategy
  69   readonly minSize: number
  70   readonly maxSize: number
  71   /** Pool utilization. */
  72   readonly utilization?: number
  73   /** Pool total worker nodes. */
  74   readonly workerNodes: number
  75   /** Pool idle worker nodes. */
  76   readonly idleWorkerNodes: number
  77   /** Pool busy worker nodes. */
  78   readonly busyWorkerNodes: number
  79   readonly executedTasks: number
  80   readonly executingTasks: number
  81   readonly queuedTasks?: number
  82   readonly maxQueuedTasks?: number
  83   readonly backPressure?: boolean
  84   readonly stolenTasks?: number
  85   readonly failedTasks: number
  86   readonly runTime?: {
  87     readonly minimum: number
  88     readonly maximum: number
  89     readonly average?: number
  90     readonly median?: number
  91   }
  92   readonly waitTime?: {
  93     readonly minimum: number
  94     readonly maximum: number
  95     readonly average?: number
  96     readonly median?: number
  97   }
  98 }
  99
 100 /**
 101  * Worker node tasks queue options.
 102  */
 103 export interface TasksQueueOptions {
 104   /**
 105    * Maximum tasks queue size per worker node flagging it as back pressured.
 106    *
 107    * @defaultValue (pool maximum size)^2
 108    */
 109   readonly size?: number
 110   /**
 111    * Maximum number of tasks that can be executed concurrently on a worker node.
 112    *
 113    * @defaultValue 1
 114    */
 115   readonly concurrency?: number
 116   /**
 117    * Whether to enable tasks stealing.
 118    *
 119    * @defaultValue true
 120    */
 121   readonly tasksStealing?: boolean
 122   /**
 123    * Whether to enable tasks stealing on back pressure.
 124    *
 125    * @defaultValue true
 126    */
 127   readonly tasksStealingOnBackPressure?: boolean
 128 }
 129
 130 /**
 131  * Options for a poolifier pool.
 132  *
 133  * @typeParam Worker - Type of worker.
 134  */
 135 export interface PoolOptions<Worker extends IWorker> {
 136   /**
 137    * A function that will listen for online event on each worker.
 138    */
 139   onlineHandler?: OnlineHandler<Worker>
 140   /**
 141    * A function that will listen for message event on each worker.
 142    */
 143   messageHandler?: MessageHandler<Worker>
 144   /**
 145    * A function that will listen for error event on each worker.
 146    */
 147   errorHandler?: ErrorHandler<Worker>
 148   /**
 149    * A function that will listen for exit event on each worker.
 150    */
 151   exitHandler?: ExitHandler<Worker>
 152   /**
 153    * Whether to start the minimum number of workers at pool initialization.
 154    *
 155    * @defaultValue false
 156    */
 157   startWorkers?: boolean
 158   /**
 159    * The worker choice strategy to use in this pool.
 160    *
 161    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 162    */
 163   workerChoiceStrategy?: WorkerChoiceStrategy
 164   /**
 165    * The worker choice strategy options.
 166    */
 167   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 168   /**
 169    * Restart worker on error.
 170    */
 171   restartWorkerOnError?: boolean
 172   /**
 173    * Pool events emission.
 174    *
 175    * @defaultValue true
 176    */
 177   enableEvents?: boolean
 178   /**
 179    * Pool worker node tasks queue.
 180    *
 181    * @defaultValue false
 182    */
 183   enableTasksQueue?: boolean
 184   /**
 185    * Pool worker node tasks queue options.
 186    */
 187   tasksQueueOptions?: TasksQueueOptions
 188 }
 189
 190 /**
 191  * Contract definition for a poolifier pool.
 192  *
 193  * @typeParam Worker - Type of worker which manages this pool.
 194  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 195  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 196  */
 197 export interface IPool<
 198   Worker extends IWorker,
 199   Data = unknown,
 200   Response = unknown
 201 > {
 202   /**
 203    * Pool information.
 204    */
 205   readonly info: PoolInfo
 206   /**
 207    * Pool worker nodes.
 208    *
 209    * @internal
 210    */
 211   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 212   /**
 213    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 214    *
 215    * @param workerNodeKey - The worker node key.
 216    * @returns `true` if the worker node has back pressure, `false` otherwise.
 217    * @internal
 218    */
 219   readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
 220   /**
 221    * Emitter on which events can be listened to.
 222    *
 223    * Events that can currently be listened to:
 224    *
 225    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
 226    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 227    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 228    * - `'destroy'`: Emitted when the pool is destroyed.
 229    * - `'error'`: Emitted when an uncaught error occurs.
 230    * - `'taskError'`: Emitted when an error occurs while executing a task.
 231    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 232    */
 233   readonly emitter?: PoolEmitter
 234   /**
 235    * Executes the specified function in the worker constructor with the task data input parameter.
 236    *
 237    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 238    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 239    * @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.
 240    * @returns Promise that will be fulfilled when the task is completed.
 241    */
 242   readonly execute: (
 243     data?: Data,
 244     name?: string,
 245     transferList?: TransferListItem[]
 246   ) => Promise<Response>
 247   /**
 248    * Starts the minimum number of workers in this pool.
 249    */
 250   readonly start: () => void
 251   /**
 252    * Terminates all workers in this pool.
 253    */
 254   readonly destroy: () => Promise<void>
 255   /**
 256    * Lists the names of task function available in this pool.
 257    *
 258    * @returns The names of task function available in this pool.
 259    */
 260   readonly listTaskFunctions: () => string[]
 261   /**
 262    * Sets the worker choice strategy in this pool.
 263    *
 264    * @param workerChoiceStrategy - The worker choice strategy.
 265    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 266    */
 267   readonly setWorkerChoiceStrategy: (
 268     workerChoiceStrategy: WorkerChoiceStrategy,
 269     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 270   ) => void
 271   /**
 272    * Sets the worker choice strategy options in this pool.
 273    *
 274    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 275    */
 276   readonly setWorkerChoiceStrategyOptions: (
 277     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 278   ) => void
 279   /**
 280    * Enables/disables the worker node tasks queue in this pool.
 281    *
 282    * @param enable - Whether to enable or disable the worker node tasks queue.
 283    * @param tasksQueueOptions - The worker node tasks queue options.
 284    */
 285   readonly enableTasksQueue: (
 286     enable: boolean,
 287     tasksQueueOptions?: TasksQueueOptions
 288   ) => void
 289   /**
 290    * Sets the worker node tasks queue options in this pool.
 291    *
 292    * @param tasksQueueOptions - The worker node tasks queue options.
 293    */
 294   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 295 }