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   onlineHandler?: OnlineHandler<Worker>
 141   /**
 142    * A function that will listen for message event on each worker.
 143    */
 144   messageHandler?: MessageHandler<Worker>
 145   /**
 146    * A function that will listen for error event on each worker.
 147    */
 148   errorHandler?: ErrorHandler<Worker>
 149   /**
 150    * A function that will listen for exit event on each worker.
 151    */
 152   exitHandler?: ExitHandler<Worker>
 153   /**
 154    * Whether to start the minimum number of workers at pool initialization.
 155    *
 156    * @defaultValue true
 157    */
 158   startWorkers?: boolean
 159   /**
 160    * The worker choice strategy to use in this pool.
 161    *
 162    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 163    */
 164   workerChoiceStrategy?: WorkerChoiceStrategy
 165   /**
 166    * The worker choice strategy options.
 167    */
 168   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 169   /**
 170    * Restart worker on error.
 171    */
 172   restartWorkerOnError?: boolean
 173   /**
 174    * Pool events emission.
 175    *
 176    * @defaultValue true
 177    */
 178   enableEvents?: boolean
 179   /**
 180    * Pool worker node tasks queue.
 181    *
 182    * @defaultValue false
 183    */
 184   enableTasksQueue?: boolean
 185   /**
 186    * Pool worker node tasks queue options.
 187    */
 188   tasksQueueOptions?: TasksQueueOptions
 189 }
 190
 191 /**
 192  * Contract definition for a poolifier pool.
 193  *
 194  * @typeParam Worker - Type of worker which manages this pool.
 195  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 196  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 197  */
 198 export interface IPool<
 199   Worker extends IWorker,
 200   Data = unknown,
 201   Response = unknown
 202 > {
 203   /**
 204    * Pool information.
 205    */
 206   readonly info: PoolInfo
 207   /**
 208    * Pool worker nodes.
 209    *
 210    * @internal
 211    */
 212   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 213   /**
 214    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 215    *
 216    * @param workerNodeKey - The worker node key.
 217    * @returns `true` if the worker node has back pressure, `false` otherwise.
 218    * @internal
 219    */
 220   readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
 221   /**
 222    * Emitter on which events can be listened to.
 223    *
 224    * Events that can currently be listened to:
 225    *
 226    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
 227    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 228    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 229    * - `'destroy'`: Emitted when the pool is destroyed.
 230    * - `'error'`: Emitted when an uncaught error occurs.
 231    * - `'taskError'`: Emitted when an error occurs while executing a task.
 232    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 233    */
 234   readonly emitter?: PoolEmitter
 235   /**
 236    * Executes the specified function in the worker constructor with the task data input parameter.
 237    *
 238    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 239    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 240    * @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.
 241    * @returns Promise that will be fulfilled when the task is completed.
 242    */
 243   readonly execute: (
 244     data?: Data,
 245     name?: string,
 246     transferList?: TransferListItem[]
 247   ) => Promise<Response>
 248   /**
 249    * Starts the minimum number of workers in this pool.
 250    */
 251   readonly start: () => void
 252   /**
 253    * Terminates all workers in this pool.
 254    */
 255   readonly destroy: () => Promise<void>
 256   /**
 257    * Whether the specified task function exists in this pool.
 258    *
 259    * @param name - The name of the task function.
 260    * @returns `true` if the task function exists, `false` otherwise.
 261    */
 262   readonly hasTaskFunction: (name: string) => boolean
 263   /**
 264    * Adds a task function to this pool.
 265    * If a task function with the same name already exists, it will be overwritten.
 266    *
 267    * @param name - The name of the task function.
 268    * @param taskFunction - The task function.
 269    * @returns `true` if the task function was added, `false` otherwise.
 270    */
 271   readonly addTaskFunction: (
 272     name: string,
 273     taskFunction: TaskFunction<Data, Response>
 274   ) => Promise<boolean>
 275   /**
 276    * Removes a task function from this pool.
 277    *
 278    * @param name - The name of the task function.
 279    * @returns `true` if the task function was removed, `false` otherwise.
 280    */
 281   readonly removeTaskFunction: (name: string) => Promise<boolean>
 282   /**
 283    * Lists the names of task function available in this pool.
 284    *
 285    * @returns The names of task function available in this pool.
 286    */
 287   readonly listTaskFunctionNames: () => string[]
 288   /**
 289    * Sets the default task function in this pool.
 290    *
 291    * @param name - The name of the task function.
 292    * @returns `true` if the default task function was set, `false` otherwise.
 293    */
 294   readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
 295   /**
 296    * Sets the worker choice strategy in this pool.
 297    *
 298    * @param workerChoiceStrategy - The worker choice strategy.
 299    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 300    */
 301   readonly setWorkerChoiceStrategy: (
 302     workerChoiceStrategy: WorkerChoiceStrategy,
 303     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 304   ) => void
 305   /**
 306    * Sets the worker choice strategy options in this pool.
 307    *
 308    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 309    */
 310   readonly setWorkerChoiceStrategyOptions: (
 311     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 312   ) => void
 313   /**
 314    * Enables/disables the worker node tasks queue in this pool.
 315    *
 316    * @param enable - Whether to enable or disable the worker node tasks queue.
 317    * @param tasksQueueOptions - The worker node tasks queue options.
 318    */
 319   readonly enableTasksQueue: (
 320     enable: boolean,
 321     tasksQueueOptions?: TasksQueueOptions
 322   ) => void
 323   /**
 324    * Sets the worker node tasks queue options in this pool.
 325    *
 326    * @param tasksQueueOptions - The worker node tasks queue options.
 327    */
 328   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 329 }