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