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   type MeasurementStatisticsRequirements,
  14   type StrategyPolicy,
  15   type TaskStatisticsRequirements,
  16   WorkerChoiceStrategies,
  17   type WorkerChoiceStrategy,
  18   type WorkerChoiceStrategyOptions
  19 } from './selection-strategies-types.js'
  20 import { WeightedRoundRobinWorkerChoiceStrategy } from './weighted-round-robin-worker-choice-strategy.js'
  21 import type { WorkerChoiceStrategiesContext } from './worker-choice-strategies-context.js'
  22
  23 const estimatedCpuSpeed = (): number => {
  24   const runs = 150000000
  25   const begin = performance.now()
  26   // eslint-disable-next-line no-empty
  27   for (let i = runs; i > 0; i--) {}
  28   const end = performance.now()
  29   const duration = end - begin
  30   return Math.trunc(runs / duration / 1000) // in MHz
  31 }
  32
  33 const getDefaultWorkerWeight = (): number => {
  34   const currentCpus = cpus()
  35   let estCpuSpeed: number | undefined
  36   // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
  37   if (currentCpus.every(cpu => cpu.speed == null || cpu.speed === 0)) {
  38     estCpuSpeed = estimatedCpuSpeed()
  39   }
  40   let cpusCycleTimeWeight = 0
  41   for (const cpu of currentCpus) {
  42     // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
  43     if (cpu.speed == null || cpu.speed === 0) {
  44       cpu.speed =
  45         // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
  46         currentCpus.find(cpu => cpu.speed != null && cpu.speed !== 0)?.speed ??
  47         estCpuSpeed ??
  48         2000
  49     }
  50     // CPU estimated cycle time
  51     const numberOfDigits = cpu.speed.toString().length - 1
  52     const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
  53     cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
  54   }
  55   return Math.round(cpusCycleTimeWeight / currentCpus.length)
  56 }
  57
  58 const getDefaultWeights = (
  59   poolMaxSize: number,
  60   defaultWorkerWeight?: number
  61 ): Record<number, number> => {
  62   defaultWorkerWeight = defaultWorkerWeight ?? getDefaultWorkerWeight()
  63   const weights: Record<number, number> = {}
  64   for (let workerNodeKey = 0; workerNodeKey < poolMaxSize; workerNodeKey++) {
  65     weights[workerNodeKey] = defaultWorkerWeight
  66   }
  67   return weights
  68 }
  69
  70 export const getWorkerChoiceStrategiesRetries = <
  71   Worker extends IWorker,
  72   Data,
  73   Response
  74 >(
  75     pool: IPool<Worker, Data, Response>,
  76     opts?: WorkerChoiceStrategyOptions
  77   ): number => {
  78   return (
  79     pool.info.maxSize +
  80     Object.keys(opts?.weights ?? getDefaultWeights(pool.info.maxSize)).length
  81   )
  82 }
  83
  84 export const buildWorkerChoiceStrategyOptions = <
  85   Worker extends IWorker,
  86   Data,
  87   Response
  88 >(
  89     pool: IPool<Worker, Data, Response>,
  90     opts?: WorkerChoiceStrategyOptions
  91   ): WorkerChoiceStrategyOptions => {
  92   opts = structuredClone(opts ?? {})
  93   opts.weights = opts.weights ?? getDefaultWeights(pool.info.maxSize)
  94   return {
  95     ...{
  96       runTime: { median: false },
  97       waitTime: { median: false },
  98       elu: { median: false }
  99     },
 100     ...opts
 101   }
 102 }
 103
 104 export const toggleMedianMeasurementStatisticsRequirements = (
 105   measurementStatisticsRequirements: MeasurementStatisticsRequirements,
 106   toggleMedian: boolean
 107 ): void => {
 108   if (measurementStatisticsRequirements.average && toggleMedian) {
 109     measurementStatisticsRequirements.average = false
 110     measurementStatisticsRequirements.median = toggleMedian
 111   }
 112   if (measurementStatisticsRequirements.median && !toggleMedian) {
 113     measurementStatisticsRequirements.average = true
 114     measurementStatisticsRequirements.median = toggleMedian
 115   }
 116 }
 117
 118 export const buildWorkerChoiceStrategiesPolicy = (
 119   workerChoiceStrategies: Map<WorkerChoiceStrategy, IWorkerChoiceStrategy>
 120 ): StrategyPolicy => {
 121   const policies: StrategyPolicy[] = []
 122   for (const workerChoiceStrategy of workerChoiceStrategies.values()) {
 123     policies.push(workerChoiceStrategy.strategyPolicy)
 124   }
 125   return {
 126     dynamicWorkerUsage: policies.some(p => p.dynamicWorkerUsage),
 127     dynamicWorkerReady: policies.some(p => p.dynamicWorkerReady)
 128   }
 129 }
 130
 131 export const buildWorkerChoiceStrategiesTaskStatisticsRequirements = (
 132   workerChoiceStrategies: Map<WorkerChoiceStrategy, IWorkerChoiceStrategy>
 133 ): TaskStatisticsRequirements => {
 134   const taskStatisticsRequirements: TaskStatisticsRequirements[] = []
 135   for (const workerChoiceStrategy of workerChoiceStrategies.values()) {
 136     taskStatisticsRequirements.push(
 137       workerChoiceStrategy.taskStatisticsRequirements
 138     )
 139   }
 140   return {
 141     runTime: {
 142       aggregate: taskStatisticsRequirements.some(r => r.runTime.aggregate),
 143       average: taskStatisticsRequirements.some(r => r.runTime.average),
 144       median: taskStatisticsRequirements.some(r => r.runTime.median)
 145     },
 146     waitTime: {
 147       aggregate: taskStatisticsRequirements.some(r => r.waitTime.aggregate),
 148       average: taskStatisticsRequirements.some(r => r.waitTime.average),
 149       median: taskStatisticsRequirements.some(r => r.waitTime.median)
 150     },
 151     elu: {
 152       aggregate: taskStatisticsRequirements.some(r => r.elu.aggregate),
 153       average: taskStatisticsRequirements.some(r => r.elu.average),
 154       median: taskStatisticsRequirements.some(r => r.elu.median)
 155     }
 156   }
 157 }
 158
 159 export const getWorkerChoiceStrategy = <Worker extends IWorker, Data, Response>(
 160   workerChoiceStrategy: WorkerChoiceStrategy,
 161   pool: IPool<Worker, Data, Response>,
 162   context: ThisType<WorkerChoiceStrategiesContext<Worker, Data, Response>>,
 163   opts?: WorkerChoiceStrategyOptions
 164 ): IWorkerChoiceStrategy => {
 165   switch (workerChoiceStrategy) {
 166     case WorkerChoiceStrategies.ROUND_ROBIN:
 167       return new (RoundRobinWorkerChoiceStrategy.bind(context))(pool, opts)
 168     case WorkerChoiceStrategies.LEAST_USED:
 169       return new (LeastUsedWorkerChoiceStrategy.bind(context))(pool, opts)
 170     case WorkerChoiceStrategies.LEAST_BUSY:
 171       return new (LeastBusyWorkerChoiceStrategy.bind(context))(pool, opts)
 172     case WorkerChoiceStrategies.LEAST_ELU:
 173       return new (LeastEluWorkerChoiceStrategy.bind(context))(pool, opts)
 174     case WorkerChoiceStrategies.FAIR_SHARE:
 175       return new (FairShareWorkerChoiceStrategy.bind(context))(pool, opts)
 176     case WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN:
 177       return new (WeightedRoundRobinWorkerChoiceStrategy.bind(context))(
 178         pool,
 179         opts
 180       )
 181     case WorkerChoiceStrategies.INTERLEAVED_WEIGHTED_ROUND_ROBIN:
 182       return new (InterleavedWeightedRoundRobinWorkerChoiceStrategy.bind(
 183         context
 184       ))(pool, opts)
 185     default:
 186       throw new Error(
 187         // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
 188         `Worker choice strategy '${workerChoiceStrategy}' is not valid`
 189       )
 190   }
 191 }