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 serializable data.
  18  * @typeParam Response - Type of execution response. This can only be serializable 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    * Worker node id where the current task will be submitted.
  53    */
  54   private currentWorkerNodeId: number = 0
  55   /**
  56    * Default worker weight.
  57    */
  58   private readonly defaultWorkerWeight: number
  59   /**
  60    * Worker virtual task runtime.
  61    */
  62   private workerVirtualTaskRunTime: number = 0
  63
  64   /** @inheritDoc */
  65   public constructor (
  66     pool: IPool<Worker, Data, Response>,
  67     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  68   ) {
  69     super(pool, opts)
  70     this.setTaskStatisticsRequirements(this.opts)
  71     this.defaultWorkerWeight = this.computeDefaultWorkerWeight()
  72   }
  73
  74   /** @inheritDoc */
  75   public reset (): boolean {
  76     this.currentWorkerNodeId = 0
  77     this.workerVirtualTaskRunTime = 0
  78     return true
  79   }
  80
  81   /** @inheritDoc */
  82   public update (): boolean {
  83     return true
  84   }
  85
  86   /** @inheritDoc */
  87   public choose (): number {
  88     const chosenWorkerNodeKey = this.currentWorkerNodeId
  89     const workerVirtualTaskRunTime = this.workerVirtualTaskRunTime
  90     const workerWeight =
  91       this.opts.weights?.[chosenWorkerNodeKey] ?? this.defaultWorkerWeight
  92     if (workerVirtualTaskRunTime < workerWeight) {
  93       this.workerVirtualTaskRunTime =
  94         workerVirtualTaskRunTime +
  95         this.getWorkerTaskRunTime(chosenWorkerNodeKey)
  96     } else {
  97       this.currentWorkerNodeId =
  98         this.currentWorkerNodeId === this.pool.workerNodes.length - 1
  99           ? 0
 100           : this.currentWorkerNodeId + 1
 101       this.workerVirtualTaskRunTime = 0
 102     }
 103     return chosenWorkerNodeKey
 104   }
 105
 106   /** @inheritDoc */
 107   public remove (workerNodeKey: number): boolean {
 108     if (this.currentWorkerNodeId === workerNodeKey) {
 109       if (this.pool.workerNodes.length === 0) {
 110         this.currentWorkerNodeId = 0
 111       } else if (this.currentWorkerNodeId > this.pool.workerNodes.length - 1) {
 112         this.currentWorkerNodeId = this.pool.workerNodes.length - 1
 113       }
 114       this.workerVirtualTaskRunTime = 0
 115     }
 116     return true
 117   }
 118 }