src/pools/worker.ts

   1 import type { MessageChannel } from 'node:worker_threads'
   2 import type { CircularArray } from '../circular-array'
   3 import type { Task } from '../utility-types'
   4
   5 /**
   6  * Callback invoked when the worker has started successfully.
   7  */
   8 export type OnlineHandler<Worker extends IWorker> = (this: Worker) => void
   9
  10 /**
  11  * Callback invoked if the worker has received a message.
  12  */
  13 export type MessageHandler<Worker extends IWorker> = (
  14   this: Worker,
  15   message: unknown
  16 ) => void
  17
  18 /**
  19  * Callback invoked if the worker raised an error.
  20  */
  21 export type ErrorHandler<Worker extends IWorker> = (
  22   this: Worker,
  23   error: Error
  24 ) => void
  25
  26 /**
  27  * Callback invoked when the worker exits successfully.
  28  */
  29 export type ExitHandler<Worker extends IWorker> = (
  30   this: Worker,
  31   exitCode: number
  32 ) => void
  33
  34 /**
  35  * Measurement statistics.
  36  *
  37  * @internal
  38  */
  39 export interface MeasurementStatistics {
  40   /**
  41    * Measurement aggregate.
  42    */
  43   aggregate?: number
  44   /**
  45    * Measurement minimum.
  46    */
  47   minimum?: number
  48   /**
  49    * Measurement maximum.
  50    */
  51   maximum?: number
  52   /**
  53    * Measurement average.
  54    */
  55   average?: number
  56   /**
  57    * Measurement median.
  58    */
  59   median?: number
  60   /**
  61    * Measurement history.
  62    */
  63   readonly history: CircularArray<number>
  64 }
  65
  66 /**
  67  * Event loop utilization measurement statistics.
  68  *
  69  * @internal
  70  */
  71 export interface EventLoopUtilizationMeasurementStatistics {
  72   readonly idle: MeasurementStatistics
  73   readonly active: MeasurementStatistics
  74   utilization?: number
  75 }
  76
  77 /**
  78  * Task statistics.
  79  *
  80  * @internal
  81  */
  82 export interface TaskStatistics {
  83   /**
  84    * Number of executed tasks.
  85    */
  86   executed: number
  87   /**
  88    * Number of executing tasks.
  89    */
  90   executing: number
  91   /**
  92    * Number of queued tasks.
  93    */
  94   readonly queued: number
  95   /**
  96    * Maximum number of queued tasks.
  97    */
  98   readonly maxQueued?: number
  99   /**
 100    * Number of stolen tasks.
 101    */
 102   stolen: number
 103   /**
 104    * Number of failed tasks.
 105    */
 106   failed: number
 107 }
 108
 109 /**
 110  * Enumeration of worker types.
 111  */
 112 export const WorkerTypes = Object.freeze({
 113   thread: 'thread',
 114   cluster: 'cluster'
 115 } as const)
 116
 117 /**
 118  * Worker type.
 119  */
 120 export type WorkerType = keyof typeof WorkerTypes
 121
 122 /**
 123  * Worker information.
 124  *
 125  * @internal
 126  */
 127 export interface WorkerInfo {
 128   /**
 129    * Worker id.
 130    */
 131   readonly id: number | undefined
 132   /**
 133    * Worker type.
 134    */
 135   readonly type: WorkerType
 136   /**
 137    * Dynamic flag.
 138    */
 139   dynamic: boolean
 140   /**
 141    * Ready flag.
 142    */
 143   ready: boolean
 144   /**
 145    * Task function names.
 146    */
 147   taskFunctionNames?: string[]
 148 }
 149
 150 /**
 151  * Worker usage statistics.
 152  *
 153  * @internal
 154  */
 155 export interface WorkerUsage {
 156   /**
 157    * Tasks statistics.
 158    */
 159   readonly tasks: TaskStatistics
 160   /**
 161    * Tasks runtime statistics.
 162    */
 163   readonly runTime: MeasurementStatistics
 164   /**
 165    * Tasks wait time statistics.
 166    */
 167   readonly waitTime: MeasurementStatistics
 168   /**
 169    * Tasks event loop utilization statistics.
 170    */
 171   readonly elu: EventLoopUtilizationMeasurementStatistics
 172 }
 173
 174 /**
 175  * Worker choice strategy data.
 176  *
 177  * @internal
 178  */
 179 export interface StrategyData {
 180   virtualTaskEndTimestamp?: number
 181 }
 182
 183 /**
 184  * Worker interface.
 185  */
 186 export interface IWorker {
 187   /**
 188    * Worker id.
 189    */
 190   readonly id?: number
 191   readonly threadId?: number
 192   /**
 193    * Registers an event listener.
 194    *
 195    * @param event - The event.
 196    * @param handler - The event handler.
 197    */
 198   readonly on: ((event: 'online', handler: OnlineHandler<this>) => void) &
 199   ((event: 'message', handler: MessageHandler<this>) => void) &
 200   ((event: 'error', handler: ErrorHandler<this>) => void) &
 201   ((event: 'exit', handler: ExitHandler<this>) => void)
 202   /**
 203    * Registers a listener to the exit event that will only be performed once.
 204    *
 205    * @param event - The `'exit'` event.
 206    * @param handler - The exit handler.
 207    */
 208   readonly once: (event: 'exit', handler: ExitHandler<this>) => void
 209 }
 210
 211 /**
 212  * Worker node event detail.
 213  *
 214  * @internal
 215  */
 216 export interface WorkerNodeEventDetail {
 217   workerId: number
 218 }
 219
 220 /**
 221  * Worker node interface.
 222  *
 223  * @typeParam Worker - Type of worker.
 224  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
 225  * @internal
 226  */
 227 export interface IWorkerNode<Worker extends IWorker, Data = unknown>
 228   extends EventTarget {
 229   /**
 230    * Worker.
 231    */
 232   readonly worker: Worker
 233   /**
 234    * Worker info.
 235    */
 236   readonly info: WorkerInfo
 237   /**
 238    * Worker usage statistics.
 239    */
 240   readonly usage: WorkerUsage
 241   /**
 242    * Worker choice strategy data.
 243    * This is used to store data that is specific to the worker choice strategy.
 244    */
 245   strategyData?: StrategyData
 246   /**
 247    * Message channel (worker_threads only).
 248    */
 249   readonly messageChannel?: MessageChannel
 250   /**
 251    * Tasks queue back pressure size.
 252    * This is the number of tasks that can be enqueued before the worker node has back pressure.
 253    */
 254   tasksQueueBackPressureSize: number
 255   /**
 256    * Tasks queue size.
 257    *
 258    * @returns The tasks queue size.
 259    */
 260   readonly tasksQueueSize: () => number
 261   /**
 262    * Enqueue task.
 263    *
 264    * @param task - The task to queue.
 265    * @returns The tasks queue size.
 266    */
 267   readonly enqueueTask: (task: Task<Data>) => number
 268   /**
 269    * Prepends a task to the tasks queue.
 270    *
 271    * @param task - The task to prepend.
 272    * @returns The tasks queue size.
 273    */
 274   readonly unshiftTask: (task: Task<Data>) => number
 275   /**
 276    * Dequeue task.
 277    *
 278    * @returns The dequeued task.
 279    */
 280   readonly dequeueTask: () => Task<Data> | undefined
 281   /**
 282    * Pops a task from the tasks queue.
 283    *
 284    * @returns The popped task.
 285    */
 286   readonly popTask: () => Task<Data> | undefined
 287   /**
 288    * Clears tasks queue.
 289    */
 290   readonly clearTasksQueue: () => void
 291   /**
 292    * Whether the worker node has back pressure (i.e. its tasks queue is full).
 293    *
 294    * @returns `true` if the worker node has back pressure, `false` otherwise.
 295    */
 296   readonly hasBackPressure: () => boolean
 297   /**
 298    * Resets usage statistics.
 299    */
 300   readonly resetUsage: () => void
 301   /**
 302    * Closes communication channel.
 303    */
 304   readonly closeChannel: () => void
 305   /**
 306    * Gets task function worker usage statistics.
 307    *
 308    * @param name - The task function name.
 309    * @returns The task function worker usage statistics if the task function worker usage statistics are initialized, `undefined` otherwise.
 310    */
 311   readonly getTaskFunctionWorkerUsage: (name: string) => WorkerUsage | undefined
 312   /**
 313    * Deletes task function worker usage statistics.
 314    *
 315    * @param name - The task function name.
 316    * @returns `true` if the task function worker usage statistics were deleted, `false` otherwise.
 317    */
 318   readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean
 319 }