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, StrategyData } 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   /** @inheritDoc */
  46   public constructor (
  47     pool: IPool<Worker, Data, Response>,
  48     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  49   ) {
  50     super(pool, opts)
  51     this.setTaskStatisticsRequirements(this.opts)
  52   }
  53
  54   /** @inheritDoc */
  55   public reset (): boolean {
  56     for (const workerNode of this.pool.workerNodes) {
  57       delete workerNode.strategyData?.virtualTaskEndTimestamp
  58     }
  59     return true
  60   }
  61
  62   /** @inheritDoc */
  63   public update (workerNodeKey: number): boolean {
  64     this.pool.workerNodes[workerNodeKey].strategyData = {
  65       virtualTaskEndTimestamp:
  66         this.computeWorkerNodeVirtualTaskEndTimestamp(workerNodeKey)
  67     }
  68     return true
  69   }
  70
  71   /** @inheritDoc */
  72   public choose (): number | undefined {
  73     this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey)
  74     this.nextWorkerNodeKey = this.fairShareNextWorkerNodeKey()
  75     return this.nextWorkerNodeKey
  76   }
  77
  78   /** @inheritDoc */
  79   public remove (): boolean {
  80     return true
  81   }
  82
  83   private fairShareNextWorkerNodeKey (): number | undefined {
  84     return this.pool.workerNodes.reduce(
  85       (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
  86         if (workerNode.strategyData?.virtualTaskEndTimestamp == null) {
  87           workerNode.strategyData = {
  88             virtualTaskEndTimestamp:
  89               this.computeWorkerNodeVirtualTaskEndTimestamp(workerNodeKey)
  90           }
  91         }
  92         return this.isWorkerNodeReady(workerNodeKey) &&
  93           (workerNode.strategyData.virtualTaskEndTimestamp as number) <
  94             ((workerNodes[minWorkerNodeKey].strategyData as StrategyData)
  95               .virtualTaskEndTimestamp as number)
  96           ? workerNodeKey
  97           : minWorkerNodeKey
  98       },
  99       0
 100     )
 101   }
 102
 103   /**
 104    * Computes the worker node key virtual task end timestamp.
 105    *
 106    * @param workerNodeKey - The worker node key.
 107    * @returns The worker node key virtual task end timestamp.
 108    */
 109   private computeWorkerNodeVirtualTaskEndTimestamp (
 110     workerNodeKey: number
 111   ): number {
 112     return this.getWorkerNodeVirtualTaskEndTimestamp(
 113       workerNodeKey,
 114       this.getWorkerNodeVirtualTaskStartTimestamp(workerNodeKey)
 115     )
 116   }
 117
 118   private getWorkerNodeVirtualTaskEndTimestamp (
 119     workerNodeKey: number,
 120     workerNodeVirtualTaskStartTimestamp: number
 121   ): number {
 122     const workerNodeTaskRunTime =
 123       this.opts.measurement === Measurements.elu
 124         ? this.getWorkerNodeTaskElu(workerNodeKey)
 125         : this.getWorkerNodeTaskRunTime(workerNodeKey)
 126     return workerNodeVirtualTaskStartTimestamp + workerNodeTaskRunTime
 127   }
 128
 129   private getWorkerNodeVirtualTaskStartTimestamp (
 130     workerNodeKey: number
 131   ): number {
 132     const virtualTaskEndTimestamp =
 133       this.pool.workerNodes[workerNodeKey]?.strategyData
 134         ?.virtualTaskEndTimestamp
 135     const now = performance.now()
 136     return now < (virtualTaskEndTimestamp ?? -Infinity)
 137       ? (virtualTaskEndTimestamp as number)
 138       : now
 139   }
 140 }