src/pools/utils.ts

   1 import cluster, { Worker as ClusterWorker } from 'node:cluster'
   2 import { existsSync } from 'node:fs'
   3 import { env } from 'node:process'
   4 import {
   5   SHARE_ENV,
   6   Worker as ThreadWorker,
   7   type WorkerOptions
   8 } from 'node:worker_threads'
   9
  10 import type { MessageValue, Task } from '../utility-types.js'
  11 import { average, isPlainObject, max, median, min } from '../utils.js'
  12 import type { TasksQueueOptions } from './pool.js'
  13 import {
  14   type MeasurementStatisticsRequirements,
  15   WorkerChoiceStrategies,
  16   type WorkerChoiceStrategy
  17 } from './selection-strategies/selection-strategies-types.js'
  18 import type { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js'
  19 import {
  20   type IWorker,
  21   type IWorkerNode,
  22   type MeasurementStatistics,
  23   type WorkerNodeOptions,
  24   type WorkerType,
  25   WorkerTypes,
  26   type WorkerUsage
  27 } from './worker.js'
  28
  29 /**
  30  * Default measurement statistics requirements.
  31  */
  32 export const DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS: MeasurementStatisticsRequirements =
  33   {
  34     aggregate: false,
  35     average: false,
  36     median: false
  37   }
  38
  39 export const getDefaultTasksQueueOptions = (
  40   poolMaxSize: number
  41 ): Required<TasksQueueOptions> => {
  42   return {
  43     size: Math.pow(poolMaxSize, 2),
  44     concurrency: 1,
  45     taskStealing: true,
  46     tasksStealingOnBackPressure: true,
  47     tasksFinishedTimeout: 2000
  48   }
  49 }
  50
  51 export const checkFilePath = (filePath: string | undefined): void => {
  52   if (filePath == null) {
  53     throw new TypeError('The worker file path must be specified')
  54   }
  55   if (typeof filePath !== 'string') {
  56     throw new TypeError('The worker file path must be a string')
  57   }
  58   if (!existsSync(filePath)) {
  59     throw new Error(`Cannot find the worker file '${filePath}'`)
  60   }
  61 }
  62
  63 export const checkDynamicPoolSize = (
  64   min: number,
  65   max: number | undefined
  66 ): void => {
  67   if (max == null) {
  68     throw new TypeError(
  69       'Cannot instantiate a dynamic pool without specifying the maximum pool size'
  70     )
  71   } else if (!Number.isSafeInteger(max)) {
  72     throw new TypeError(
  73       'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
  74     )
  75   } else if (min > max) {
  76     throw new RangeError(
  77       'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
  78     )
  79   } else if (max === 0) {
  80     throw new RangeError(
  81       'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
  82     )
  83   } else if (min === max) {
  84     throw new RangeError(
  85       'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
  86     )
  87   }
  88 }
  89
  90 export const checkValidWorkerChoiceStrategy = (
  91   workerChoiceStrategy: WorkerChoiceStrategy | undefined
  92 ): void => {
  93   if (
  94     workerChoiceStrategy != null &&
  95     !Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)
  96   ) {
  97     throw new Error(`Invalid worker choice strategy '${workerChoiceStrategy}'`)
  98   }
  99 }
 100
 101 export const checkValidTasksQueueOptions = (
 102   tasksQueueOptions: TasksQueueOptions | undefined
 103 ): void => {
 104   if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
 105     throw new TypeError('Invalid tasks queue options: must be a plain object')
 106   }
 107   if (
 108     tasksQueueOptions?.concurrency != null &&
 109     !Number.isSafeInteger(tasksQueueOptions.concurrency)
 110   ) {
 111     throw new TypeError(
 112       'Invalid worker node tasks concurrency: must be an integer'
 113     )
 114   }
 115   if (
 116     tasksQueueOptions?.concurrency != null &&
 117     tasksQueueOptions.concurrency <= 0
 118   ) {
 119     throw new RangeError(
 120       `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
 121     )
 122   }
 123   if (
 124     tasksQueueOptions?.size != null &&
 125     !Number.isSafeInteger(tasksQueueOptions.size)
 126   ) {
 127     throw new TypeError(
 128       'Invalid worker node tasks queue size: must be an integer'
 129     )
 130   }
 131   if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) {
 132     throw new RangeError(
 133       `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero`
 134     )
 135   }
 136 }
 137
 138 export const checkWorkerNodeArguments = (
 139   type: WorkerType | undefined,
 140   filePath: string | undefined,
 141   opts: WorkerNodeOptions | undefined
 142 ): void => {
 143   if (type == null) {
 144     throw new TypeError('Cannot construct a worker node without a worker type')
 145   }
 146   if (!Object.values(WorkerTypes).includes(type)) {
 147     throw new TypeError(
 148       `Cannot construct a worker node with an invalid worker type '${type}'`
 149     )
 150   }
 151   checkFilePath(filePath)
 152   if (opts == null) {
 153     throw new TypeError(
 154       'Cannot construct a worker node without worker node options'
 155     )
 156   }
 157   if (!isPlainObject(opts)) {
 158     throw new TypeError(
 159       'Cannot construct a worker node with invalid options: must be a plain object'
 160     )
 161   }
 162   if (opts.tasksQueueBackPressureSize == null) {
 163     throw new TypeError(
 164       'Cannot construct a worker node without a tasks queue back pressure size option'
 165     )
 166   }
 167   if (!Number.isSafeInteger(opts.tasksQueueBackPressureSize)) {
 168     throw new TypeError(
 169       'Cannot construct a worker node with a tasks queue back pressure size option that is not an integer'
 170     )
 171   }
 172   if (opts.tasksQueueBackPressureSize <= 0) {
 173     throw new RangeError(
 174       'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
 175     )
 176   }
 177 }
 178
 179 /**
 180  * Updates the given measurement statistics.
 181  *
 182  * @param measurementStatistics - The measurement statistics to update.
 183  * @param measurementRequirements - The measurement statistics requirements.
 184  * @param measurementValue - The measurement value.
 185  * @internal
 186  */
 187 const updateMeasurementStatistics = (
 188   measurementStatistics: MeasurementStatistics,
 189   measurementRequirements: MeasurementStatisticsRequirements | undefined,
 190   measurementValue: number | undefined
 191 ): void => {
 192   if (
 193     measurementRequirements != null &&
 194     measurementValue != null &&
 195     measurementRequirements.aggregate
 196   ) {
 197     measurementStatistics.aggregate =
 198       (measurementStatistics.aggregate ?? 0) + measurementValue
 199     measurementStatistics.minimum = min(
 200       measurementValue,
 201       measurementStatistics.minimum ?? Infinity
 202     )
 203     measurementStatistics.maximum = max(
 204       measurementValue,
 205       measurementStatistics.maximum ?? -Infinity
 206     )
 207     if (measurementRequirements.average || measurementRequirements.median) {
 208       measurementStatistics.history.push(measurementValue)
 209       if (measurementRequirements.average) {
 210         measurementStatistics.average = average(measurementStatistics.history)
 211       } else if (measurementStatistics.average != null) {
 212         delete measurementStatistics.average
 213       }
 214       if (measurementRequirements.median) {
 215         measurementStatistics.median = median(measurementStatistics.history)
 216       } else if (measurementStatistics.median != null) {
 217         delete measurementStatistics.median
 218       }
 219     }
 220   }
 221 }
 222 if (env.NODE_ENV === 'test') {
 223   // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
 224   exports.updateMeasurementStatistics = updateMeasurementStatistics
 225 }
 226
 227 export const updateWaitTimeWorkerUsage = <
 228   Worker extends IWorker,
 229   Data = unknown,
 230   Response = unknown
 231 >(
 232     workerChoiceStrategyContext:
 233     | WorkerChoiceStrategiesContext<Worker, Data, Response>
 234     | undefined,
 235     workerUsage: WorkerUsage,
 236     task: Task<Data>
 237   ): void => {
 238   const timestamp = performance.now()
 239   const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
 240   updateMeasurementStatistics(
 241     workerUsage.waitTime,
 242     workerChoiceStrategyContext?.getTaskStatisticsRequirements().waitTime,
 243     taskWaitTime
 244   )
 245 }
 246
 247 export const updateTaskStatisticsWorkerUsage = <Response = unknown>(
 248   workerUsage: WorkerUsage,
 249   message: MessageValue<Response>
 250 ): void => {
 251   const workerTaskStatistics = workerUsage.tasks
 252   if (
 253     // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
 254     workerTaskStatistics.executing != null &&
 255     workerTaskStatistics.executing > 0
 256   ) {
 257     --workerTaskStatistics.executing
 258   }
 259   if (message.workerError == null) {
 260     ++workerTaskStatistics.executed
 261   } else {
 262     ++workerTaskStatistics.failed
 263   }
 264 }
 265
 266 export const updateRunTimeWorkerUsage = <
 267   Worker extends IWorker,
 268   Data = unknown,
 269   Response = unknown
 270 >(
 271     workerChoiceStrategyContext:
 272     | WorkerChoiceStrategiesContext<Worker, Data, Response>
 273     | undefined,
 274     workerUsage: WorkerUsage,
 275     message: MessageValue<Response>
 276   ): void => {
 277   if (message.workerError != null) {
 278     return
 279   }
 280   updateMeasurementStatistics(
 281     workerUsage.runTime,
 282     workerChoiceStrategyContext?.getTaskStatisticsRequirements().runTime,
 283     message.taskPerformance?.runTime ?? 0
 284   )
 285 }
 286
 287 export const updateEluWorkerUsage = <
 288   Worker extends IWorker,
 289   Data = unknown,
 290   Response = unknown
 291 >(
 292     workerChoiceStrategyContext:
 293     | WorkerChoiceStrategiesContext<Worker, Data, Response>
 294     | undefined,
 295     workerUsage: WorkerUsage,
 296     message: MessageValue<Response>
 297   ): void => {
 298   if (message.workerError != null) {
 299     return
 300   }
 301   const eluTaskStatisticsRequirements =
 302     workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu
 303   updateMeasurementStatistics(
 304     workerUsage.elu.active,
 305     eluTaskStatisticsRequirements,
 306     message.taskPerformance?.elu?.active ?? 0
 307   )
 308   updateMeasurementStatistics(
 309     workerUsage.elu.idle,
 310     eluTaskStatisticsRequirements,
 311     message.taskPerformance?.elu?.idle ?? 0
 312   )
 313   if (eluTaskStatisticsRequirements?.aggregate === true) {
 314     if (message.taskPerformance?.elu != null) {
 315       if (workerUsage.elu.utilization != null) {
 316         workerUsage.elu.utilization =
 317           (workerUsage.elu.utilization +
 318             message.taskPerformance.elu.utilization) /
 319           2
 320       } else {
 321         workerUsage.elu.utilization = message.taskPerformance.elu.utilization
 322       }
 323     }
 324   }
 325 }
 326
 327 export const createWorker = <Worker extends IWorker>(
 328   type: WorkerType,
 329   filePath: string,
 330   opts: { env?: Record<string, unknown>, workerOptions?: WorkerOptions }
 331 ): Worker => {
 332   switch (type) {
 333     case WorkerTypes.thread:
 334       return new ThreadWorker(filePath, {
 335         env: SHARE_ENV,
 336         ...opts.workerOptions
 337       }) as unknown as Worker
 338     case WorkerTypes.cluster:
 339       return cluster.fork(opts.env) as unknown as Worker
 340     default:
 341       // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
 342       throw new Error(`Unknown worker type '${type}'`)
 343   }
 344 }
 345
 346 /**
 347  * Returns the worker type of the given worker.
 348  *
 349  * @param worker - The worker to get the type of.
 350  * @returns The worker type of the given worker.
 351  * @internal
 352  */
 353 export const getWorkerType = (worker: IWorker): WorkerType | undefined => {
 354   if (worker instanceof ThreadWorker) {
 355     return WorkerTypes.thread
 356   } else if (worker instanceof ClusterWorker) {
 357     return WorkerTypes.cluster
 358   }
 359 }
 360
 361 /**
 362  * Returns the worker id of the given worker.
 363  *
 364  * @param worker - The worker to get the id of.
 365  * @returns The worker id of the given worker.
 366  * @internal
 367  */
 368 export const getWorkerId = (worker: IWorker): number | undefined => {
 369   if (worker instanceof ThreadWorker) {
 370     return worker.threadId
 371   } else if (worker instanceof ClusterWorker) {
 372     return worker.id
 373   }
 374 }
 375
 376 export const waitWorkerNodeEvents = async <
 377   Worker extends IWorker,
 378   Data = unknown
 379 >(
 380   workerNode: IWorkerNode<Worker, Data>,
 381   workerNodeEvent: string,
 382   numberOfEventsToWait: number,
 383   timeout: number
 384 ): Promise<number> => {
 385   return await new Promise<number>(resolve => {
 386     let events = 0
 387     if (numberOfEventsToWait === 0) {
 388       resolve(events)
 389       return
 390     }
 391     workerNode.on(workerNodeEvent, () => {
 392       ++events
 393       if (events === numberOfEventsToWait) {
 394         resolve(events)
 395       }
 396     })
 397     if (timeout >= 0) {
 398       setTimeout(() => {
 399         resolve(events)
 400       }, timeout)
 401     }
 402   })
 403 }