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