src/pools/pool.ts

   1 import type { TransferListItem } 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 idle worker nodes. */
  73   readonly idleWorkerNodes: number
  74   /** Pool busy worker nodes. */
  75   readonly busyWorkerNodes: number
  76   readonly executedTasks: number
  77   readonly executingTasks: number
  78   readonly queuedTasks?: number
  79   readonly maxQueuedTasks?: number
  80   readonly backPressure?: boolean
  81   readonly stolenTasks?: number
  82   readonly failedTasks: number
  83   readonly runTime?: {
  84     readonly minimum: number
  85     readonly maximum: number
  86     readonly average?: number
  87     readonly median?: number
  88   }
  89   readonly waitTime?: {
  90     readonly minimum: number
  91     readonly maximum: number
  92     readonly average?: number
  93     readonly median?: number
  94   }
  95 }
  96
  97 /**
  98  * Worker node tasks queue options.
  99  */
 100 export interface TasksQueueOptions {
 101   /**
 102    * Maximum tasks queue size per worker node flagging it as back pressured.
 103    *
 104    * @defaultValue (pool maximum size)^2
 105    */
 106   readonly size?: number
 107   /**
 108    * Maximum number of tasks that can be executed concurrently on a worker node.
 109    *
 110    * @defaultValue 1
 111    */
 112   readonly concurrency?: number
 113   /**
 114    * Whether to enable task stealing on idle.
 115    *
 116    * @defaultValue true
 117    */
 118   readonly taskStealing?: boolean
 119   /**
 120    * Whether to enable tasks stealing under back pressure.
 121    *
 122    * @defaultValue true
 123    */
 124   readonly tasksStealingOnBackPressure?: boolean
 125 }
 126
 127 /**
 128  * Options for a poolifier pool.
 129  *
 130  * @typeParam Worker - Type of worker.
 131  */
 132 export interface PoolOptions<Worker extends IWorker> {
 133   /**
 134    * A function that will listen for online event on each worker.
 135    *
 136    * @defaultValue `() => {}`
 137    */
 138   onlineHandler?: OnlineHandler<Worker>
 139   /**
 140    * A function that will listen for message event on each worker.
 141    *
 142    * @defaultValue `() => {}`
 143    */
 144   messageHandler?: MessageHandler<Worker>
 145   /**
 146    * A function that will listen for error event on each worker.
 147    *
 148    * @defaultValue `() => {}`
 149    */
 150   errorHandler?: ErrorHandler<Worker>
 151   /**
 152    * A function that will listen for exit event on each worker.
 153    *
 154    * @defaultValue `() => {}`
 155    */
 156   exitHandler?: ExitHandler<Worker>
 157   /**
 158    * Whether to start the minimum number of workers at pool initialization.
 159    *
 160    * @defaultValue true
 161    */
 162   startWorkers?: boolean
 163   /**
 164    * The worker choice strategy to use in this pool.
 165    *
 166    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 167    */
 168   workerChoiceStrategy?: WorkerChoiceStrategy
 169   /**
 170    * The worker choice strategy options.
 171    */
 172   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 173   /**
 174    * Restart worker on error.
 175    */
 176   restartWorkerOnError?: boolean
 177   /**
 178    * Pool events integrated with async resource emission.
 179    *
 180    * @defaultValue true
 181    */
 182   enableEvents?: boolean
 183   /**
 184    * Pool worker node tasks queue.
 185    *
 186    * @defaultValue false
 187    */
 188   enableTasksQueue?: boolean
 189   /**
 190    * Pool worker node tasks queue options.
 191    */
 192   tasksQueueOptions?: TasksQueueOptions
 193   /**
 194    * Worker options.
 195    *
 196    * @see https://nodejs.org/api/worker_threads.html#new-workerfilename-options
 197    */
 198   workerOptions?: WorkerOptions
 199   /**
 200    * Key/value pairs to add to worker process environment.
 201    *
 202    * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
 203    */
 204   env?: Record<string, unknown>
 205   /**
 206    * Cluster settings.
 207    *
 208    * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
 209    */
 210   settings?: ClusterSettings
 211 }
 212
 213 /**
 214  * Contract definition for a poolifier pool.
 215  *
 216  * @typeParam Worker - Type of worker which manages this pool.
 217  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 218  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
 219  */
 220 export interface IPool<
 221   Worker extends IWorker,
 222   Data = unknown,
 223   Response = unknown
 224 > {
 225   /**
 226    * Pool information.
 227    */
 228   readonly info: PoolInfo
 229   /**
 230    * Pool worker nodes.
 231    *
 232    * @internal
 233    */
 234   readonly workerNodes: Array<IWorkerNode<Worker, Data>>
 235   /**
 236    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 237    *
 238    * @param workerNodeKey - The worker node key.
 239    * @returns `true` if the worker node has back pressure, `false` otherwise.
 240    * @internal
 241    */
 242   readonly hasWorkerNodeBackPressure: (workerNodeKey: number) => boolean
 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 }