src/pools/pool.ts

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