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