src/pools/pool.ts

   1 import type { ClusterSettings } from 'node:cluster'
   2 import type { EventEmitterAsyncResource } from 'node:events'
   3 import type { TransferListItem, WorkerOptions } from 'node:worker_threads'
   4
   5 import type { TaskFunctionProperties } from '../utility-types.js'
   6 import type {
   7   TaskFunction,
   8   TaskFunctionObject
   9 } from '../worker/task-functions.js'
  10 import type {
  11   WorkerChoiceStrategy,
  12   WorkerChoiceStrategyOptions
  13 } from './selection-strategies/selection-strategies-types.js'
  14 import type {
  15   ErrorHandler,
  16   ExitHandler,
  17   IWorker,
  18   IWorkerNode,
  19   MessageHandler,
  20   OnlineHandler,
  21   WorkerType
  22 } from './worker.js'
  23
  24 /**
  25  * Enumeration of pool types.
  26  */
  27 export const PoolTypes: Readonly<{
  28   fixed: 'fixed'
  29   dynamic: 'dynamic'
  30 }> = Object.freeze({
  31   /**
  32    * Fixed pool type.
  33    */
  34   fixed: 'fixed',
  35   /**
  36    * Dynamic pool type.
  37    */
  38   dynamic: 'dynamic'
  39 } as const)
  40
  41 /**
  42  * Pool type.
  43  */
  44 export type PoolType = keyof typeof PoolTypes
  45
  46 /**
  47  * Enumeration of pool events.
  48  */
  49 export const PoolEvents: Readonly<{
  50   ready: 'ready'
  51   busy: 'busy'
  52   full: 'full'
  53   empty: 'empty'
  54   destroy: 'destroy'
  55   error: 'error'
  56   taskError: 'taskError'
  57   backPressure: 'backPressure'
  58 }> = Object.freeze({
  59   ready: 'ready',
  60   busy: 'busy',
  61   full: 'full',
  62   empty: 'empty',
  63   destroy: 'destroy',
  64   error: 'error',
  65   taskError: 'taskError',
  66   backPressure: 'backPressure'
  67 } as const)
  68
  69 /**
  70  * Pool event.
  71  */
  72 export type PoolEvent = keyof typeof PoolEvents
  73
  74 /**
  75  * Pool information.
  76  */
  77 export interface PoolInfo {
  78   readonly version: string
  79   readonly type: PoolType
  80   readonly worker: WorkerType
  81   readonly started: boolean
  82   readonly ready: boolean
  83   readonly defaultStrategy: WorkerChoiceStrategy
  84   readonly strategyRetries: number
  85   readonly minSize: number
  86   readonly maxSize: number
  87   /** Pool utilization. */
  88   readonly utilization?: number
  89   /** Pool total worker nodes. */
  90   readonly workerNodes: number
  91   /** Pool stealing worker nodes. */
  92   readonly stealingWorkerNodes?: number
  93   /** Pool idle worker nodes. */
  94   readonly idleWorkerNodes: number
  95   /** Pool busy worker nodes. */
  96   readonly busyWorkerNodes: number
  97   readonly executedTasks: number
  98   readonly executingTasks: number
  99   readonly queuedTasks?: number
 100   readonly maxQueuedTasks?: number
 101   readonly backPressure?: boolean
 102   readonly stolenTasks?: number
 103   readonly failedTasks: number
 104   readonly runTime?: {
 105     readonly minimum: number
 106     readonly maximum: number
 107     readonly average?: number
 108     readonly median?: number
 109   }
 110   readonly waitTime?: {
 111     readonly minimum: number
 112     readonly maximum: number
 113     readonly average?: number
 114     readonly median?: number
 115   }
 116   readonly elu?: {
 117     idle: {
 118       readonly minimum: number
 119       readonly maximum: number
 120       readonly average?: number
 121       readonly median?: number
 122     }
 123     active: {
 124       readonly minimum: number
 125       readonly maximum: number
 126       readonly average?: number
 127       readonly median?: number
 128     }
 129     utilization: {
 130       readonly average?: number
 131       readonly median?: number
 132     }
 133   }
 134 }
 135
 136 /**
 137  * Worker node tasks queue options.
 138  */
 139 export interface TasksQueueOptions {
 140   /**
 141    * Maximum tasks queue size per worker node flagging it as back pressured.
 142    *
 143    * @defaultValue (pool maximum size)^2
 144    */
 145   readonly size?: number
 146   /**
 147    * Maximum number of tasks that can be executed concurrently on a worker node.
 148    *
 149    * @defaultValue 1
 150    */
 151   readonly concurrency?: number
 152   /**
 153    * Whether to enable task stealing on idle.
 154    *
 155    * @defaultValue true
 156    */
 157   readonly taskStealing?: boolean
 158   /**
 159    * Whether to enable tasks stealing under back pressure.
 160    *
 161    * @defaultValue false
 162    */
 163   readonly tasksStealingOnBackPressure?: boolean
 164   /**
 165    * Queued tasks finished timeout in milliseconds at worker node termination.
 166    *
 167    * @defaultValue 2000
 168    */
 169   readonly tasksFinishedTimeout?: number
 170 }
 171
 172 /**
 173  * Options for a poolifier pool.
 174  *
 175  * @typeParam Worker - Type of worker.
 176  */
 177 export interface PoolOptions<Worker extends IWorker> {
 178   /**
 179    * A function that will listen for online event on each worker.
 180    *
 181    * @defaultValue `() => {}`
 182    */
 183   onlineHandler?: OnlineHandler<Worker>
 184   /**
 185    * A function that will listen for message event on each worker.
 186    *
 187    * @defaultValue `() => {}`
 188    */
 189   messageHandler?: MessageHandler<Worker>
 190   /**
 191    * A function that will listen for error event on each worker.
 192    *
 193    * @defaultValue `() => {}`
 194    */
 195   errorHandler?: ErrorHandler<Worker>
 196   /**
 197    * A function that will listen for exit event on each worker.
 198    *
 199    * @defaultValue `() => {}`
 200    */
 201   exitHandler?: ExitHandler<Worker>
 202   /**
 203    * Whether to start the minimum number of workers at pool initialization.
 204    *
 205    * @defaultValue true
 206    */
 207   startWorkers?: boolean
 208   /**
 209    * The default worker choice strategy to use in this pool.
 210    *
 211    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 212    */
 213   workerChoiceStrategy?: WorkerChoiceStrategy
 214   /**
 215    * The worker choice strategy options.
 216    */
 217   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 218   /**
 219    * Restart worker on error.
 220    */
 221   restartWorkerOnError?: boolean
 222   /**
 223    * Pool events integrated with async resource emission.
 224    *
 225    * @defaultValue true
 226    */
 227   enableEvents?: boolean
 228   /**
 229    * Pool worker node tasks queue.
 230    *
 231    * @defaultValue false
 232    */
 233   enableTasksQueue?: boolean
 234   /**
 235    * Pool worker node tasks queue options.
 236    */
 237   tasksQueueOptions?: TasksQueueOptions
 238   /**
 239    * Worker options.
 240    *
 241    * @see https://nodejs.org/api/worker_threads.html#new-workerfilename-options
 242    */
 243   workerOptions?: WorkerOptions
 244   /**
 245    * Key/value pairs to add to worker process environment.
 246    *
 247    * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
 248    */
 249   env?: Record<string, unknown>
 250   /**
 251    * Cluster settings.
 252    *
 253    * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
 254    */
 255   settings?: ClusterSettings
 256 }
 257
 258 /**
 259  * Contract definition for a poolifier pool.
 260  *
 261  * @typeParam Worker - Type of worker which manages this pool.
 262  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 263  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 264  */
 265 export interface IPool<
 266   Worker extends IWorker,
 267   Data = unknown,
 268   Response = unknown
 269 > {
 270   /**
 271    * Pool information.
 272    */
 273   readonly info: PoolInfo
 274   /**
 275    * Pool worker nodes.
 276    *
 277    * @internal
 278    */
 279   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 280   /**
 281    * Pool event emitter integrated with async resource.
 282    * The async tracking tooling identifier is `poolifier:<PoolType>-<WorkerType>-pool`.
 283    *
 284    * Events that can currently be listened to:
 285    *
 286    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready. If the pool is dynamic with a minimum number of workers is set to zero, this event is emitted when at least one dynamic worker is ready.
 287    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 288    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 289    * - `'empty'`: Emitted when the pool is dynamic with a minimum number of workers set to zero and the number of workers has reached the minimum size expected.
 290    * - `'destroy'`: Emitted when the pool is destroyed.
 291    * - `'error'`: Emitted when an uncaught error occurs.
 292    * - `'taskError'`: Emitted when an error occurs while executing a task.
 293    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 294    */
 295   readonly emitter?: EventEmitterAsyncResource
 296   /**
 297    * Executes the specified function in the worker constructor with the task data input parameter.
 298    *
 299    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 300    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 301    * @param transferList - An optional array of transferable objects to transfer ownership of. Ownership of the transferred objects is given to the chosen pool's worker_threads worker and they should not be used in the main thread afterwards.
 302    * @returns Promise that will be fulfilled when the task is completed.
 303    */
 304   readonly execute: (
 305     data?: Data,
 306     name?: string,
 307     transferList?: readonly TransferListItem[]
 308   ) => Promise<Response>
 309   /**
 310    * Starts the minimum number of workers in this pool.
 311    */
 312   readonly start: () => void
 313   /**
 314    * Terminates all workers in this pool.
 315    */
 316   readonly destroy: () => Promise<void>
 317   /**
 318    * Whether the specified task function exists in this pool.
 319    *
 320    * @param name - The name of the task function.
 321    * @returns `true` if the task function exists, `false` otherwise.
 322    */
 323   readonly hasTaskFunction: (name: string) => boolean
 324   /**
 325    * Adds a task function to this pool.
 326    * If a task function with the same name already exists, it will be overwritten.
 327    *
 328    * @param name - The name of the task function.
 329    * @param fn - The task function.
 330    * @returns `true` if the task function was added, `false` otherwise.
 331    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string or an empty string.
 332    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function or task function object.
 333    */
 334   readonly addTaskFunction: (
 335     name: string,
 336     fn: TaskFunction<Data, Response> | TaskFunctionObject<Data, Response>
 337   ) => Promise<boolean>
 338   /**
 339    * Removes a task function from this pool.
 340    *
 341    * @param name - The name of the task function.
 342    * @returns `true` if the task function was removed, `false` otherwise.
 343    */
 344   readonly removeTaskFunction: (name: string) => Promise<boolean>
 345   /**
 346    * Lists the properties of task functions available in this pool.
 347    *
 348    * @returns The properties of task functions available in this pool.
 349    */
 350   readonly listTaskFunctionsProperties: () => TaskFunctionProperties[]
 351   /**
 352    * Sets the default task function in this pool.
 353    *
 354    * @param name - The name of the task function.
 355    * @returns `true` if the default task function was set, `false` otherwise.
 356    */
 357   readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
 358   /**
 359    * Sets the default worker choice strategy in this pool.
 360    *
 361    * @param workerChoiceStrategy - The default worker choice strategy.
 362    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 363    */
 364   readonly setWorkerChoiceStrategy: (
 365     workerChoiceStrategy: WorkerChoiceStrategy,
 366     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 367   ) => void
 368   /**
 369    * Sets the worker choice strategy options in this pool.
 370    *
 371    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 372    * @returns `true` if the worker choice strategy options were set, `false` otherwise.
 373    */
 374   readonly setWorkerChoiceStrategyOptions: (
 375     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 376   ) => boolean
 377   /**
 378    * Enables/disables the worker node tasks queue in this pool.
 379    *
 380    * @param enable - Whether to enable or disable the worker node tasks queue.
 381    * @param tasksQueueOptions - The worker node tasks queue options.
 382    */
 383   readonly enableTasksQueue: (
 384     enable: boolean,
 385     tasksQueueOptions?: TasksQueueOptions
 386   ) => void
 387   /**
 388    * Sets the worker node tasks queue options in this pool.
 389    *
 390    * @param tasksQueueOptions - The worker node tasks queue options.
 391    */
 392   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 393 }