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