src/pools/pool.ts

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