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

   1 import type { IWorker } from '../worker'
   2 import type { IPool } from '../pool'
   3 import {
   4   DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
   5   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
   6 } from '../../utils'
   7 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   8 import type {
   9   IWorkerChoiceStrategy,
  10   StrategyPolicy,
  11   TaskStatisticsRequirements,
  12   WorkerChoiceStrategyOptions
  13 } from './selection-strategies-types'
  14
  15 /**
  16  * Selects the next worker with a weighted round robin scheduling algorithm.
  17  * Loosely modeled after the weighted round robin queueing algorithm: https://en.wikipedia.org/wiki/Weighted_round_robin.
  18  *
  19  * @typeParam Worker - Type of worker which manages the strategy.
  20  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  21  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  22  */
  23 export class WeightedRoundRobinWorkerChoiceStrategy<
  24     Worker extends IWorker,
  25     Data = unknown,
  26     Response = unknown
  27   >
  28   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  29   implements IWorkerChoiceStrategy {
  30   /** @inheritDoc */
  31   public readonly strategyPolicy: StrategyPolicy = {
  32     useDynamicWorker: true
  33   }
  34
  35   /** @inheritDoc */
  36   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  37     runTime: {
  38       aggregate: true,
  39       average: true,
  40       median: false
  41     },
  42     waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  43     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  44   }
  45
  46   /**
  47    * Default worker weight.
  48    */
  49   private readonly defaultWorkerWeight: number
  50   /**
  51    * Worker virtual task runtime.
  52    */
  53   private workerVirtualTaskRunTime: number = 0
  54
  55   /** @inheritDoc */
  56   public constructor (
  57     pool: IPool<Worker, Data, Response>,
  58     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  59   ) {
  60     super(pool, opts)
  61     this.setTaskStatisticsRequirements(this.opts)
  62     this.defaultWorkerWeight = this.computeDefaultWorkerWeight()
  63   }
  64
  65   /** @inheritDoc */
  66   public reset (): boolean {
  67     this.nextWorkerNodeKey = 0
  68     this.workerVirtualTaskRunTime = 0
  69     return true
  70   }
  71
  72   /** @inheritDoc */
  73   public update (): boolean {
  74     return true
  75   }
  76
  77   /** @inheritDoc */
  78   public choose (): number {
  79     const chosenWorkerNodeKey = this.nextWorkerNodeKey
  80     do {
  81       this.weightedRoundRobinNextWorkerNodeKey()
  82     } while (!this.isWorkerNodeEligible(this.nextWorkerNodeKey))
  83     return chosenWorkerNodeKey
  84   }
  85
  86   /** @inheritDoc */
  87   public remove (workerNodeKey: number): boolean {
  88     if (this.nextWorkerNodeKey === workerNodeKey) {
  89       if (this.pool.workerNodes.length === 0) {
  90         this.nextWorkerNodeKey = 0
  91       } else if (this.nextWorkerNodeKey > this.pool.workerNodes.length - 1) {
  92         this.nextWorkerNodeKey = this.pool.workerNodes.length - 1
  93       }
  94       this.workerVirtualTaskRunTime = 0
  95     }
  96     return true
  97   }
  98
  99   private weightedRoundRobinNextWorkerNodeKey (): number {
 100     const workerVirtualTaskRunTime = this.workerVirtualTaskRunTime
 101     const workerWeight =
 102       this.opts.weights?.[this.nextWorkerNodeKey] ?? this.defaultWorkerWeight
 103     if (workerVirtualTaskRunTime < workerWeight) {
 104       this.workerVirtualTaskRunTime =
 105         workerVirtualTaskRunTime +
 106         this.getWorkerTaskRunTime(this.nextWorkerNodeKey)
 107     } else {
 108       this.nextWorkerNodeKey =
 109         this.nextWorkerNodeKey === this.pool.workerNodes.length - 1
 110           ? 0
 111           : this.nextWorkerNodeKey + 1
 112       this.workerVirtualTaskRunTime = 0
 113     }
 114     return this.nextWorkerNodeKey
 115   }
 116 }