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.js'
   5 import type {
   6   ErrorHandler,
   7   ExitHandler,
   8   IWorker,
   9   IWorkerNode,
  10   MessageHandler,
  11   OnlineHandler,
  12   WorkerType
  13 } from './worker.js'
  14 import type {
  15   WorkerChoiceStrategy,
  16   WorkerChoiceStrategyOptions
  17 } from './selection-strategies/selection-strategies-types.js'
  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    * Event emitter integrated with async resource on which events can be listened to.
 245    * The async tracking tooling identifier is `poolifier:<PoolType>-<WorkerType>-pool`.
 246    *
 247    * Events that can currently be listened to:
 248    *
 249    * - `'ready'`: Emitted when the number of workers created in the pool has reached the minimum size expected and are ready.
 250    * - `'busy'`: Emitted when the number of workers created in the pool has reached the maximum size expected and are executing concurrently their tasks quota.
 251    * - `'full'`: Emitted when the pool is dynamic and the number of workers created has reached the maximum size expected.
 252    * - `'destroy'`: Emitted when the pool is destroyed.
 253    * - `'error'`: Emitted when an uncaught error occurs.
 254    * - `'taskError'`: Emitted when an error occurs while executing a task.
 255    * - `'backPressure'`: Emitted when all worker nodes have back pressure (i.e. their tasks queue is full: queue size \>= maximum queue size).
 256    */
 257   readonly emitter?: EventEmitterAsyncResource
 258   /**
 259    * Executes the specified function in the worker constructor with the task data input parameter.
 260    *
 261    * @param data - The optional task input data for the specified task function. This can only be structured-cloneable data.
 262    * @param name - The optional name of the task function to execute. If not specified, the default task function will be executed.
 263    * @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.
 264    * @returns Promise that will be fulfilled when the task is completed.
 265    */
 266   readonly execute: (
 267     data?: Data,
 268     name?: string,
 269     transferList?: TransferListItem[]
 270   ) => Promise<Response>
 271   /**
 272    * Starts the minimum number of workers in this pool.
 273    */
 274   readonly start: () => void
 275   /**
 276    * Terminates all workers in this pool.
 277    */
 278   readonly destroy: () => Promise<void>
 279   /**
 280    * Whether the specified task function exists in this pool.
 281    *
 282    * @param name - The name of the task function.
 283    * @returns `true` if the task function exists, `false` otherwise.
 284    */
 285   readonly hasTaskFunction: (name: string) => boolean
 286   /**
 287    * Adds a task function to this pool.
 288    * If a task function with the same name already exists, it will be overwritten.
 289    *
 290    * @param name - The name of the task function.
 291    * @param fn - The task function.
 292    * @returns `true` if the task function was added, `false` otherwise.
 293    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `name` parameter is not a string or an empty string.
 294    * @throws {@link https://nodejs.org/api/errors.html#class-typeerror} If the `fn` parameter is not a function.
 295    */
 296   readonly addTaskFunction: (
 297     name: string,
 298     fn: TaskFunction<Data, Response>
 299   ) => Promise<boolean>
 300   /**
 301    * Removes a task function from this pool.
 302    *
 303    * @param name - The name of the task function.
 304    * @returns `true` if the task function was removed, `false` otherwise.
 305    */
 306   readonly removeTaskFunction: (name: string) => Promise<boolean>
 307   /**
 308    * Lists the names of task function available in this pool.
 309    *
 310    * @returns The names of task function available in this pool.
 311    */
 312   readonly listTaskFunctionNames: () => string[]
 313   /**
 314    * Sets the default task function in this pool.
 315    *
 316    * @param name - The name of the task function.
 317    * @returns `true` if the default task function was set, `false` otherwise.
 318    */
 319   readonly setDefaultTaskFunction: (name: string) => Promise<boolean>
 320   /**
 321    * Sets the worker choice strategy in this pool.
 322    *
 323    * @param workerChoiceStrategy - The worker choice strategy.
 324    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 325    */
 326   readonly setWorkerChoiceStrategy: (
 327     workerChoiceStrategy: WorkerChoiceStrategy,
 328     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 329   ) => void
 330   /**
 331    * Sets the worker choice strategy options in this pool.
 332    *
 333    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 334    */
 335   readonly setWorkerChoiceStrategyOptions: (
 336     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 337   ) => void
 338   /**
 339    * Enables/disables the worker node tasks queue in this pool.
 340    *
 341    * @param enable - Whether to enable or disable the worker node tasks queue.
 342    * @param tasksQueueOptions - The worker node tasks queue options.
 343    */
 344   readonly enableTasksQueue: (
 345     enable: boolean,
 346     tasksQueueOptions?: TasksQueueOptions
 347   ) => void
 348   /**
 349    * Sets the worker node tasks queue options in this pool.
 350    *
 351    * @param tasksQueueOptions - The worker node tasks queue options.
 352    */
 353   readonly setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 354 }