src/pools/utils.ts

   1 import { existsSync } from 'node:fs'
   2 import { average, isPlainObject, max, median, min } from '../utils'
   3 import {
   4   type MeasurementStatisticsRequirements,
   5   WorkerChoiceStrategies,
   6   type WorkerChoiceStrategy
   7 } from './selection-strategies/selection-strategies-types'
   8 import type { TasksQueueOptions } from './pool'
   9 import type { IWorker, MeasurementStatistics } from './worker'
  10
  11 export const checkFilePath = (filePath: string): void => {
  12   if (!existsSync(filePath)) {
  13     throw new Error(`Cannot find the worker file '${filePath}'`)
  14   }
  15 }
  16
  17 export const checkDynamicPoolSize = (min: number, max: number): void => {
  18   if (max == null) {
  19     throw new TypeError(
  20       'Cannot instantiate a dynamic pool without specifying the maximum pool size'
  21     )
  22   } else if (!Number.isSafeInteger(max)) {
  23     throw new TypeError(
  24       'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
  25     )
  26   } else if (min > max) {
  27     throw new RangeError(
  28       'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
  29     )
  30   } else if (max === 0) {
  31     throw new RangeError(
  32       'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
  33     )
  34   } else if (min === max) {
  35     throw new RangeError(
  36       'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
  37     )
  38   }
  39 }
  40
  41 export const checkValidWorkerChoiceStrategy = (
  42   workerChoiceStrategy: WorkerChoiceStrategy
  43 ): void => {
  44   if (
  45     workerChoiceStrategy != null &&
  46     !Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)
  47   ) {
  48     throw new Error(`Invalid worker choice strategy '${workerChoiceStrategy}'`)
  49   }
  50 }
  51
  52 export const checkValidTasksQueueOptions = (
  53   tasksQueueOptions: TasksQueueOptions
  54 ): void => {
  55   if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
  56     throw new TypeError('Invalid tasks queue options: must be a plain object')
  57   }
  58   if (
  59     tasksQueueOptions?.concurrency != null &&
  60     !Number.isSafeInteger(tasksQueueOptions.concurrency)
  61   ) {
  62     throw new TypeError(
  63       'Invalid worker node tasks concurrency: must be an integer'
  64     )
  65   }
  66   if (
  67     tasksQueueOptions?.concurrency != null &&
  68     tasksQueueOptions.concurrency <= 0
  69   ) {
  70     throw new RangeError(
  71       `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
  72     )
  73   }
  74   if (
  75     tasksQueueOptions?.size != null &&
  76     !Number.isSafeInteger(tasksQueueOptions.size)
  77   ) {
  78     throw new TypeError(
  79       'Invalid worker node tasks queue size: must be an integer'
  80     )
  81   }
  82   if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) {
  83     throw new RangeError(
  84       `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero`
  85     )
  86   }
  87 }
  88
  89 export const checkWorkerNodeArguments = <Worker extends IWorker>(
  90   worker: Worker,
  91   tasksQueueBackPressureSize: number
  92 ): void => {
  93   if (worker == null) {
  94     throw new TypeError('Cannot construct a worker node without a worker')
  95   }
  96   if (tasksQueueBackPressureSize == null) {
  97     throw new TypeError(
  98       'Cannot construct a worker node without a tasks queue back pressure size'
  99     )
 100   }
 101   if (!Number.isSafeInteger(tasksQueueBackPressureSize)) {
 102     throw new TypeError(
 103       'Cannot construct a worker node with a tasks queue back pressure size that is not an integer'
 104     )
 105   }
 106   if (tasksQueueBackPressureSize <= 0) {
 107     throw new RangeError(
 108       'Cannot construct a worker node with a tasks queue back pressure size that is not a positive integer'
 109     )
 110   }
 111 }
 112
 113 /**
 114  * Updates the given measurement statistics.
 115  *
 116  * @param measurementStatistics - The measurement statistics to update.
 117  * @param measurementRequirements - The measurement statistics requirements.
 118  * @param measurementValue - The measurement value.
 119  * @param numberOfMeasurements - The number of measurements.
 120  * @internal
 121  */
 122 export const updateMeasurementStatistics = (
 123   measurementStatistics: MeasurementStatistics,
 124   measurementRequirements: MeasurementStatisticsRequirements,
 125   measurementValue: number
 126 ): void => {
 127   if (measurementRequirements.aggregate) {
 128     measurementStatistics.aggregate =
 129       (measurementStatistics.aggregate ?? 0) + measurementValue
 130     measurementStatistics.minimum = min(
 131       measurementValue,
 132       measurementStatistics.minimum ?? Infinity
 133     )
 134     measurementStatistics.maximum = max(
 135       measurementValue,
 136       measurementStatistics.maximum ?? -Infinity
 137     )
 138     if (
 139       (measurementRequirements.average || measurementRequirements.median) &&
 140       measurementValue != null
 141     ) {
 142       measurementStatistics.history.push(measurementValue)
 143       if (measurementRequirements.average) {
 144         measurementStatistics.average = average(measurementStatistics.history)
 145       } else if (measurementStatistics.average != null) {
 146         delete measurementStatistics.average
 147       }
 148       if (measurementRequirements.median) {
 149         measurementStatistics.median = median(measurementStatistics.history)
 150       } else if (measurementStatistics.median != null) {
 151         delete measurementStatistics.median
 152       }
 153     }
 154   }
 155 }