src/pools/worker.ts

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