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