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 virtual task runtime.
  57    */
  58   private workerVirtualTaskRunTime: 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.workerVirtualTaskRunTime = 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.workerVirtualTaskRunTime !== 0
 102         ) {
 103           this.workerVirtualTaskRunTime = 0
 104         }
 105         const workerWeight =
 106           this.opts.weights?.[workerNodeKey] ?? this.defaultWorkerWeight
 107         if (
 108           workerWeight >= this.roundWeights[roundIndex] &&
 109           this.workerVirtualTaskRunTime < workerWeight
 110         ) {
 111           this.workerVirtualTaskRunTime =
 112             this.workerVirtualTaskRunTime +
 113             this.getWorkerTaskRunTime(workerNodeKey)
 114           this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey)
 115           this.nextWorkerNodeKey = workerNodeKey
 116           return this.nextWorkerNodeKey
 117         }
 118       }
 119     }
 120     this.interleavedWeightedRoundRobinNextWorkerNodeId()
 121   }
 122
 123   private interleavedWeightedRoundRobinNextWorkerNodeId (): void {
 124     if (
 125       this.roundId === this.roundWeights.length - 1 &&
 126       this.workerNodeId === this.pool.workerNodes.length - 1
 127     ) {
 128       this.roundId = 0
 129       this.workerNodeId = 0
 130     } else if (this.workerNodeId === this.pool.workerNodes.length - 1) {
 131       this.roundId = this.roundId + 1
 132       this.workerNodeId = 0
 133     } else {
 134       this.workerNodeId = this.workerNodeId + 1
 135     }
 136   }
 137
 138   /** @inheritDoc */
 139   public remove (workerNodeKey: number): boolean {
 140     if (this.pool.workerNodes.length === 0) {
 141       this.reset()
 142     }
 143     if (
 144       this.workerNodeId === workerNodeKey &&
 145       this.workerNodeId > this.pool.workerNodes.length - 1
 146     ) {
 147       this.workerNodeId = this.pool.workerNodes.length - 1
 148     }
 149     if (
 150       this.previousWorkerNodeKey === workerNodeKey &&
 151       this.previousWorkerNodeKey > this.pool.workerNodes.length - 1
 152     ) {
 153       this.previousWorkerNodeKey = this.pool.workerNodes.length - 1
 154     }
 155     return true
 156   }
 157
 158   /** @inheritDoc */
 159   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 160     super.setOptions(opts)
 161     this.roundWeights = this.getRoundWeights()
 162   }
 163
 164   private getRoundWeights (): number[] {
 165     if (this.opts.weights == null) {
 166       return [this.defaultWorkerWeight]
 167     }
 168     return [
 169       ...new Set(
 170         Object.values(this.opts.weights)
 171           .slice()
 172           .sort((a, b) => a - b)
 173       )
 174     ]
 175   }
 176 }