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

   1 import type { IWorker } from '../worker'
   2 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   3 import type {
   4   IWorkerChoiceStrategy,
   5   RequiredStatistics
   6 } from './selection-strategies-types'
   7
   8 /**
   9  * Worker virtual task timestamp.
  10  */
  11 interface WorkerVirtualTaskTimestamp {
  12   start: number
  13   end: number
  14 }
  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 serializable data.
  22  * @typeParam Response - Type of execution response. This can only be serializable 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 requiredStatistics: RequiredStatistics = {
  33     runTime: true,
  34     avgRunTime: true,
  35     medRunTime: false
  36   }
  37
  38   /**
  39    * Worker last virtual task execution timestamp.
  40    */
  41   private readonly workerLastVirtualTaskTimestamp: Map<
  42   number,
  43   WorkerVirtualTaskTimestamp
  44   > = new Map<number, WorkerVirtualTaskTimestamp>()
  45
  46   /** @inheritDoc */
  47   public reset (): boolean {
  48     this.workerLastVirtualTaskTimestamp.clear()
  49     return true
  50   }
  51
  52   /** @inheritDoc */
  53   public choose (): number {
  54     let minWorkerVirtualTaskEndTimestamp = Infinity
  55     let chosenWorkerNodeKey!: number
  56     for (const [index] of this.pool.workerNodes.entries()) {
  57       this.computeWorkerLastVirtualTaskTimestamp(index)
  58       const workerLastVirtualTaskEndTimestamp =
  59         this.workerLastVirtualTaskTimestamp.get(index)?.end ?? 0
  60       if (
  61         workerLastVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  62       ) {
  63         minWorkerVirtualTaskEndTimestamp = workerLastVirtualTaskEndTimestamp
  64         chosenWorkerNodeKey = index
  65       }
  66     }
  67     return chosenWorkerNodeKey
  68   }
  69
  70   /** @inheritDoc */
  71   public remove (workerNodeKey: number): boolean {
  72     const deleted = this.workerLastVirtualTaskTimestamp.delete(workerNodeKey)
  73     for (const [key, value] of this.workerLastVirtualTaskTimestamp.entries()) {
  74       if (key > workerNodeKey) {
  75         this.workerLastVirtualTaskTimestamp.set(key - 1, value)
  76       }
  77     }
  78     return deleted
  79   }
  80
  81   /**
  82    * Computes worker last virtual task timestamp.
  83    *
  84    * @param workerNodeKey - The worker node key.
  85    */
  86   private computeWorkerLastVirtualTaskTimestamp (workerNodeKey: number): void {
  87     const workerVirtualTaskStartTimestamp = Math.max(
  88       performance.now(),
  89       this.workerLastVirtualTaskTimestamp.get(workerNodeKey)?.end ?? -Infinity
  90     )
  91     const workerVirtualTaskTRunTime = this.requiredStatistics.medRunTime
  92       ? this.pool.workerNodes[workerNodeKey].tasksUsage.medRunTime
  93       : this.pool.workerNodes[workerNodeKey].tasksUsage.avgRunTime
  94     this.workerLastVirtualTaskTimestamp.set(workerNodeKey, {
  95       start: workerVirtualTaskStartTimestamp,
  96       end: workerVirtualTaskStartTimestamp + (workerVirtualTaskTRunTime ?? 0)
  97     })
  98   }
  99 }