src/pools/worker.ts

   1 import type { MessageChannel, WorkerOptions } 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    * Stealing flag.
 159    * This flag is set to `true` when worker node is stealing tasks from another worker node.
 160    */
 161   stealing: boolean
 162   /**
 163    * Task function names.
 164    */
 165   taskFunctionNames?: string[]
 166 }
 167
 168 /**
 169  * Worker usage statistics.
 170  *
 171  * @internal
 172  */
 173 export interface WorkerUsage {
 174   /**
 175    * Tasks statistics.
 176    */
 177   readonly tasks: TaskStatistics
 178   /**
 179    * Tasks runtime statistics.
 180    */
 181   readonly runTime: MeasurementStatistics
 182   /**
 183    * Tasks wait time statistics.
 184    */
 185   readonly waitTime: MeasurementStatistics
 186   /**
 187    * Tasks event loop utilization statistics.
 188    */
 189   readonly elu: EventLoopUtilizationMeasurementStatistics
 190 }
 191
 192 /**
 193  * Worker choice strategy data.
 194  *
 195  * @internal
 196  */
 197 export interface StrategyData {
 198   virtualTaskEndTimestamp?: number
 199 }
 200
 201 /**
 202  * Worker interface.
 203  */
 204 export interface IWorker {
 205   /**
 206    * Cluster worker id.
 207    */
 208   readonly id?: number
 209   /**
 210    * Worker thread worker id.
 211    */
 212   readonly threadId?: number
 213   /**
 214    * Registers an event handler.
 215    *
 216    * @param event - The event.
 217    * @param handler - The event handler.
 218    */
 219   readonly on: (
 220     event: string,
 221     handler:
 222     | OnlineHandler<this>
 223     | MessageHandler<this>
 224     | ErrorHandler<this>
 225     | ExitHandler<this>
 226   ) => void
 227   /**
 228    * Registers once an event handler.
 229    *
 230    * @param event - The event.
 231    * @param handler - The event handler.
 232    */
 233   readonly once: (
 234     event: string,
 235     handler:
 236     | OnlineHandler<this>
 237     | MessageHandler<this>
 238     | ErrorHandler<this>
 239     | ExitHandler<this>
 240   ) => void
 241   /**
 242    * Stop all JavaScript execution in the worker thread as soon as possible.
 243    * Returns a Promise for the exit code that is fulfilled when the `'exit' event` is emitted.
 244    */
 245   readonly terminate?: () => Promise<number>
 246   /**
 247    * Cluster worker disconnect.
 248    */
 249   readonly disconnect?: () => void
 250   /**
 251    * Cluster worker kill.
 252    */
 253   readonly kill?: (signal?: string) => void
 254 }
 255
 256 /**
 257  * Worker node options.
 258  *
 259  * @internal
 260  */
 261 export interface WorkerNodeOptions {
 262   workerOptions?: WorkerOptions
 263   env?: Record<string, unknown>
 264   tasksQueueBackPressureSize: number
 265 }
 266
 267 /**
 268  * Worker node interface.
 269  *
 270  * @typeParam Worker - Type of worker.
 271  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 272  * @internal
 273  */
 274 export interface IWorkerNode<Worker extends IWorker, Data = unknown>
 275   extends EventEmitter {
 276   /**
 277    * Worker.
 278    */
 279   readonly worker: Worker
 280   /**
 281    * Worker info.
 282    */
 283   readonly info: WorkerInfo
 284   /**
 285    * Worker usage statistics.
 286    */
 287   readonly usage: WorkerUsage
 288   /**
 289    * Worker choice strategy data.
 290    * This is used to store data that are specific to the worker choice strategy.
 291    */
 292   strategyData?: StrategyData
 293   /**
 294    * Message channel (worker thread only).
 295    */
 296   readonly messageChannel?: MessageChannel
 297   /**
 298    * Tasks queue back pressure size.
 299    * This is the number of tasks that can be enqueued before the worker node has back pressure.
 300    */
 301   tasksQueueBackPressureSize: number
 302   /**
 303    * Tasks queue size.
 304    *
 305    * @returns The tasks queue size.
 306    */
 307   readonly tasksQueueSize: () => number
 308   /**
 309    * Enqueue task.
 310    *
 311    * @param task - The task to queue.
 312    * @returns The tasks queue size.
 313    */
 314   readonly enqueueTask: (task: Task<Data>) => number
 315   /**
 316    * Prepends a task to the tasks queue.
 317    *
 318    * @param task - The task to prepend.
 319    * @returns The tasks queue size.
 320    */
 321   readonly unshiftTask: (task: Task<Data>) => number
 322   /**
 323    * Dequeue task.
 324    *
 325    * @returns The dequeued task.
 326    */
 327   readonly dequeueTask: () => Task<Data> | undefined
 328   /**
 329    * Pops a task from the tasks queue.
 330    *
 331    * @returns The popped task.
 332    */
 333   readonly popTask: () => Task<Data> | undefined
 334   /**
 335    * Clears tasks queue.
 336    */
 337   readonly clearTasksQueue: () => void
 338   /**
 339    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 340    *
 341    * @returns `true` if the worker node has back pressure, `false` otherwise.
 342    */
 343   readonly hasBackPressure: () => boolean
 344   /**
 345    * Resets usage statistics.
 346    */
 347   readonly resetUsage: () => void
 348   /**
 349    * Terminates the worker node.
 350    */
 351   readonly terminate: () => Promise<void>
 352   /**
 353    * Registers a worker event handler.
 354    *
 355    * @param event - The event.
 356    * @param handler - The event handler.
 357    */
 358   readonly registerWorkerEventHandler: (
 359     event: string,
 360     handler:
 361     | OnlineHandler<Worker>
 362     | MessageHandler<Worker>
 363     | ErrorHandler<Worker>
 364     | ExitHandler<Worker>
 365   ) => void
 366   /**
 367    * Registers once a worker event handler.
 368    *
 369    * @param event - The event.
 370    * @param handler - The event handler.
 371    */
 372   readonly registerOnceWorkerEventHandler: (
 373     event: string,
 374     handler:
 375     | OnlineHandler<Worker>
 376     | MessageHandler<Worker>
 377     | ErrorHandler<Worker>
 378     | ExitHandler<Worker>
 379   ) => void
 380   /**
 381    * Gets task function worker usage statistics.
 382    *
 383    * @param name - The task function name.
 384    * @returns The task function worker usage statistics if the task function worker usage statistics are initialized, `undefined` otherwise.
 385    */
 386   readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined
 387   /**
 388    * Deletes task function worker usage statistics.
 389    *
 390    * @param name - The task function name.
 391    * @returns `true` if the task function worker usage statistics were deleted, `false` otherwise.
 392    */
 393   readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean
 394 }
 395
 396 /**
 397  * Worker node event detail.
 398  *
 399  * @internal
 400  */
 401 export interface WorkerNodeEventDetail {
 402   workerId: number
 403   workerNodeKey?: number
 404 }