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 ready: boolean
  67   readonly strategy: WorkerChoiceStrategy
  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 idle worker nodes. */
  75   readonly idleWorkerNodes: number
  76   /** Pool busy worker nodes. */
  77   readonly busyWorkerNodes: number
  78   readonly executedTasks: number
  79   readonly executingTasks: number
  80   readonly queuedTasks?: number
  81   readonly maxQueuedTasks?: number
  82   readonly backPressure?: boolean
  83   readonly failedTasks: number
  84   readonly runTime?: {
  85     readonly minimum: number
  86     readonly maximum: number
  87     readonly average: number
  88     readonly median?: number
  89   }
  90   readonly waitTime?: {
  91     readonly minimum: number
  92     readonly maximum: number
  93     readonly average: number
  94     readonly median?: number
  95   }
  96 }
  97
  98 /**
  99  * Worker tasks queue options.
 100  */
 101 export interface TasksQueueOptions {
 102   /**
 103    * Maximum number of tasks that can be executed concurrently on a worker.
 104    *
 105    * @defaultValue 1
 106    */
 107   readonly concurrency?: number
 108 }
 109
 110 /**
 111  * Options for a poolifier pool.
 112  *
 113  * @typeParam Worker - Type of worker.
 114  */
 115 export interface PoolOptions<Worker extends IWorker> {
 116   /**
 117    * A function that will listen for online event on each worker.
 118    */
 119   onlineHandler?: OnlineHandler<Worker>
 120   /**
 121    * A function that will listen for message event on each worker.
 122    */
 123   messageHandler?: MessageHandler<Worker>
 124   /**
 125    * A function that will listen for error event on each worker.
 126    */
 127   errorHandler?: ErrorHandler<Worker>
 128   /**
 129    * A function that will listen for exit event on each worker.
 130    */
 131   exitHandler?: ExitHandler<Worker>
 132   /**
 133    * The worker choice strategy to use in this pool.
 134    *
 135    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 136    */
 137   workerChoiceStrategy?: WorkerChoiceStrategy
 138   /**
 139    * The worker choice strategy options.
 140    */
 141   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 142   /**
 143    * Restart worker on error.
 144    */
 145   restartWorkerOnError?: boolean
 146   /**
 147    * Pool events emission.
 148    *
 149    * @defaultValue true
 150    */
 151   enableEvents?: boolean
 152   /**
 153    * Pool worker tasks queue.
 154    *
 155    * @defaultValue false
 156    */
 157   enableTasksQueue?: boolean
 158   /**
 159    * Pool worker tasks queue options.
 160    */
 161   tasksQueueOptions?: TasksQueueOptions
 162 }
 163
 164 /**
 165  * Contract definition for a poolifier pool.
 166  *
 167  * @typeParam Worker - Type of worker which manages this pool.
 168  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 169  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 170  */
 171 export interface IPool<
 172   Worker extends IWorker,
 173   Data = unknown,
 174   Response = unknown
 175 > {
 176   /**
 177    * Pool information.
 178    */
 179   readonly info: PoolInfo
 180   /**
 181    * Pool worker nodes.
 182    *
 183    * @internal
 184    */
 185   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 186   /**
 187    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 188    *
 189    * @param workerNodeKey - The worker node key.
 190    * @returns `true` if the worker node has back pressure, `false` otherwise.
 191    * @internal
 192    */
 193   readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
 194   /**
 195    * Emitter on which events can be listened to.
 196    *
 197    * Events that can currently be listened to:
 198    *
 199    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
 200    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing at least one task.
 201    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 202    * - '`destroy`': Emitted when the pool is destroyed.
 203    * - `'error'`: Emitted when an uncaught error occurs.
 204    * - `'taskError'`: Emitted when an error occurs while executing a task.
 205    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= pool maximum size^2).
 206    */
 207   readonly emitter?: PoolEmitter
 208   /**
 209    * Executes the specified function in the worker constructor with the task data input parameter.
 210    *
 211    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 212    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 213    * @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.
 214    * @returns Promise that will be fulfilled when the task is completed.
 215    */
 216   readonly execute: (
 217     data?: Data,
 218     name?: string,
 219     transferList?: TransferListItem[]
 220   ) => Promise<Response>
 221   /**
 222    * Terminates all workers in this pool.
 223    */
 224   readonly destroy: () => Promise<void>
 225   /**
 226    * Lists the names of task function available in this pool.
 227    *
 228    * @returns The names of task function available in this pool.
 229    */
 230   readonly listTaskFunctions: () => string[]
 231   /**
 232    * Sets the worker choice strategy in this pool.
 233    *
 234    * @param workerChoiceStrategy - The worker choice strategy.
 235    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 236    */
 237   readonly setWorkerChoiceStrategy: (
 238     workerChoiceStrategy: WorkerChoiceStrategy,
 239     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 240   ) => void
 241   /**
 242    * Sets the worker choice strategy options in this pool.
 243    *
 244    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 245    */
 246   readonly setWorkerChoiceStrategyOptions: (
 247     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 248   ) => void
 249   /**
 250    * Enables/disables the worker tasks queue in this pool.
 251    *
 252    * @param enable - Whether to enable or disable the worker tasks queue.
 253    * @param tasksQueueOptions - The worker tasks queue options.
 254    */
 255   readonly enableTasksQueue: (
 256     enable: boolean,
 257     tasksQueueOptions?: TasksQueueOptions
 258   ) => void
 259   /**
 260    * Sets the worker tasks queue options in this pool.
 261    *
 262    * @param tasksQueueOptions - The worker tasks queue options.
 263    */
 264   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 265 }