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