src/pools/worker.ts

   1 import type { MessageChannel } from 'node:worker_threads'
   2 import type { EventEmitter } from 'node:events'
   3 import type { CircularArray } from '../circular-array'
   4 import type { Task } from '../utility-types'
   5
   6 /**
   7  * Callback invoked when the worker has started successfully.
   8  *
   9  * @typeParam Worker - Type of worker.
  10  */
  11 export type OnlineHandler<Worker extends IWorker> = (this: Worker) => void
  12
  13 /**
  14  * Callback invoked if the worker has received a message.
  15  *
  16  * @typeParam Worker - Type of worker.
  17  */
  18 export type MessageHandler<Worker extends IWorker> = (
  19   this: Worker,
  20   message: unknown
  21 ) => void
  22
  23 /**
  24  * Callback invoked if the worker raised an error.
  25  *
  26  * @typeParam Worker - Type of worker.
  27  */
  28 export type ErrorHandler<Worker extends IWorker> = (
  29   this: Worker,
  30   error: Error
  31 ) => void
  32
  33 /**
  34  * Callback invoked when the worker exits successfully.
  35  *
  36  * @typeParam Worker - Type of worker.
  37  */
  38 export type ExitHandler<Worker extends IWorker> = (
  39   this: Worker,
  40   exitCode: number
  41 ) => void
  42
  43 /**
  44  * Measurement statistics.
  45  *
  46  * @internal
  47  */
  48 export interface MeasurementStatistics {
  49   /**
  50    * Measurement aggregate.
  51    */
  52   aggregate?: number
  53   /**
  54    * Measurement minimum.
  55    */
  56   minimum?: number
  57   /**
  58    * Measurement maximum.
  59    */
  60   maximum?: number
  61   /**
  62    * Measurement average.
  63    */
  64   average?: number
  65   /**
  66    * Measurement median.
  67    */
  68   median?: number
  69   /**
  70    * Measurement history.
  71    */
  72   readonly history: CircularArray<number>
  73 }
  74
  75 /**
  76  * Event loop utilization measurement statistics.
  77  *
  78  * @internal
  79  */
  80 export interface EventLoopUtilizationMeasurementStatistics {
  81   readonly idle: MeasurementStatistics
  82   readonly active: MeasurementStatistics
  83   utilization?: number
  84 }
  85
  86 /**
  87  * Task statistics.
  88  *
  89  * @internal
  90  */
  91 export interface TaskStatistics {
  92   /**
  93    * Number of executed tasks.
  94    */
  95   executed: number
  96   /**
  97    * Number of executing tasks.
  98    */
  99   executing: number
 100   /**
 101    * Number of queued tasks.
 102    */
 103   readonly queued: number
 104   /**
 105    * Maximum number of queued tasks.
 106    */
 107   readonly maxQueued?: number
 108   /**
 109    * Number of sequentially stolen tasks.
 110    */
 111   sequentiallyStolen: number
 112   /**
 113    * Number of stolen tasks.
 114    */
 115   stolen: number
 116   /**
 117    * Number of failed tasks.
 118    */
 119   failed: number
 120 }
 121
 122 /**
 123  * Enumeration of worker types.
 124  */
 125 export const WorkerTypes = Object.freeze({
 126   thread: 'thread',
 127   cluster: 'cluster'
 128 } as const)
 129
 130 /**
 131  * Worker type.
 132  */
 133 export type WorkerType = keyof typeof WorkerTypes
 134
 135 /**
 136  * Worker information.
 137  *
 138  * @internal
 139  */
 140 export interface WorkerInfo {
 141   /**
 142    * Worker id.
 143    */
 144   readonly id: number | undefined
 145   /**
 146    * Worker type.
 147    */
 148   readonly type: WorkerType
 149   /**
 150    * Dynamic flag.
 151    */
 152   dynamic: boolean
 153   /**
 154    * Ready flag.
 155    */
 156   ready: boolean
 157   /**
 158    * Task function names.
 159    */
 160   taskFunctionNames?: string[]
 161 }
 162
 163 /**
 164  * Worker usage statistics.
 165  *
 166  * @internal
 167  */
 168 export interface WorkerUsage {
 169   /**
 170    * Tasks statistics.
 171    */
 172   readonly tasks: TaskStatistics
 173   /**
 174    * Tasks runtime statistics.
 175    */
 176   readonly runTime: MeasurementStatistics
 177   /**
 178    * Tasks wait time statistics.
 179    */
 180   readonly waitTime: MeasurementStatistics
 181   /**
 182    * Tasks event loop utilization statistics.
 183    */
 184   readonly elu: EventLoopUtilizationMeasurementStatistics
 185 }
 186
 187 /**
 188  * Worker choice strategy data.
 189  *
 190  * @internal
 191  */
 192 export interface StrategyData {
 193   virtualTaskEndTimestamp?: number
 194 }
 195
 196 /**
 197  * Worker interface.
 198  */
 199 export interface IWorker {
 200   /**
 201    * Cluster worker id.
 202    */
 203   readonly id?: number
 204   /**
 205    * Worker thread worker id.
 206    */
 207   readonly threadId?: number
 208   /**
 209    * Registers an event handler.
 210    *
 211    * @param event - The event.
 212    * @param handler - The event handler.
 213    */
 214   readonly on: (
 215     event: string,
 216     handler:
 217     | OnlineHandler<this>
 218     | MessageHandler<this>
 219     | ErrorHandler<this>
 220     | ExitHandler<this>
 221   ) => void
 222   /**
 223    * Registers once an event handler.
 224    *
 225    * @param event - The event.
 226    * @param handler - The event handler.
 227    */
 228   readonly once: (
 229     event: string,
 230     handler:
 231     | OnlineHandler<this>
 232     | MessageHandler<this>
 233     | ErrorHandler<this>
 234     | ExitHandler<this>
 235   ) => void
 236   /**
 237    * Stop all JavaScript execution in the worker thread as soon as possible.
 238    * Returns a Promise for the exit code that is fulfilled when the `'exit' event` is emitted.
 239    */
 240   readonly terminate?: () => Promise<number>
 241   /**
 242    * Cluster worker disconnect.
 243    */
 244   readonly disconnect?: () => void
 245   /**
 246    * Cluster worker kill.
 247    */
 248   readonly kill?: (signal?: string) => void
 249 }
 250
 251 /**
 252  * Worker node options.
 253  *
 254  * @internal
 255  */
 256 export interface WorkerNodeOptions {
 257   workerOptions?: WorkerOptions
 258   env?: Record<string, unknown>
 259   tasksQueueBackPressureSize: number
 260 }
 261
 262 /**
 263  * Worker node interface.
 264  *
 265  * @typeParam Worker - Type of worker.
 266  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 267  * @internal
 268  */
 269 export interface IWorkerNode<Worker extends IWorker, Data = unknown>
 270   extends EventEmitter {
 271   /**
 272    * Worker.
 273    */
 274   readonly worker: Worker
 275   /**
 276    * Worker info.
 277    */
 278   readonly info: WorkerInfo
 279   /**
 280    * Worker usage statistics.
 281    */
 282   readonly usage: WorkerUsage
 283   /**
 284    * Worker choice strategy data.
 285    * This is used to store data that are specific to the worker choice strategy.
 286    */
 287   strategyData?: StrategyData
 288   /**
 289    * Message channel (worker thread only).
 290    */
 291   readonly messageChannel?: MessageChannel
 292   /**
 293    * Tasks queue back pressure size.
 294    * This is the number of tasks that can be enqueued before the worker node has back pressure.
 295    */
 296   tasksQueueBackPressureSize: number
 297   /**
 298    * Tasks queue size.
 299    *
 300    * @returns The tasks queue size.
 301    */
 302   readonly tasksQueueSize: () => number
 303   /**
 304    * Enqueue task.
 305    *
 306    * @param task - The task to queue.
 307    * @returns The tasks queue size.
 308    */
 309   readonly enqueueTask: (task: Task<Data>) => number
 310   /**
 311    * Prepends a task to the tasks queue.
 312    *
 313    * @param task - The task to prepend.
 314    * @returns The tasks queue size.
 315    */
 316   readonly unshiftTask: (task: Task<Data>) => number
 317   /**
 318    * Dequeue task.
 319    *
 320    * @returns The dequeued task.
 321    */
 322   readonly dequeueTask: () => Task<Data> | undefined
 323   /**
 324    * Pops a task from the tasks queue.
 325    *
 326    * @returns The popped task.
 327    */
 328   readonly popTask: () => Task<Data> | undefined
 329   /**
 330    * Clears tasks queue.
 331    */
 332   readonly clearTasksQueue: () => void
 333   /**
 334    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 335    *
 336    * @returns `true` if the worker node has back pressure, `false` otherwise.
 337    */
 338   readonly hasBackPressure: () => boolean
 339   /**
 340    * Resets usage statistics.
 341    */
 342   readonly resetUsage: () => void
 343   /**
 344    * Terminates the worker node.
 345    */
 346   readonly terminate: () => Promise<void>
 347   /**
 348    * Registers a worker event handler.
 349    *
 350    * @param event - The event.
 351    * @param handler - The event handler.
 352    */
 353   readonly registerWorkerEventHandler: (
 354     event: string,
 355     handler:
 356     | OnlineHandler<Worker>
 357     | MessageHandler<Worker>
 358     | ErrorHandler<Worker>
 359     | ExitHandler<Worker>
 360   ) => void
 361   /**
 362    * Registers once a worker event handler.
 363    *
 364    * @param event - The event.
 365    * @param handler - The event handler.
 366    */
 367   readonly registerOnceWorkerEventHandler: (
 368     event: string,
 369     handler:
 370     | OnlineHandler<Worker>
 371     | MessageHandler<Worker>
 372     | ErrorHandler<Worker>
 373     | ExitHandler<Worker>
 374   ) => void
 375   /**
 376    * Gets task function worker usage statistics.
 377    *
 378    * @param name - The task function name.
 379    * @returns The task function worker usage statistics if the task function worker usage statistics are initialized, `undefined` otherwise.
 380    */
 381   readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined
 382   /**
 383    * Deletes task function worker usage statistics.
 384    *
 385    * @param name - The task function name.
 386    * @returns `true` if the task function worker usage statistics were deleted, `false` otherwise.
 387    */
 388   readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean
 389 }
 390
 391 /**
 392  * Worker node event detail.
 393  *
 394  * @internal
 395  */
 396 export interface WorkerNodeEventDetail {
 397   workerId: number
 398   workerNodeKey?: number
 399 }