src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts

   1 import { cpus } from 'node:os'
   2 import type { IWorker } from '../worker'
   3 import type { IPool } from '../pool'
   4 import { DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS } from '../../utils'
   5 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   6 import type {
   7   IWorkerChoiceStrategy,
   8   RequiredStatistics,
   9   WorkerChoiceStrategyOptions
  10 } from './selection-strategies-types'
  11
  12 /**
  13  * Selects the next worker with an interleaved weighted round robin scheduling algorithm.
  14  *
  15  * @typeParam Worker - Type of worker which manages the strategy.
  16  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  17  * @typeParam Response - Type of execution response. This can only be serializable data.
  18  */
  19 export class InterleavedWeightedRoundRobinWorkerChoiceStrategy<
  20     Worker extends IWorker,
  21     Data = unknown,
  22     Response = unknown
  23   >
  24   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  25   implements IWorkerChoiceStrategy {
  26   /** @inheritDoc */
  27   public readonly requiredStatistics: RequiredStatistics = {
  28     runTime: true,
  29     avgRunTime: true,
  30     medRunTime: false
  31   }
  32
  33   /**
  34    * Worker node id where the current task will be submitted.
  35    */
  36   private currentWorkerNodeId: number = 0
  37   /**
  38    * Current round id.
  39    * This is used to determine the current round weight.
  40    */
  41   private currentRoundId: number = 0
  42   /**
  43    * Round weights.
  44    */
  45   private roundWeights: number[]
  46   /**
  47    * Default worker weight.
  48    */
  49   private readonly defaultWorkerWeight: number
  50
  51   /** @inheritDoc */
  52   public constructor (
  53     pool: IPool<Worker, Data, Response>,
  54     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  55   ) {
  56     super(pool, opts)
  57     this.checkOptions(this.opts)
  58     this.defaultWorkerWeight = this.computeDefaultWorkerWeight()
  59     this.roundWeights = this.getRoundWeights()
  60   }
  61
  62   /** @inheritDoc */
  63   public reset (): boolean {
  64     this.currentWorkerNodeId = 0
  65     this.currentRoundId = 0
  66     return true
  67   }
  68
  69   /** @inheritDoc */
  70   public update (): boolean {
  71     return true
  72   }
  73
  74   /** @inheritDoc */
  75   public choose (): number {
  76     let roundId: number | undefined
  77     let workerNodeId: number | undefined
  78     for (
  79       let roundIndex = this.currentRoundId;
  80       roundIndex < this.roundWeights.length;
  81       roundIndex++
  82     ) {
  83       for (
  84         let workerNodeKey = this.currentWorkerNodeId;
  85         workerNodeKey < this.pool.workerNodes.length;
  86         workerNodeKey++
  87       ) {
  88         const workerWeight =
  89           this.opts.weights?.[workerNodeKey] ?? this.defaultWorkerWeight
  90         if (workerWeight >= this.roundWeights[roundIndex]) {
  91           roundId = roundIndex
  92           workerNodeId = workerNodeKey
  93           break
  94         }
  95       }
  96     }
  97     this.currentRoundId = roundId ?? 0
  98     this.currentWorkerNodeId = workerNodeId ?? 0
  99     const chosenWorkerNodeKey = this.currentWorkerNodeId
 100     if (this.currentWorkerNodeId === this.pool.workerNodes.length - 1) {
 101       this.currentWorkerNodeId = 0
 102       this.currentRoundId =
 103         this.currentRoundId === this.roundWeights.length - 1
 104           ? 0
 105           : this.currentRoundId + 1
 106     } else {
 107       this.currentWorkerNodeId = this.currentWorkerNodeId + 1
 108     }
 109     return chosenWorkerNodeKey
 110   }
 111
 112   /** @inheritDoc */
 113   public remove (workerNodeKey: number): boolean {
 114     if (this.currentWorkerNodeId === workerNodeKey) {
 115       if (this.pool.workerNodes.length === 0) {
 116         this.currentWorkerNodeId = 0
 117       } else if (this.currentWorkerNodeId > this.pool.workerNodes.length - 1) {
 118         this.currentWorkerNodeId = this.pool.workerNodes.length - 1
 119         this.currentRoundId =
 120           this.currentRoundId === this.roundWeights.length - 1
 121             ? 0
 122             : this.currentRoundId + 1
 123       }
 124     }
 125     return true
 126   }
 127
 128   /** @inheritDoc */
 129   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 130     super.setOptions(opts)
 131     this.roundWeights = this.getRoundWeights()
 132   }
 133
 134   private computeDefaultWorkerWeight (): number {
 135     let cpusCycleTimeWeight = 0
 136     for (const cpu of cpus()) {
 137       // CPU estimated cycle time
 138       const numberOfDigits = cpu.speed.toString().length - 1
 139       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 140       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 141     }
 142     return Math.round(cpusCycleTimeWeight / cpus().length)
 143   }
 144
 145   private getRoundWeights (): number[] {
 146     if (this.opts.weights == null) {
 147       return [this.defaultWorkerWeight]
 148     }
 149     return [
 150       ...new Set(
 151         Object.values(this.opts.weights)
 152           .slice()
 153           .sort((a, b) => a - b)
 154       )
 155     ]
 156   }
 157 }