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

   1 import type { IWorker } from '../worker'
   2 import type { IPool } from '../pool'
   3 import { DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS } from '../../utils'
   4 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   5 import type {
   6   IWorkerChoiceStrategy,
   7   InternalWorkerChoiceStrategyOptions,
   8   TaskStatisticsRequirements
   9 } from './selection-strategies-types'
  10
  11 /**
  12  * Selects the next worker with an interleaved weighted round robin scheduling algorithm.
  13  *
  14  * @typeParam Worker - Type of worker which manages the strategy.
  15  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  16  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  17  */
  18 export class InterleavedWeightedRoundRobinWorkerChoiceStrategy<
  19     Worker extends IWorker,
  20     Data = unknown,
  21     Response = unknown
  22   >
  23   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  24   implements IWorkerChoiceStrategy {
  25   /** @inheritDoc */
  26   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  27     runTime: {
  28       aggregate: true,
  29       average: true,
  30       median: false
  31     },
  32     waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  33     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  34   }
  35
  36   /**
  37    * Round id.
  38    */
  39   private roundId: number = 0
  40   /**
  41    * Default worker weight.
  42    */
  43   private readonly defaultWorkerWeight: number
  44   /**
  45    * Round weights.
  46    */
  47   private roundWeights: number[]
  48   /**
  49    * Worker node id.
  50    */
  51   private workerNodeId: number = 0
  52   /**
  53    * Worker node virtual task runtime.
  54    */
  55   private workerNodeVirtualTaskRunTime: number = 0
  56
  57   /** @inheritDoc */
  58   public constructor (
  59     pool: IPool<Worker, Data, Response>,
  60     opts: InternalWorkerChoiceStrategyOptions
  61   ) {
  62     super(pool, opts)
  63     this.setTaskStatisticsRequirements(this.opts)
  64     this.defaultWorkerWeight = this.computeDefaultWorkerWeight()
  65     this.roundWeights = this.getRoundWeights()
  66   }
  67
  68   /** @inheritDoc */
  69   public reset (): boolean {
  70     this.resetWorkerNodeKeyProperties()
  71     this.roundId = 0
  72     this.workerNodeId = 0
  73     this.workerNodeVirtualTaskRunTime = 0
  74     return true
  75   }
  76
  77   /** @inheritDoc */
  78   public update (): boolean {
  79     return true
  80   }
  81
  82   /** @inheritDoc */
  83   public choose (): number | undefined {
  84     for (
  85       let roundIndex = this.roundId;
  86       roundIndex < this.roundWeights.length;
  87       roundIndex++
  88     ) {
  89       this.roundId = roundIndex
  90       for (
  91         let workerNodeKey = this.workerNodeId;
  92         workerNodeKey < this.pool.workerNodes.length;
  93         workerNodeKey++
  94       ) {
  95         this.workerNodeId = workerNodeKey
  96         if (
  97           this.workerNodeId !== this.nextWorkerNodeKey &&
  98           this.workerNodeVirtualTaskRunTime !== 0
  99         ) {
 100           this.workerNodeVirtualTaskRunTime = 0
 101         }
 102         const workerWeight =
 103           this.opts.weights?.[workerNodeKey] ?? this.defaultWorkerWeight
 104         if (
 105           this.isWorkerNodeReady(workerNodeKey) &&
 106           workerWeight >= this.roundWeights[roundIndex] &&
 107           this.workerNodeVirtualTaskRunTime < workerWeight
 108         ) {
 109           this.workerNodeVirtualTaskRunTime =
 110             this.workerNodeVirtualTaskRunTime +
 111             this.getWorkerNodeTaskRunTime(workerNodeKey)
 112           this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey)
 113           this.nextWorkerNodeKey = workerNodeKey
 114           return this.nextWorkerNodeKey
 115         }
 116       }
 117     }
 118     this.interleavedWeightedRoundRobinNextWorkerNodeId()
 119   }
 120
 121   private interleavedWeightedRoundRobinNextWorkerNodeId (): void {
 122     if (
 123       this.roundId === this.roundWeights.length - 1 &&
 124       this.workerNodeId === this.pool.workerNodes.length - 1
 125     ) {
 126       this.roundId = 0
 127       this.workerNodeId = 0
 128     } else if (this.workerNodeId === this.pool.workerNodes.length - 1) {
 129       this.roundId = this.roundId + 1
 130       this.workerNodeId = 0
 131     } else {
 132       this.workerNodeId = this.workerNodeId + 1
 133     }
 134   }
 135
 136   /** @inheritDoc */
 137   public remove (workerNodeKey: number): boolean {
 138     if (this.pool.workerNodes.length === 0) {
 139       this.reset()
 140     }
 141     if (
 142       this.workerNodeId === workerNodeKey &&
 143       this.workerNodeId > this.pool.workerNodes.length - 1
 144     ) {
 145       this.workerNodeId = this.pool.workerNodes.length - 1
 146     }
 147     if (
 148       this.previousWorkerNodeKey === workerNodeKey &&
 149       this.previousWorkerNodeKey > this.pool.workerNodes.length - 1
 150     ) {
 151       this.previousWorkerNodeKey = this.pool.workerNodes.length - 1
 152     }
 153     return true
 154   }
 155
 156   /** @inheritDoc */
 157   public setOptions (opts: InternalWorkerChoiceStrategyOptions): void {
 158     super.setOptions(opts)
 159     this.roundWeights = this.getRoundWeights()
 160   }
 161
 162   private getRoundWeights (): number[] {
 163     if (this.opts.weights == null) {
 164       return [this.defaultWorkerWeight]
 165     }
 166     return [
 167       ...new Set(
 168         Object.values(this.opts.weights)
 169           .slice()
 170           .sort((a, b) => a - b)
 171       )
 172     ]
 173   }
 174 }