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