src/pools/worker.ts

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