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