src/utils.ts

   1 import * as os from 'node:os'
   2 import { getRandomValues, randomInt } from 'node:crypto'
   3 import { Worker as ClusterWorker } from 'node:cluster'
   4 import { Worker as ThreadWorker } from 'node:worker_threads'
   5 import { cpus } from 'node:os'
   6 import type {
   7   MeasurementStatisticsRequirements,
   8   WorkerChoiceStrategyOptions
   9 } from './pools/selection-strategies/selection-strategies-types.js'
  10 import type { KillBehavior } from './worker/worker-options.js'
  11 import { type IWorker, type WorkerType, WorkerTypes } from './pools/worker.js'
  12 import type { IPool } from './pools/pool.js'
  13
  14 /**
  15  * Default task name.
  16  */
  17 export const DEFAULT_TASK_NAME = 'default'
  18
  19 /**
  20  * An intentional empty function.
  21  */
  22 export const EMPTY_FUNCTION: () => void = Object.freeze(() => {
  23   /* Intentionally empty */
  24 })
  25
  26 /**
  27  * Default measurement statistics requirements.
  28  */
  29 export const DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS: MeasurementStatisticsRequirements =
  30   {
  31     aggregate: false,
  32     average: false,
  33     median: false
  34   }
  35
  36 /**
  37  * Returns safe host OS optimized estimate of the default amount of parallelism a pool should use.
  38  * Always returns a value greater than zero.
  39  *
  40  * @returns The host OS optimized maximum pool size.
  41  */
  42 export const availableParallelism = (): number => {
  43   let availableParallelism = 1
  44   try {
  45     availableParallelism = os.availableParallelism()
  46   } catch {
  47     const cpus = os.cpus()
  48     if (Array.isArray(cpus) && cpus.length > 0) {
  49       availableParallelism = cpus.length
  50     }
  51   }
  52   return availableParallelism
  53 }
  54
  55 /**
  56  * Returns the worker type of the given worker.
  57  *
  58  * @param worker - The worker to get the type of.
  59  * @returns The worker type of the given worker.
  60  * @internal
  61  */
  62 export const getWorkerType = (worker: IWorker): WorkerType | undefined => {
  63   if (worker instanceof ThreadWorker) {
  64     return WorkerTypes.thread
  65   } else if (worker instanceof ClusterWorker) {
  66     return WorkerTypes.cluster
  67   }
  68 }
  69
  70 /**
  71  * Returns the worker id of the given worker.
  72  *
  73  * @param worker - The worker to get the id of.
  74  * @returns The worker id of the given worker.
  75  * @internal
  76  */
  77 export const getWorkerId = (worker: IWorker): number | undefined => {
  78   if (worker instanceof ThreadWorker) {
  79     return worker.threadId
  80   } else if (worker instanceof ClusterWorker) {
  81     return worker.id
  82   }
  83 }
  84
  85 /**
  86  * Sleeps for the given amount of milliseconds.
  87  *
  88  * @param ms - The amount of milliseconds to sleep.
  89  * @returns A promise that resolves after the given amount of milliseconds.
  90  * @internal
  91  */
  92 export const sleep = async (ms: number): Promise<void> => {
  93   await new Promise(resolve => {
  94     setTimeout(resolve, ms)
  95   })
  96 }
  97
  98 /**
  99  * Computes the retry delay in milliseconds using an exponential back off algorithm.
 100  *
 101  * @param retryNumber - The number of retries that have already been attempted
 102  * @param delayFactor - The base delay factor in milliseconds
 103  * @returns Delay in milliseconds
 104  * @internal
 105  */
 106 export const exponentialDelay = (
 107   retryNumber = 0,
 108   delayFactor = 100
 109 ): number => {
 110   const delay = Math.pow(2, retryNumber) * delayFactor
 111   const randomSum = delay * 0.2 * secureRandom() // 0-20% of the delay
 112   return delay + randomSum
 113 }
 114
 115 /**
 116  * Computes the average of the given data set.
 117  *
 118  * @param dataSet - Data set.
 119  * @returns The average of the given data set.
 120  * @internal
 121  */
 122 export const average = (dataSet: number[]): number => {
 123   if (Array.isArray(dataSet) && dataSet.length === 0) {
 124     return 0
 125   }
 126   if (Array.isArray(dataSet) && dataSet.length === 1) {
 127     return dataSet[0]
 128   }
 129   return (
 130     dataSet.reduce((accumulator, number) => accumulator + number, 0) /
 131     dataSet.length
 132   )
 133 }
 134
 135 /**
 136  * Computes the median of the given data set.
 137  *
 138  * @param dataSet - Data set.
 139  * @returns The median of the given data set.
 140  * @internal
 141  */
 142 export const median = (dataSet: number[]): number => {
 143   if (Array.isArray(dataSet) && dataSet.length === 0) {
 144     return 0
 145   }
 146   if (Array.isArray(dataSet) && dataSet.length === 1) {
 147     return dataSet[0]
 148   }
 149   const sortedDataSet = dataSet.slice().sort((a, b) => a - b)
 150   return (
 151     (sortedDataSet[(sortedDataSet.length - 1) >> 1] +
 152       sortedDataSet[sortedDataSet.length >> 1]) /
 153     2
 154   )
 155 }
 156
 157 /**
 158  * Rounds the given number to the given scale.
 159  * The rounding is done using the "round half away from zero" method.
 160  *
 161  * @param num - The number to round.
 162  * @param scale - The scale to round to.
 163  * @returns The rounded number.
 164  * @internal
 165  */
 166 export const round = (num: number, scale = 2): number => {
 167   const rounder = Math.pow(10, scale)
 168   return Math.round(num * rounder * (1 + Number.EPSILON)) / rounder
 169 }
 170
 171 /**
 172  * Is the given object a plain object?
 173  *
 174  * @param obj - The object to check.
 175  * @returns `true` if the given object is a plain object, `false` otherwise.
 176  * @internal
 177  */
 178 export const isPlainObject = (obj: unknown): boolean =>
 179   typeof obj === 'object' &&
 180   obj !== null &&
 181   obj?.constructor === Object &&
 182   Object.prototype.toString.call(obj) === '[object Object]'
 183
 184 /**
 185  * Detects whether the given value is a kill behavior or not.
 186  *
 187  * @typeParam KB - Which specific KillBehavior type to test against.
 188  * @param killBehavior - Which kind of kill behavior to detect.
 189  * @param value - Any value.
 190  * @returns `true` if `value` was strictly equals to `killBehavior`, otherwise `false`.
 191  * @internal
 192  */
 193 export const isKillBehavior = <KB extends KillBehavior>(
 194   killBehavior: KB,
 195   value: unknown
 196 ): value is KB => {
 197   return value === killBehavior
 198 }
 199
 200 /**
 201  * Detects whether the given value is an asynchronous function or not.
 202  *
 203  * @param fn - Any value.
 204  * @returns `true` if `fn` was an asynchronous function, otherwise `false`.
 205  * @internal
 206  */
 207 export const isAsyncFunction = (
 208   fn: unknown
 209 ): fn is (...args: unknown[]) => Promise<unknown> => {
 210   return typeof fn === 'function' && fn.constructor.name === 'AsyncFunction'
 211 }
 212
 213 /**
 214  * Generates a cryptographically secure random number in the [0,1[ range
 215  *
 216  * @returns A number in the [0,1[ range
 217  * @internal
 218  */
 219 export const secureRandom = (): number => {
 220   return getRandomValues(new Uint32Array(1))[0] / 0x100000000
 221 }
 222
 223 /**
 224  * Returns the minimum of the given numbers.
 225  * If no numbers are given, `Infinity` is returned.
 226  *
 227  * @param args - The numbers to get the minimum of.
 228  * @returns The minimum of the given numbers.
 229  * @internal
 230  */
 231 export const min = (...args: number[]): number =>
 232   args.reduce((minimum, num) => (minimum < num ? minimum : num), Infinity)
 233
 234 /**
 235  * Returns the maximum of the given numbers.
 236  * If no numbers are given, `-Infinity` is returned.
 237  *
 238  * @param args - The numbers to get the maximum of.
 239  * @returns The maximum of the given numbers.
 240  * @internal
 241  */
 242 export const max = (...args: number[]): number =>
 243   args.reduce((maximum, num) => (maximum > num ? maximum : num), -Infinity)
 244
 245 /**
 246  * Wraps a function so that it can only be called once.
 247  *
 248  * @param fn - The function to wrap.
 249  * @param context - The context to bind the function to.
 250  * @returns The wrapped function.
 251  * @internal
 252  */
 253 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 254 export const once = <T, A extends any[], R>(
 255   fn: (...args: A) => R,
 256   context: T
 257 ): ((...args: A) => R) => {
 258   let result: R
 259   return (...args: A) => {
 260     if (fn != null) {
 261       result = fn.apply<T, A, R>(context, args)
 262       ;(fn as unknown as undefined) = (context as unknown as undefined) =
 263         undefined
 264     }
 265     return result
 266   }
 267 }
 268
 269 export const getWorkerChoiceStrategyRetries = <
 270   Worker extends IWorker,
 271   Data,
 272   Response
 273 >(
 274     pool: IPool<Worker, Data, Response>,
 275     opts?: WorkerChoiceStrategyOptions
 276   ): number => {
 277   return (
 278     pool.info.maxSize +
 279     Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length
 280   )
 281 }
 282
 283 const clone = <T>(object: T): T => {
 284   return structuredClone<T>(object)
 285 }
 286
 287 export const buildWorkerChoiceStrategyOptions = <
 288   Worker extends IWorker,
 289   Data,
 290   Response
 291 >(
 292     pool: IPool<Worker, Data, Response>,
 293     opts?: WorkerChoiceStrategyOptions
 294   ): WorkerChoiceStrategyOptions => {
 295   opts = clone(opts ?? {})
 296   opts.weights = opts?.weights ?? getDefaultWeights(pool.info.maxSize)
 297   return {
 298     ...{
 299       runTime: { median: false },
 300       waitTime: { median: false },
 301       elu: { median: false }
 302     },
 303     ...opts
 304   }
 305 }
 306
 307 const getDefaultWeights = (
 308   poolMaxSize: number,
 309   defaultWorkerWeight?: number
 310 ): Record<number, number> => {
 311   defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight()
 312   const weights: Record<number, number> = {}
 313   for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) {
 314     weights[workerNodeKey] = defaultWorkerWeight
 315   }
 316   return weights
 317 }
 318
 319 const getDefaultWorkerWeight = (): number => {
 320   const cpuSpeed = randomInt(500, 2500)
 321   let cpusCycleTimeWeight = 0
 322   for (const cpu of cpus()) {
 323     if (cpu.speed == null || cpu.speed === 0) {
 324       cpu.speed =
 325         cpus().find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ??
 326         cpuSpeed
 327     }
 328     // CPU estimated cycle time
 329     const numberOfDigits = cpu.speed.toString().length - 1
 330     const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 331     cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 332   }
 333   return Math.round(cpusCycleTimeWeight / cpus().length)
 334 }