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