src/pools/selection-strategies/selection-strategies-utils.ts

   1 import { cpus } from 'node:os'
   2
   3 import type { IPool } from '../pool.js'
   4 import type { IWorker } from '../worker.js'
   5 import { FairShareWorkerChoiceStrategy } from './fair-share-worker-choice-strategy.js'
   6 import { InterleavedWeightedRoundRobinWorkerChoiceStrategy } from './interleaved-weighted-round-robin-worker-choice-strategy.js'
   7 import { LeastBusyWorkerChoiceStrategy } from './least-busy-worker-choice-strategy.js'
   8 import { LeastEluWorkerChoiceStrategy } from './least-elu-worker-choice-strategy.js'
   9 import { LeastUsedWorkerChoiceStrategy } from './least-used-worker-choice-strategy.js'
  10 import { RoundRobinWorkerChoiceStrategy } from './round-robin-worker-choice-strategy.js'
  11 import {
  12   type IWorkerChoiceStrategy,
  13   WorkerChoiceStrategies,
  14   type WorkerChoiceStrategy,
  15   type WorkerChoiceStrategyOptions
  16 } from './selection-strategies-types.js'
  17 import { WeightedRoundRobinWorkerChoiceStrategy } from './weighted-round-robin-worker-choice-strategy.js'
  18 import type { WorkerChoiceStrategiesContext } from './worker-choice-strategies-context.js'
  19
  20 const clone = <T>(object: T): T => {
  21   return structuredClone<T>(object)
  22 }
  23
  24 const estimatedCpuSpeed = (): number => {
  25   const runs = 150000000
  26   const begin = performance.now()
  27   // eslint-disable-next-line no-empty
  28   for (let i = runs; i > 0; i--) {}
  29   const end = performance.now()
  30   const duration = end - begin
  31   return Math.trunc(runs / duration / 1000) // in MHz
  32 }
  33
  34 const getDefaultWorkerWeight = (): number => {
  35   const currentCpus = cpus()
  36   let estCpuSpeed: number | undefined
  37   // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
  38   if (currentCpus.every(cpu => cpu.speed == null || cpu.speed === 0)) {
  39     estCpuSpeed = estimatedCpuSpeed()
  40   }
  41   let cpusCycleTimeWeight = 0
  42   for (const cpu of currentCpus) {
  43     // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
  44     if (cpu.speed == null || cpu.speed === 0) {
  45       cpu.speed =
  46         // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
  47         currentCpus.find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ??
  48         estCpuSpeed ??
  49         2000
  50     }
  51     // CPU estimated cycle time
  52     const numberOfDigits = cpu.speed.toString().length - 1
  53     const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
  54     cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
  55   }
  56   return Math.round(cpusCycleTimeWeight / currentCpus.length)
  57 }
  58
  59 const getDefaultWeights = (
  60   poolMaxSize: number,
  61   defaultWorkerWeight?: number
  62 ): Record<number, number> => {
  63   defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight()
  64   const weights: Record<number, number> = {}
  65   for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) {
  66     weights[workerNodeKey] = defaultWorkerWeight
  67   }
  68   return weights
  69 }
  70
  71 export const getWorkerChoiceStrategiesRetries = <
  72   Worker extends IWorker,
  73   Data,
  74   Response
  75 >(
  76     pool: IPool<Worker, Data, Response>,
  77     opts?: WorkerChoiceStrategyOptions
  78   ): number => {
  79   return (
  80     pool.info.maxSize +
  81     Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length
  82   )
  83 }
  84
  85 export const buildWorkerChoiceStrategyOptions = <
  86   Worker extends IWorker,
  87   Data,
  88   Response
  89 >(
  90     pool: IPool<Worker, Data, Response>,
  91     opts?: WorkerChoiceStrategyOptions
  92   ): WorkerChoiceStrategyOptions => {
  93   opts = clone(opts ?? {})
  94   opts.weights = opts.weights ?? getDefaultWeights(pool.info.maxSize)
  95   return {
  96     ...{
  97       runTime: { median: false },
  98       waitTime: { median: false },
  99       elu: { median: false }
 100     },
 101     ...opts
 102   }
 103 }
 104
 105 export const getWorkerChoiceStrategy = <Worker extends IWorker, Data, Response>(
 106   workerChoiceStrategy: WorkerChoiceStrategy,
 107   pool: IPool<Worker, Data, Response>,
 108   context: ThisType<WorkerChoiceStrategiesContext<Worker, Data, Response>>,
 109   opts?: WorkerChoiceStrategyOptions
 110 ): IWorkerChoiceStrategy => {
 111   switch (workerChoiceStrategy) {
 112     case WorkerChoiceStrategies.ROUND_ROBIN:
 113       return new (RoundRobinWorkerChoiceStrategy.bind(context))(pool, opts)
 114     case WorkerChoiceStrategies.LEAST_USED:
 115       return new (LeastUsedWorkerChoiceStrategy.bind(context))(pool, opts)
 116     case WorkerChoiceStrategies.LEAST_BUSY:
 117       return new (LeastBusyWorkerChoiceStrategy.bind(context))(pool, opts)
 118     case WorkerChoiceStrategies.LEAST_ELU:
 119       return new (LeastEluWorkerChoiceStrategy.bind(context))(pool, opts)
 120     case WorkerChoiceStrategies.FAIR_SHARE:
 121       return new (FairShareWorkerChoiceStrategy.bind(context))(pool, opts)
 122     case WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN:
 123       return new (WeightedRoundRobinWorkerChoiceStrategy.bind(context))(
 124         pool,
 125         opts
 126       )
 127     case WorkerChoiceStrategies.INTERLEAVED_WEIGHTED_ROUND_ROBIN:
 128       return new (InterleavedWeightedRoundRobinWorkerChoiceStrategy.bind(
 129         context
 130       ))(pool, opts)
 131     default:
 132       throw new Error(
 133         // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
 134         `Worker choice strategy '${workerChoiceStrategy}' is not valid`
 135       )
 136   }
 137 }