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