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