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     this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey)
  74     return this.weightedRoundRobinNextWorkerNodeKey()
  75   }
  76
  77   /** @inheritDoc */
  78   public remove (workerNodeKey: number): boolean {
  79     if (this.pool.workerNodes.length === 0) {
  80       this.reset()
  81     }
  82     if (this.nextWorkerNodeKey === workerNodeKey) {
  83       this.workerVirtualTaskRunTime = 0
  84       if (this.nextWorkerNodeKey > this.pool.workerNodes.length - 1) {
  85         this.nextWorkerNodeKey = this.pool.workerNodes.length - 1
  86       }
  87     }
  88     if (
  89       this.previousWorkerNodeKey === workerNodeKey &&
  90       this.previousWorkerNodeKey > this.pool.workerNodes.length - 1
  91     ) {
  92       this.previousWorkerNodeKey = this.pool.workerNodes.length - 1
  93     }
  94     return true
  95   }
  96
  97   private weightedRoundRobinNextWorkerNodeKey (): number | undefined {
  98     const workerWeight =
  99       this.opts.weights?.[
 100         this.nextWorkerNodeKey ?? this.previousWorkerNodeKey
 101       ] ?? this.defaultWorkerWeight
 102     if (this.workerVirtualTaskRunTime < workerWeight) {
 103       this.workerVirtualTaskRunTime =
 104         this.workerVirtualTaskRunTime +
 105         this.getWorkerTaskRunTime(
 106           this.nextWorkerNodeKey ?? this.previousWorkerNodeKey
 107         )
 108     } else {
 109       this.nextWorkerNodeKey =
 110         this.nextWorkerNodeKey === this.pool.workerNodes.length - 1
 111           ? 0
 112           : (this.nextWorkerNodeKey ?? this.previousWorkerNodeKey) + 1
 113       this.workerVirtualTaskRunTime = 0
 114     }
 115     return this.nextWorkerNodeKey
 116   }
 117 }