src/pools/pool.ts

   1 import type { TransferListItem } from 'node:worker_threads'
   2 import type { EventEmitter, EventEmitterAsyncResource } from 'node:events'
   3 import type { TaskFunction } from '../worker/task-functions'
   4 import type {
   5   ErrorHandler,
   6   ExitHandler,
   7   IWorker,
   8   IWorkerNode,
   9   MessageHandler,
  10   OnlineHandler,
  11   WorkerType
  12 } from './worker'
  13 import type {
  14   WorkerChoiceStrategy,
  15   WorkerChoiceStrategyOptions
  16 } from './selection-strategies/selection-strategies-types'
  17
  18 /**
  19  * Enumeration of pool types.
  20  */
  21 export const PoolTypes = Object.freeze({
  22   /**
  23    * Fixed pool type.
  24    */
  25   fixed: 'fixed',
  26   /**
  27    * Dynamic pool type.
  28    */
  29   dynamic: 'dynamic'
  30 } as const)
  31
  32 /**
  33  * Pool type.
  34  */
  35 export type PoolType = keyof typeof PoolTypes
  36
  37 /**
  38  * Enumeration of pool events.
  39  */
  40 export const PoolEvents = Object.freeze({
  41   ready: 'ready',
  42   busy: 'busy',
  43   full: 'full',
  44   destroy: 'destroy',
  45   error: 'error',
  46   taskError: 'taskError',
  47   backPressure: 'backPressure'
  48 } as const)
  49
  50 /**
  51  * Pool event.
  52  */
  53 export type PoolEvent = keyof typeof PoolEvents
  54
  55 /**
  56  * Pool information.
  57  */
  58 export interface PoolInfo {
  59   readonly version: string
  60   readonly type: PoolType
  61   readonly worker: WorkerType
  62   readonly started: boolean
  63   readonly ready: boolean
  64   readonly strategy: WorkerChoiceStrategy
  65   readonly minSize: number
  66   readonly maxSize: number
  67   /** Pool utilization. */
  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 backPressure?: boolean
  80   readonly stolenTasks?: 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 node tasks queue options.
  98  */
  99 export interface TasksQueueOptions {
 100   /**
 101    * Maximum tasks queue size per worker node flagging it as back pressured.
 102    *
 103    * @defaultValue (pool maximum size)^2
 104    */
 105   readonly size?: number
 106   /**
 107    * Maximum number of tasks that can be executed concurrently on a worker node.
 108    *
 109    * @defaultValue 1
 110    */
 111   readonly concurrency?: number
 112   /**
 113    * Whether to enable task stealing on empty queue.
 114    *
 115    * @defaultValue true
 116    */
 117   readonly taskStealing?: boolean
 118   /**
 119    * Whether to enable tasks stealing under back pressure.
 120    *
 121    * @defaultValue true
 122    */
 123   readonly tasksStealingOnBackPressure?: boolean
 124 }
 125
 126 /**
 127  * Options for a poolifier pool.
 128  *
 129  * @typeParam Worker - Type of worker.
 130  */
 131 export interface PoolOptions<Worker extends IWorker> {
 132   /**
 133    * A function that will listen for online event on each worker.
 134    *
 135    * @defaultValue `() => {}`
 136    */
 137   onlineHandler?: OnlineHandler<Worker>
 138   /**
 139    * A function that will listen for message event on each worker.
 140    *
 141    * @defaultValue `() => {}`
 142    */
 143   messageHandler?: MessageHandler<Worker>
 144   /**
 145    * A function that will listen for error event on each worker.
 146    *
 147    * @defaultValue `() => {}`
 148    */
 149   errorHandler?: ErrorHandler<Worker>
 150   /**
 151    * A function that will listen for exit event on each worker.
 152    *
 153    * @defaultValue `() => {}`
 154    */
 155   exitHandler?: ExitHandler<Worker>
 156   /**
 157    * Whether to start the minimum number of workers at pool initialization.
 158    *
 159    * @defaultValue true
 160    */
 161   startWorkers?: boolean
 162   /**
 163    * The worker choice strategy to use in this pool.
 164    *
 165    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 166    */
 167   workerChoiceStrategy?: WorkerChoiceStrategy
 168   /**
 169    * The worker choice strategy options.
 170    */
 171   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 172   /**
 173    * Restart worker on error.
 174    */
 175   restartWorkerOnError?: boolean
 176   /**
 177    * Pool events integrated with async resource emission.
 178    *
 179    * @defaultValue true
 180    */
 181   enableEvents?: boolean
 182   /**
 183    * Pool worker node tasks queue.
 184    *
 185    * @defaultValue false
 186    */
 187   enableTasksQueue?: boolean
 188   /**
 189    * Pool worker node tasks queue options.
 190    */
 191   tasksQueueOptions?: TasksQueueOptions
 192 }
 193
 194 /**
 195  * Contract definition for a poolifier pool.
 196  *
 197  * @typeParam Worker - Type of worker which manages this pool.
 198  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 199  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 200  */
 201 export interface IPool<
 202   Worker extends IWorker,
 203   Data = unknown,
 204   Response = unknown
 205 > {
 206   /**
 207    * Pool information.
 208    */
 209   readonly info: PoolInfo
 210   /**
 211    * Pool worker nodes.
 212    *
 213    * @internal
 214    */
 215   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 216   /**
 217    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 218    *
 219    * @param workerNodeKey - The worker node key.
 220    * @returns `true` if the worker node has back pressure, `false` otherwise.
 221    * @internal
 222    */
 223   readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
 224   /**
 225    * Event emitter integrated with async resource on which events can be listened to.
 226    * The async tracking tooling identifier is `poolifier:<PoolType>-<WorkerType>-pool`.
 227    *
 228    * Events that can currently be listened to:
 229    *
 230    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
 231    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 232    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 233    * - `'destroy'`: Emitted when the pool is destroyed.
 234    * - `'error'`: Emitted when an uncaught error occurs.
 235    * - `'taskError'`: Emitted when an error occurs while executing a task.
 236    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 237    */
 238   readonly emitter?: EventEmitter | EventEmitterAsyncResource
 239   /**
 240    * Executes the specified function in the worker constructor with the task data input parameter.
 241    *
 242    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 243    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 244    * @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.
 245    * @returns Promise that will be fulfilled when the task is completed.
 246    */
 247   readonly execute: (
 248     data?: Data,
 249     name?: string,
 250     transferList?: TransferListItem[]
 251   ) => Promise<Response>
 252   /**
 253    * Starts the minimum number of workers in this pool.
 254    */
 255   readonly start: () => void
 256   /**
 257    * Terminates all workers in this pool.
 258    */
 259   readonly destroy: () => Promise<void>
 260   /**
 261    * Whether the specified task function exists in this pool.
 262    *
 263    * @param name - The name of the task function.
 264    * @returns `true` if the task function exists, `false` otherwise.
 265    */
 266   readonly hasTaskFunction: (name: string) => boolean
 267   /**
 268    * Adds a task function to this pool.
 269    * If a task function with the same name already exists, it will be overwritten.
 270    *
 271    * @param name - The name of the task function.
 272    * @param fn - The task function.
 273    * @returns `true` if the task function was added, `false` otherwise.
 274    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string or an empty string.
 275    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function.
 276    */
 277   readonly addTaskFunction: (
 278     name: string,
 279     fn: TaskFunction<Data, Response>
 280   ) => Promise<boolean>
 281   /**
 282    * Removes a task function from this pool.
 283    *
 284    * @param name - The name of the task function.
 285    * @returns `true` if the task function was removed, `false` otherwise.
 286    */
 287   readonly removeTaskFunction: (name: string) => Promise<boolean>
 288   /**
 289    * Lists the names of task function available in this pool.
 290    *
 291    * @returns The names of task function available in this pool.
 292    */
 293   readonly listTaskFunctionNames: () => string[]
 294   /**
 295    * Sets the default task function in this pool.
 296    *
 297    * @param name - The name of the task function.
 298    * @returns `true` if the default task function was set, `false` otherwise.
 299    */
 300   readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
 301   /**
 302    * Sets the worker choice strategy in this pool.
 303    *
 304    * @param workerChoiceStrategy - The worker choice strategy.
 305    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 306    */
 307   readonly setWorkerChoiceStrategy: (
 308     workerChoiceStrategy: WorkerChoiceStrategy,
 309     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 310   ) => void
 311   /**
 312    * Sets the worker choice strategy options in this pool.
 313    *
 314    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 315    */
 316   readonly setWorkerChoiceStrategyOptions: (
 317     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 318   ) => void
 319   /**
 320    * Enables/disables the worker node tasks queue in this pool.
 321    *
 322    * @param enable - Whether to enable or disable the worker node tasks queue.
 323    * @param tasksQueueOptions - The worker node tasks queue options.
 324    */
 325   readonly enableTasksQueue: (
 326     enable: boolean,
 327     tasksQueueOptions?: TasksQueueOptions
 328   ) => void
 329   /**
 330    * Sets the worker node tasks queue options in this pool.
 331    *
 332    * @param tasksQueueOptions - The worker node tasks queue options.
 333    */
 334   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 335 }