src/pools/selection-strategies/fair-share-worker-choice-strategy.ts

   1 import {
   2   DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
   3   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
   4 } from '../../utils'
   5 import type { IPool } from '../pool'
   6 import type { IWorker } from '../worker'
   7 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   8 import {
   9   type IWorkerChoiceStrategy,
  10   Measurements,
  11   type TaskStatisticsRequirements,
  12   type WorkerChoiceStrategyOptions
  13 } from './selection-strategies-types'
  14
  15 /**
  16  * Selects the next worker with a fair share scheduling algorithm.
  17  * Loosely modeled after the fair queueing algorithm: https://en.wikipedia.org/wiki/Fair_queuing.
  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 FairShareWorkerChoiceStrategy<
  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 taskStatisticsRequirements: TaskStatisticsRequirements = {
  32     runTime: {
  33       aggregate: true,
  34       average: true,
  35       median: false
  36     },
  37     waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  38     elu: {
  39       aggregate: true,
  40       average: true,
  41       median: false
  42     }
  43   }
  44
  45   /**
  46    * Workers' virtual task end execution timestamp.
  47    */
  48   private workersVirtualTaskEndTimestamp: number[] = []
  49
  50   /** @inheritDoc */
  51   public constructor (
  52     pool: IPool<Worker, Data, Response>,
  53     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  54   ) {
  55     super(pool, opts)
  56     this.setTaskStatisticsRequirements(this.opts)
  57   }
  58
  59   /** @inheritDoc */
  60   public reset (): boolean {
  61     this.workersVirtualTaskEndTimestamp = []
  62     return true
  63   }
  64
  65   /** @inheritDoc */
  66   public update (workerNodeKey: number): boolean {
  67     this.computeWorkerVirtualTaskEndTimestamp(workerNodeKey)
  68     return true
  69   }
  70
  71   /** @inheritDoc */
  72   public choose (): number | undefined {
  73     this.nextWorkerNodeKey = this.fairShareNextWorkerNodeKey()
  74     return this.nextWorkerNodeKey
  75   }
  76
  77   /** @inheritDoc */
  78   public remove (workerNodeKey: number): boolean {
  79     this.workersVirtualTaskEndTimestamp.splice(workerNodeKey, 1)
  80     return true
  81   }
  82
  83   private fairShareNextWorkerNodeKey (): number | undefined {
  84     let minWorkerVirtualTaskEndTimestamp = Infinity
  85     let chosenWorkerNodeKey: number | undefined
  86     for (const [workerNodeKey] of this.pool.workerNodes.entries()) {
  87       if (this.workersVirtualTaskEndTimestamp[workerNodeKey] == null) {
  88         this.computeWorkerVirtualTaskEndTimestamp(workerNodeKey)
  89       }
  90       const workerVirtualTaskEndTimestamp =
  91         this.workersVirtualTaskEndTimestamp[workerNodeKey]
  92       if (
  93         this.isWorkerNodeEligible(workerNodeKey) &&
  94         workerVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  95       ) {
  96         minWorkerVirtualTaskEndTimestamp = workerVirtualTaskEndTimestamp
  97         chosenWorkerNodeKey = workerNodeKey
  98       }
  99     }
 100     return chosenWorkerNodeKey
 101   }
 102
 103   /**
 104    * Computes the worker node key virtual task end timestamp.
 105    *
 106    * @param workerNodeKey - The worker node key.
 107    */
 108   private computeWorkerVirtualTaskEndTimestamp (workerNodeKey: number): void {
 109     this.workersVirtualTaskEndTimestamp[workerNodeKey] =
 110       this.getWorkerVirtualTaskEndTimestamp(
 111         workerNodeKey,
 112         this.getWorkerVirtualTaskStartTimestamp(workerNodeKey)
 113       )
 114   }
 115
 116   private getWorkerVirtualTaskEndTimestamp (
 117     workerNodeKey: number,
 118     workerVirtualTaskStartTimestamp: number
 119   ): number {
 120     const workerTaskRunTime =
 121       this.opts.measurement === Measurements.elu
 122         ? this.getWorkerTaskElu(workerNodeKey)
 123         : this.getWorkerTaskRunTime(workerNodeKey)
 124     return workerVirtualTaskStartTimestamp + workerTaskRunTime
 125   }
 126
 127   private getWorkerVirtualTaskStartTimestamp (workerNodeKey: number): number {
 128     return Math.max(
 129       performance.now(),
 130       this.workersVirtualTaskEndTimestamp[workerNodeKey] ?? -Infinity
 131     )
 132   }
 133 }