src/pools/pool.ts

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