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 { DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS } from '../../utils'
   4 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   5 import type {
   6   IWorkerChoiceStrategy,
   7   WorkerChoiceStrategyOptions
   8 } from './selection-strategies-types'
   9
  10 /**
  11  * Selects the next worker with an interleaved weighted round robin scheduling algorithm.
  12  *
  13  * @typeParam Worker - Type of worker which manages the strategy.
  14  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  15  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  16  */
  17 export class InterleavedWeightedRoundRobinWorkerChoiceStrategy<
  18     Worker extends IWorker,
  19     Data = unknown,
  20     Response = unknown
  21   >
  22   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  23   implements IWorkerChoiceStrategy {
  24   /**
  25    * Round id.
  26    * This is used to determine the current round weight.
  27    */
  28   private roundId: number = 0
  29   /**
  30    * Round weights.
  31    */
  32   private roundWeights: number[]
  33   /**
  34    * Default worker weight.
  35    */
  36   private readonly defaultWorkerWeight: number
  37
  38   /** @inheritDoc */
  39   public constructor (
  40     pool: IPool<Worker, Data, Response>,
  41     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  42   ) {
  43     super(pool, opts)
  44     this.setTaskStatisticsRequirements(this.opts)
  45     this.defaultWorkerWeight = this.computeDefaultWorkerWeight()
  46     this.roundWeights = this.getRoundWeights()
  47   }
  48
  49   /** @inheritDoc */
  50   public reset (): boolean {
  51     this.resetWorkerNodeKeyProperties()
  52     this.roundId = 0
  53     return true
  54   }
  55
  56   /** @inheritDoc */
  57   public update (): boolean {
  58     return true
  59   }
  60
  61   /** @inheritDoc */
  62   public choose (): number | undefined {
  63     let roundId!: number
  64     let workerNodeId: number | undefined
  65     for (
  66       let roundIndex = this.roundId;
  67       roundIndex < this.roundWeights.length;
  68       roundIndex++
  69     ) {
  70       roundId = roundIndex
  71       for (
  72         let workerNodeKey =
  73           this.nextWorkerNodeKey ?? this.previousWorkerNodeKey;
  74         workerNodeKey < this.pool.workerNodes.length;
  75         workerNodeKey++
  76       ) {
  77         const workerWeight =
  78           this.opts.weights?.[workerNodeKey] ?? this.defaultWorkerWeight
  79         if (
  80           this.isWorkerNodeEligible(workerNodeKey) &&
  81           workerWeight >= this.roundWeights[roundIndex]
  82         ) {
  83           workerNodeId = workerNodeKey
  84           break
  85         }
  86       }
  87     }
  88     this.roundId = roundId
  89     if (workerNodeId == null) {
  90       this.previousWorkerNodeKey =
  91         this.nextWorkerNodeKey ?? this.previousWorkerNodeKey
  92     }
  93     this.nextWorkerNodeKey = workerNodeId
  94     const chosenWorkerNodeKey = this.nextWorkerNodeKey
  95     if (this.nextWorkerNodeKey === this.pool.workerNodes.length - 1) {
  96       this.nextWorkerNodeKey = 0
  97       this.roundId =
  98         this.roundId === this.roundWeights.length - 1 ? 0 : this.roundId + 1
  99     } else {
 100       this.nextWorkerNodeKey =
 101         (this.nextWorkerNodeKey ?? this.previousWorkerNodeKey) + 1
 102     }
 103     return chosenWorkerNodeKey
 104   }
 105
 106   /** @inheritDoc */
 107   public remove (workerNodeKey: number): boolean {
 108     if (this.nextWorkerNodeKey === workerNodeKey) {
 109       if (this.pool.workerNodes.length === 0) {
 110         this.nextWorkerNodeKey = 0
 111       } else if (this.nextWorkerNodeKey > this.pool.workerNodes.length - 1) {
 112         this.nextWorkerNodeKey = this.pool.workerNodes.length - 1
 113         this.roundId =
 114           this.roundId === this.roundWeights.length - 1 ? 0 : this.roundId + 1
 115       }
 116     }
 117     return true
 118   }
 119
 120   /** @inheritDoc */
 121   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 122     super.setOptions(opts)
 123     this.roundWeights = this.getRoundWeights()
 124   }
 125
 126   private getRoundWeights (): number[] {
 127     if (this.opts.weights == null) {
 128       return [this.defaultWorkerWeight]
 129     }
 130     return [
 131       ...new Set(
 132         Object.values(this.opts.weights)
 133           .slice()
 134           .sort((a, b) => a - b)
 135       )
 136     ]
 137   }
 138 }