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    * Worker id.
 202    */
 203   readonly id?: number
 204   readonly threadId?: number
 205   /**
 206    * Registers an event listener.
 207    *
 208    * @param event - The event.
 209    * @param handler - The event handler.
 210    */
 211   readonly on: (
 212     event: string,
 213     handler:
 214     | OnlineHandler<this>
 215     | MessageHandler<this>
 216     | ErrorHandler<this>
 217     | ExitHandler<this>
 218   ) => void
 219   /**
 220    * Registers once an event listener.
 221    *
 222    * @param event - The event.
 223    * @param handler - The event handler.
 224    */
 225   readonly once: (
 226     event: string,
 227     handler:
 228     | OnlineHandler<this>
 229     | MessageHandler<this>
 230     | ErrorHandler<this>
 231     | ExitHandler<this>
 232   ) => void
 233 }
 234
 235 /**
 236  * Worker node options.
 237  *
 238  * @internal
 239  */
 240 export interface WorkerNodeOptions {
 241   workerOptions?: WorkerOptions
 242   env?: Record<string, unknown>
 243   tasksQueueBackPressureSize: number
 244 }
 245
 246 /**
 247  * Worker node interface.
 248  *
 249  * @typeParam Worker - Type of worker.
 250  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 251  * @internal
 252  */
 253 export interface IWorkerNode<Worker extends IWorker, Data = unknown>
 254   extends EventEmitter {
 255   /**
 256    * Worker.
 257    */
 258   readonly worker: Worker
 259   /**
 260    * Worker info.
 261    */
 262   readonly info: WorkerInfo
 263   /**
 264    * Worker usage statistics.
 265    */
 266   readonly usage: WorkerUsage
 267   /**
 268    * Worker choice strategy data.
 269    * This is used to store data that are specific to the worker choice strategy.
 270    */
 271   strategyData?: StrategyData
 272   /**
 273    * Message channel (worker_threads only).
 274    */
 275   readonly messageChannel?: MessageChannel
 276   /**
 277    * Tasks queue back pressure size.
 278    * This is the number of tasks that can be enqueued before the worker node has back pressure.
 279    */
 280   tasksQueueBackPressureSize: number
 281   /**
 282    * Tasks queue size.
 283    *
 284    * @returns The tasks queue size.
 285    */
 286   readonly tasksQueueSize: () => number
 287   /**
 288    * Enqueue task.
 289    *
 290    * @param task - The task to queue.
 291    * @returns The tasks queue size.
 292    */
 293   readonly enqueueTask: (task: Task<Data>) => number
 294   /**
 295    * Prepends a task to the tasks queue.
 296    *
 297    * @param task - The task to prepend.
 298    * @returns The tasks queue size.
 299    */
 300   readonly unshiftTask: (task: Task<Data>) => number
 301   /**
 302    * Dequeue task.
 303    *
 304    * @returns The dequeued task.
 305    */
 306   readonly dequeueTask: () => Task<Data> | undefined
 307   /**
 308    * Pops a task from the tasks queue.
 309    *
 310    * @returns The popped task.
 311    */
 312   readonly popTask: () => Task<Data> | undefined
 313   /**
 314    * Clears tasks queue.
 315    */
 316   readonly clearTasksQueue: () => void
 317   /**
 318    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 319    *
 320    * @returns `true` if the worker node has back pressure, `false` otherwise.
 321    */
 322   readonly hasBackPressure: () => boolean
 323   /**
 324    * Resets usage statistics.
 325    */
 326   readonly resetUsage: () => void
 327   /**
 328    * Closes communication channel.
 329    */
 330   readonly closeChannel: () => void
 331   /**
 332    * Registers a worker event handler.
 333    *
 334    * @param event - The event.
 335    * @param listener - The event listener.
 336    */
 337   readonly registerWorkerEventHandler: (
 338     event: string,
 339     listener:
 340     | OnlineHandler<Worker>
 341     | MessageHandler<Worker>
 342     | ErrorHandler<Worker>
 343     | ExitHandler<Worker>
 344   ) => void
 345   /**
 346    * Registers once a worker event handler.
 347    *
 348    * @param event - The event.
 349    * @param listener - The event listener.
 350    */
 351   readonly registerOnceWorkerEventHandler: (
 352     event: string,
 353     listener:
 354     | OnlineHandler<Worker>
 355     | MessageHandler<Worker>
 356     | ErrorHandler<Worker>
 357     | ExitHandler<Worker>
 358   ) => void
 359   /**
 360    * Gets task function worker usage statistics.
 361    *
 362    * @param name - The task function name.
 363    * @returns The task function worker usage statistics if the task function worker usage statistics are initialized, `undefined` otherwise.
 364    */
 365   readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined
 366   /**
 367    * Deletes task function worker usage statistics.
 368    *
 369    * @param name - The task function name.
 370    * @returns `true` if the task function worker usage statistics were deleted, `false` otherwise.
 371    */
 372   readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean
 373 }
 374
 375 /**
 376  * Worker node event detail.
 377  *
 378  * @internal
 379  */
 380 export interface WorkerNodeEventDetail {
 381   workerId: number
 382   workerNodeKey?: number
 383 }