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