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

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