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 type {
   6   IWorkerChoiceStrategy,
   7   TaskStatisticsRequirements,
   8   WorkerChoiceStrategyOptions
   9 } from './selection-strategies-types'
  10
  11 /**
  12  * Selects the next worker with a fair share scheduling algorithm.
  13  * Loosely modeled after the fair queueing algorithm: https://en.wikipedia.org/wiki/Fair_queuing.
  14  *
  15  * @typeParam Worker - Type of worker which manages the strategy.
  16  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  17  * @typeParam Response - Type of execution response. This can only be serializable data.
  18  */
  19 export class FairShareWorkerChoiceStrategy<
  20     Worker extends IWorker,
  21     Data = unknown,
  22     Response = unknown
  23   >
  24   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  25   implements IWorkerChoiceStrategy {
  26   /** @inheritDoc */
  27   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  28     runTime: {
  29       aggregate: true,
  30       average: true,
  31       median: false
  32     },
  33     waitTime: {
  34       aggregate: false,
  35       average: false,
  36       median: false
  37     },
  38     elu: {
  39       aggregate: false,
  40       average: false,
  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 {
  73     let minWorkerVirtualTaskEndTimestamp = Infinity
  74     let chosenWorkerNodeKey!: number
  75     for (const [workerNodeKey] of this.pool.workerNodes.entries()) {
  76       if (this.workersVirtualTaskEndTimestamp[workerNodeKey] == null) {
  77         this.computeWorkerVirtualTaskEndTimestamp(workerNodeKey)
  78       }
  79       const workerVirtualTaskEndTimestamp =
  80         this.workersVirtualTaskEndTimestamp[workerNodeKey]
  81       if (workerVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp) {
  82         minWorkerVirtualTaskEndTimestamp = workerVirtualTaskEndTimestamp
  83         chosenWorkerNodeKey = workerNodeKey
  84       }
  85     }
  86     return chosenWorkerNodeKey
  87   }
  88
  89   /** @inheritDoc */
  90   public remove (workerNodeKey: number): boolean {
  91     this.workersVirtualTaskEndTimestamp.splice(workerNodeKey, 1)
  92     return true
  93   }
  94
  95   /**
  96    * Computes the worker node key virtual task end timestamp.
  97    *
  98    * @param workerNodeKey - The worker node key.
  99    */
 100   private computeWorkerVirtualTaskEndTimestamp (workerNodeKey: number): void {
 101     this.workersVirtualTaskEndTimestamp[workerNodeKey] =
 102       this.getWorkerVirtualTaskEndTimestamp(
 103         workerNodeKey,
 104         this.getWorkerVirtualTaskStartTimestamp(workerNodeKey)
 105       )
 106   }
 107
 108   private getWorkerVirtualTaskEndTimestamp (
 109     workerNodeKey: number,
 110     workerVirtualTaskStartTimestamp: number
 111   ): number {
 112     return (
 113       workerVirtualTaskStartTimestamp + this.getWorkerTaskRunTime(workerNodeKey)
 114     )
 115   }
 116
 117   private getWorkerVirtualTaskStartTimestamp (workerNodeKey: number): number {
 118     return Math.max(
 119       performance.now(),
 120       this.workersVirtualTaskEndTimestamp[workerNodeKey] ?? -Infinity
 121     )
 122   }
 123 }