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     // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
 261     if (fn != null) {
 262       result = fn.apply<T, A, R>(context, args)
 263       ;(fn as unknown as undefined) = (context as unknown as undefined) =
 264         undefined
 265     }
 266     return result
 267   }
 268 }
 269
 270 export const getWorkerChoiceStrategyRetries = <
 271   Worker extends IWorker,
 272   Data,
 273   Response
 274 >(
 275     pool: IPool<Worker, Data, Response>,
 276     opts?: WorkerChoiceStrategyOptions
 277   ): number => {
 278   return (
 279     pool.info.maxSize +
 280     Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length
 281   )
 282 }
 283
 284 const clone = <T>(object: T): T => {
 285   return structuredClone<T>(object)
 286 }
 287
 288 export const buildWorkerChoiceStrategyOptions = <
 289   Worker extends IWorker,
 290   Data,
 291   Response
 292 >(
 293     pool: IPool<Worker, Data, Response>,
 294     opts?: WorkerChoiceStrategyOptions
 295   ): WorkerChoiceStrategyOptions => {
 296   opts = clone(opts ?? {})
 297   opts.weights = opts.weights ?? getDefaultWeights(pool.info.maxSize)
 298   return {
 299     ...{
 300       runTime: { median: false },
 301       waitTime: { median: false },
 302       elu: { median: false }
 303     },
 304     ...opts
 305   }
 306 }
 307
 308 const getDefaultWeights = (
 309   poolMaxSize: number,
 310   defaultWorkerWeight?: number
 311 ): Record<number, number> => {
 312   defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight()
 313   const weights: Record<number, number> = {}
 314   for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) {
 315     weights[workerNodeKey] = defaultWorkerWeight
 316   }
 317   return weights
 318 }
 319
 320 const getDefaultWorkerWeight = (): number => {
 321   const cpuSpeed = randomInt(500, 2500)
 322   let cpusCycleTimeWeight = 0
 323   for (const cpu of cpus()) {
 324     // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
 325     if (cpu.speed == null || cpu.speed === 0) {
 326       cpu.speed =
 327         // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
 328         cpus().find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ??
 329         cpuSpeed
 330     }
 331     // CPU estimated cycle time
 332     const numberOfDigits = cpu.speed.toString().length - 1
 333     const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 334     cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 335   }
 336   return Math.round(cpusCycleTimeWeight / cpus().length)
 337 }