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   StrategyPolicy,
   8   WorkerChoiceStrategyOptions
   9 } from './selection-strategies-types'
  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 strategyPolicy: StrategyPolicy = {
  27     dynamicWorkerUsage: false,
  28     dynamicWorkerReady: true
  29   }
  30
  31   /**
  32    * Round id.
  33    * This is used to determine the current round weight.
  34    */
  35   private roundId: number = 0
  36   /**
  37    * Round weights.
  38    */
  39   private roundWeights: number[]
  40   /**
  41    * Default worker weight.
  42    */
  43   private readonly defaultWorkerWeight: number
  44
  45   /** @inheritDoc */
  46   public constructor (
  47     pool: IPool<Worker, Data, Response>,
  48     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  49   ) {
  50     super(pool, opts)
  51     this.setTaskStatisticsRequirements(this.opts)
  52     this.defaultWorkerWeight = this.computeDefaultWorkerWeight()
  53     this.roundWeights = this.getRoundWeights()
  54   }
  55
  56   /** @inheritDoc */
  57   public reset (): boolean {
  58     this.nextWorkerNodeKey = 0
  59     this.roundId = 0
  60     return true
  61   }
  62
  63   /** @inheritDoc */
  64   public update (): boolean {
  65     return true
  66   }
  67
  68   /** @inheritDoc */
  69   public choose (): number | undefined {
  70     let roundId: number = this.roundId
  71     let workerNodeId: number | undefined
  72     for (
  73       let roundIndex = this.roundId;
  74       roundIndex < this.roundWeights.length;
  75       roundIndex++
  76     ) {
  77       roundId = roundIndex
  78       for (
  79         let workerNodeKey =
  80           this.nextWorkerNodeKey ?? this.previousWorkerNodeKey;
  81         workerNodeKey < this.pool.workerNodes.length;
  82         workerNodeKey++
  83       ) {
  84         const workerWeight =
  85           this.opts.weights?.[workerNodeKey] ?? this.defaultWorkerWeight
  86         if (
  87           this.isWorkerNodeEligible(workerNodeKey) &&
  88           workerWeight >= this.roundWeights[roundIndex]
  89         ) {
  90           workerNodeId = workerNodeKey
  91           break
  92         }
  93       }
  94     }
  95     this.roundId = roundId
  96     if (workerNodeId == null) {
  97       this.previousWorkerNodeKey =
  98         this.nextWorkerNodeKey ?? this.previousWorkerNodeKey
  99     }
 100     this.nextWorkerNodeKey = workerNodeId
 101     const chosenWorkerNodeKey = this.nextWorkerNodeKey
 102     if (this.nextWorkerNodeKey === this.pool.workerNodes.length - 1) {
 103       this.nextWorkerNodeKey = 0
 104       this.roundId =
 105         this.roundId === this.roundWeights.length - 1 ? 0 : this.roundId + 1
 106     } else {
 107       this.nextWorkerNodeKey =
 108         (this.nextWorkerNodeKey ?? this.previousWorkerNodeKey) + 1
 109     }
 110     return chosenWorkerNodeKey
 111   }
 112
 113   /** @inheritDoc */
 114   public remove (workerNodeKey: number): boolean {
 115     if (this.nextWorkerNodeKey === workerNodeKey) {
 116       if (this.pool.workerNodes.length === 0) {
 117         this.nextWorkerNodeKey = 0
 118       } else if (this.nextWorkerNodeKey > this.pool.workerNodes.length - 1) {
 119         this.nextWorkerNodeKey = this.pool.workerNodes.length - 1
 120         this.roundId =
 121           this.roundId === this.roundWeights.length - 1 ? 0 : this.roundId + 1
 122       }
 123     }
 124     return true
 125   }
 126
 127   /** @inheritDoc */
 128   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 129     super.setOptions(opts)
 130     this.roundWeights = this.getRoundWeights()
 131   }
 132
 133   private getRoundWeights (): number[] {
 134     if (this.opts.weights == null) {
 135       return [this.defaultWorkerWeight]
 136     }
 137     return [
 138       ...new Set(
 139         Object.values(this.opts.weights)
 140           .slice()
 141           .sort((a, b) => a - b)
 142       )
 143     ]
 144   }
 145 }