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