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

   1 import type { IPool } from '../pool.js'
   2 import { DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS } from '../utils.js'
   3 import type { IWorker } from '../worker.js'
   4 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy.js'
   5 import type {
   6   IWorkerChoiceStrategy,
   7   TaskStatisticsRequirements,
   8   WorkerChoiceStrategyOptions
   9 } from './selection-strategies-types.js'
  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: {
  33       aggregate: true,
  34       average: true,
  35       median: false
  36     },
  37     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  38   }
  39
  40   /**
  41    * Round id.
  42    */
  43   private roundId = 0
  44   /**
  45    * Round weights.
  46    */
  47   private roundWeights: number[]
  48   /**
  49    * Worker node id.
  50    */
  51   private workerNodeId = 0
  52   /**
  53    * Worker node virtual execution time.
  54    */
  55   private workerNodeVirtualTaskExecutionTime = 0
  56
  57   /** @inheritDoc */
  58   public constructor (
  59     pool: IPool<Worker, Data, Response>,
  60     opts?: WorkerChoiceStrategyOptions
  61   ) {
  62     super(pool, opts)
  63     this.setTaskStatisticsRequirements(this.opts)
  64     this.roundWeights = this.getRoundWeights()
  65   }
  66
  67   /** @inheritDoc */
  68   public reset (): boolean {
  69     this.resetWorkerNodeKeyProperties()
  70     this.roundId = 0
  71     this.workerNodeId = 0
  72     this.workerNodeVirtualTaskExecutionTime = 0
  73     return true
  74   }
  75
  76   /** @inheritDoc */
  77   public update (): boolean {
  78     return true
  79   }
  80
  81   /** @inheritDoc */
  82   public choose (): number | undefined {
  83     for (
  84       let roundIndex = this.roundId;
  85       roundIndex < this.roundWeights.length;
  86       roundIndex++
  87     ) {
  88       this.roundId = roundIndex
  89       for (
  90         let workerNodeKey = this.workerNodeId;
  91         workerNodeKey < this.pool.workerNodes.length;
  92         workerNodeKey++
  93       ) {
  94         this.workerNodeId = workerNodeKey
  95         if (
  96           this.workerNodeId !== this.nextWorkerNodeKey &&
  97           this.workerNodeVirtualTaskExecutionTime !== 0
  98         ) {
  99           this.workerNodeVirtualTaskExecutionTime = 0
 100         }
 101         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 102         const workerWeight = this.opts!.weights![workerNodeKey]
 103         if (
 104           this.isWorkerNodeReady(workerNodeKey) &&
 105           workerWeight >= this.roundWeights[roundIndex] &&
 106           this.workerNodeVirtualTaskExecutionTime < workerWeight
 107         ) {
 108           this.workerNodeVirtualTaskExecutionTime +=
 109             this.getWorkerNodeTaskWaitTime(workerNodeKey) +
 110             this.getWorkerNodeTaskRunTime(workerNodeKey)
 111           this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey)
 112           this.nextWorkerNodeKey = workerNodeKey
 113           return this.nextWorkerNodeKey
 114         }
 115       }
 116     }
 117     this.interleavedWeightedRoundRobinNextWorkerNodeId()
 118   }
 119
 120   private interleavedWeightedRoundRobinNextWorkerNodeId (): void {
 121     if (this.pool.workerNodes.length === 0) {
 122       this.workerNodeId = 0
 123     } else if (
 124       this.roundId === this.roundWeights.length - 1 &&
 125       this.workerNodeId === this.pool.workerNodes.length - 1
 126     ) {
 127       this.roundId = 0
 128       this.workerNodeId = 0
 129     } else if (this.workerNodeId === this.pool.workerNodes.length - 1) {
 130       this.roundId = this.roundId + 1
 131       this.workerNodeId = 0
 132     } else {
 133       this.workerNodeId = this.workerNodeId + 1
 134     }
 135   }
 136
 137   /** @inheritDoc */
 138   public remove (workerNodeKey: number): boolean {
 139     if (this.pool.workerNodes.length === 0) {
 140       this.resetWorkerNodeKeyProperties()
 141       this.workerNodeId = 0
 142       this.workerNodeVirtualTaskExecutionTime = 0
 143       return true
 144     }
 145     if (
 146       this.workerNodeId === workerNodeKey &&
 147       this.workerNodeId > this.pool.workerNodes.length - 1
 148     ) {
 149       this.workerNodeId = this.pool.workerNodes.length - 1
 150     }
 151     if (
 152       this.previousWorkerNodeKey === workerNodeKey &&
 153       this.previousWorkerNodeKey > this.pool.workerNodes.length - 1
 154     ) {
 155       this.previousWorkerNodeKey = this.pool.workerNodes.length - 1
 156     }
 157     return true
 158   }
 159
 160   /** @inheritDoc */
 161   public setOptions (opts: WorkerChoiceStrategyOptions | undefined): void {
 162     super.setOptions(opts)
 163     this.roundWeights = this.getRoundWeights()
 164   }
 165
 166   private getRoundWeights (): number[] {
 167     return [
 168       ...new Set(
 169         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 170         Object.values(this.opts!.weights!)
 171           .slice()
 172           .sort((a, b) => a - b)
 173       )
 174     ]
 175   }
 176 }