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.nextWorkerNodeId = 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.nextWorkerNodeId
  80     const workerVirtualTaskRunTime = this.workerVirtualTaskRunTime
  81     const workerWeight =
  82       this.opts.weights?.[chosenWorkerNodeKey] ?? this.defaultWorkerWeight
  83     if (workerVirtualTaskRunTime < workerWeight) {
  84       this.workerVirtualTaskRunTime =
  85         workerVirtualTaskRunTime +
  86         this.getWorkerTaskRunTime(chosenWorkerNodeKey)
  87     } else {
  88       this.nextWorkerNodeId =
  89         this.nextWorkerNodeId === this.pool.workerNodes.length - 1
  90           ? 0
  91           : this.nextWorkerNodeId + 1
  92       this.workerVirtualTaskRunTime = 0
  93     }
  94     return chosenWorkerNodeKey
  95   }
  96
  97   /** @inheritDoc */
  98   public remove (workerNodeKey: number): boolean {
  99     if (this.nextWorkerNodeId === workerNodeKey) {
 100       if (this.pool.workerNodes.length === 0) {
 101         this.nextWorkerNodeId = 0
 102       } else if (this.nextWorkerNodeId > this.pool.workerNodes.length - 1) {
 103         this.nextWorkerNodeId = this.pool.workerNodes.length - 1
 104       }
 105       this.workerVirtualTaskRunTime = 0
 106     }
 107     return true
 108   }
 109 }