src/pools/pool.ts

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