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     if (
 126       this.roundId === this.roundWeights.length - 1 &&
 127       this.workerNodeId === this.pool.workerNodes.length - 1
 128     ) {
 129       this.roundId = 0
 130       this.workerNodeId = 0
 131     } else if (this.workerNodeId === this.pool.workerNodes.length - 1) {
 132       this.roundId = this.roundId + 1
 133       this.workerNodeId = 0
 134     } else {
 135       this.workerNodeId = this.workerNodeId + 1
 136     }
 137   }
 138
 139   /** @inheritDoc */
 140   public remove (workerNodeKey: number): boolean {
 141     if (this.pool.workerNodes.length === 0) {
 142       this.reset()
 143     }
 144     if (
 145       this.workerNodeId === workerNodeKey &&
 146       this.workerNodeId > this.pool.workerNodes.length - 1
 147     ) {
 148       this.workerNodeId = this.pool.workerNodes.length - 1
 149     }
 150     if (
 151       this.previousWorkerNodeKey === workerNodeKey &&
 152       this.previousWorkerNodeKey > this.pool.workerNodes.length - 1
 153     ) {
 154       this.previousWorkerNodeKey = this.pool.workerNodes.length - 1
 155     }
 156     return true
 157   }
 158
 159   /** @inheritDoc */
 160   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 161     super.setOptions(opts)
 162     this.roundWeights = this.getRoundWeights()
 163   }
 164
 165   private getRoundWeights (): number[] {
 166     if (this.opts.weights == null) {
 167       return [this.defaultWorkerWeight]
 168     }
 169     return [
 170       ...new Set(
 171         Object.values(this.opts.weights)
 172           .slice()
 173           .sort((a, b) => a - b)
 174       )
 175     ]
 176   }
 177 }