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