src/pools/utils.ts

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