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

   1 import { DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS } from '../../utils.js'
   2 import type { IPool } from '../pool.js'
   3 import type { IWorker } from '../worker.js'
   4 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy.js'
   5 import {
   6   type IWorkerChoiceStrategy,
   7   Measurements,
   8   type TaskStatisticsRequirements,
   9   type WorkerChoiceStrategyOptions
  10 } from './selection-strategies-types.js'
  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 structured-cloneable data.
  18  * @typeParam Response - Type of execution response. This can only be structured-cloneable 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: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  35     elu: {
  36       aggregate: true,
  37       average: true,
  38       median: false
  39     }
  40   }
  41
  42   /** @inheritDoc */
  43   public constructor (
  44     pool: IPool<Worker, Data, Response>,
  45     opts?: WorkerChoiceStrategyOptions
  46   ) {
  47     super(pool, opts)
  48     this.setTaskStatisticsRequirements(this.opts)
  49   }
  50
  51   /** @inheritDoc */
  52   public reset (): boolean {
  53     for (const workerNode of this.pool.workerNodes) {
  54       delete workerNode.strategyData?.virtualTaskEndTimestamp
  55     }
  56     return true
  57   }
  58
  59   /** @inheritDoc */
  60   public update (workerNodeKey: number): boolean {
  61     this.pool.workerNodes[workerNodeKey].strategyData = {
  62       virtualTaskEndTimestamp:
  63         this.computeWorkerNodeVirtualTaskEndTimestamp(workerNodeKey)
  64     }
  65     return true
  66   }
  67
  68   /** @inheritDoc */
  69   public choose (): number | undefined {
  70     this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey)
  71     this.nextWorkerNodeKey = this.fairShareNextWorkerNodeKey()
  72     return this.nextWorkerNodeKey
  73   }
  74
  75   /** @inheritDoc */
  76   public remove (): boolean {
  77     return true
  78   }
  79
  80   private fairShareNextWorkerNodeKey (): number | undefined {
  81     if (this.pool.workerNodes.length === 0) {
  82       return undefined
  83     }
  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           // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
  94           workerNode.strategyData.virtualTaskEndTimestamp! <
  95             // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
  96             workerNodes[minWorkerNodeKey].strategyData!.virtualTaskEndTimestamp!
  97           ? workerNodeKey
  98           : minWorkerNodeKey
  99       },
 100       0
 101     )
 102   }
 103
 104   /**
 105    * Computes the worker node key virtual task end timestamp.
 106    *
 107    * @param workerNodeKey - The worker node key.
 108    * @returns The worker node key virtual task end timestamp.
 109    */
 110   private computeWorkerNodeVirtualTaskEndTimestamp (
 111     workerNodeKey: number
 112   ): number {
 113     return this.getWorkerNodeVirtualTaskEndTimestamp(
 114       workerNodeKey,
 115       this.getWorkerNodeVirtualTaskStartTimestamp(workerNodeKey)
 116     )
 117   }
 118
 119   private getWorkerNodeVirtualTaskEndTimestamp (
 120     workerNodeKey: number,
 121     workerNodeVirtualTaskStartTimestamp: number
 122   ): number {
 123     const workerNodeTaskRunTime =
 124       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 125       this.opts!.measurement === Measurements.elu
 126         ? this.getWorkerNodeTaskElu(workerNodeKey)
 127         : this.getWorkerNodeTaskRunTime(workerNodeKey)
 128     return workerNodeVirtualTaskStartTimestamp + workerNodeTaskRunTime
 129   }
 130
 131   private getWorkerNodeVirtualTaskStartTimestamp (
 132     workerNodeKey: number
 133   ): number {
 134     const virtualTaskEndTimestamp =
 135       this.pool.workerNodes[workerNodeKey]?.strategyData
 136         ?.virtualTaskEndTimestamp
 137     const now = performance.now()
 138     return now < (virtualTaskEndTimestamp ?? -Infinity)
 139       ? // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 140       virtualTaskEndTimestamp!
 141       : now
 142   }
 143 }