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