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: ((event: 'online', handler: OnlineHandler<this>) => void) &
 212   ((event: 'message', handler: MessageHandler<this>) => void) &
 213   ((event: 'error', handler: ErrorHandler<this>) => void) &
 214   ((event: 'exit', handler: ExitHandler<this>) => void)
 215   /**
 216    * Registers a listener to the exit event that will only be performed once.
 217    *
 218    * @param event - The `'exit'` event.
 219    * @param handler - The exit handler.
 220    */
 221   readonly once: (event: 'exit', handler: ExitHandler<this>) => void
 222 }
 223
 224 /**
 225  * Worker node event detail.
 226  *
 227  * @internal
 228  */
 229 export interface WorkerNodeEventDetail {
 230   workerId: number
 231   workerNodeKey?: number
 232 }
 233
 234 /**
 235  * Worker node interface.
 236  *
 237  * @typeParam Worker - Type of worker.
 238  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 239  * @internal
 240  */
 241 export interface IWorkerNode<Worker extends IWorker, Data = unknown>
 242   extends EventEmitter {
 243   /**
 244    * Worker.
 245    */
 246   readonly worker: Worker
 247   /**
 248    * Worker info.
 249    */
 250   readonly info: WorkerInfo
 251   /**
 252    * Worker usage statistics.
 253    */
 254   readonly usage: WorkerUsage
 255   /**
 256    * Worker choice strategy data.
 257    * This is used to store data that are specific to the worker choice strategy.
 258    */
 259   strategyData?: StrategyData
 260   /**
 261    * Message channel (worker_threads only).
 262    */
 263   readonly messageChannel?: MessageChannel
 264   /**
 265    * Tasks queue back pressure size.
 266    * This is the number of tasks that can be enqueued before the worker node has back pressure.
 267    */
 268   tasksQueueBackPressureSize: number
 269   /**
 270    * Tasks queue size.
 271    *
 272    * @returns The tasks queue size.
 273    */
 274   readonly tasksQueueSize: () => number
 275   /**
 276    * Enqueue task.
 277    *
 278    * @param task - The task to queue.
 279    * @returns The tasks queue size.
 280    */
 281   readonly enqueueTask: (task: Task<Data>) => number
 282   /**
 283    * Prepends a task to the tasks queue.
 284    *
 285    * @param task - The task to prepend.
 286    * @returns The tasks queue size.
 287    */
 288   readonly unshiftTask: (task: Task<Data>) => number
 289   /**
 290    * Dequeue task.
 291    *
 292    * @returns The dequeued task.
 293    */
 294   readonly dequeueTask: () => Task<Data> | undefined
 295   /**
 296    * Pops a task from the tasks queue.
 297    *
 298    * @returns The popped task.
 299    */
 300   readonly popTask: () => Task<Data> | undefined
 301   /**
 302    * Clears tasks queue.
 303    */
 304   readonly clearTasksQueue: () => void
 305   /**
 306    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 307    *
 308    * @returns `true` if the worker node has back pressure, `false` otherwise.
 309    */
 310   readonly hasBackPressure: () => boolean
 311   /**
 312    * Resets usage statistics.
 313    */
 314   readonly resetUsage: () => void
 315   /**
 316    * Closes communication channel.
 317    */
 318   readonly closeChannel: () => void
 319   /**
 320    * Gets task function worker usage statistics.
 321    *
 322    * @param name - The task function name.
 323    * @returns The task function worker usage statistics if the task function worker usage statistics are initialized, `undefined` otherwise.
 324    */
 325   readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined
 326   /**
 327    * Deletes task function worker usage statistics.
 328    *
 329    * @param name - The task function name.
 330    * @returns `true` if the task function worker usage statistics were deleted, `false` otherwise.
 331    */
 332   readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean
 333 }