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

   1 import type { IPool } from '../pool.js'
   2 import type { IWorker } from '../worker.js'
   3 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy.js'
   4 import {
   5   type IWorkerChoiceStrategy,
   6   Measurements,
   7   type TaskStatisticsRequirements,
   8   type WorkerChoiceStrategyOptions
   9 } from './selection-strategies-types.js'
  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 structured-cloneable data.
  17  * @typeParam Response - Type of execution response. This can only be structured-cloneable 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: true,
  35       average: true,
  36       median: false
  37     },
  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
  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     if (this.pool.workerNodes.length === 0) {
  85       return undefined
  86     }
  87     return this.pool.workerNodes.reduce(
  88       (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
  89         if (workerNode.strategyData?.virtualTaskEndTimestamp == null) {
  90           workerNode.strategyData = {
  91             virtualTaskEndTimestamp:
  92               this.computeWorkerNodeVirtualTaskEndTimestamp(workerNodeKey)
  93           }
  94         }
  95         return this.isWorkerNodeReady(workerNodeKey) &&
  96           // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
  97           workerNode.strategyData.virtualTaskEndTimestamp! <
  98             // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
  99             workerNodes[minWorkerNodeKey].strategyData!.virtualTaskEndTimestamp!
 100           ? workerNodeKey
 101           : minWorkerNodeKey
 102       },
 103       0
 104     )
 105   }
 106
 107   /**
 108    * Computes the worker node key virtual task end timestamp.
 109    *
 110    * @param workerNodeKey - The worker node key.
 111    * @returns The worker node key virtual task end timestamp.
 112    */
 113   private computeWorkerNodeVirtualTaskEndTimestamp (
 114     workerNodeKey: number
 115   ): number {
 116     return this.getWorkerNodeVirtualTaskEndTimestamp(
 117       workerNodeKey,
 118       this.getWorkerNodeVirtualTaskStartTimestamp(workerNodeKey)
 119     )
 120   }
 121
 122   private getWorkerNodeVirtualTaskEndTimestamp (
 123     workerNodeKey: number,
 124     workerNodeVirtualTaskStartTimestamp: number
 125   ): number {
 126     const workerNodeTaskExecutionTime =
 127       this.getWorkerNodeTaskWaitTime(workerNodeKey) +
 128       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 129       (this.opts!.measurement === Measurements.elu
 130         ? this.getWorkerNodeTaskElu(workerNodeKey)
 131         : this.getWorkerNodeTaskRunTime(workerNodeKey))
 132     return workerNodeVirtualTaskStartTimestamp + workerNodeTaskExecutionTime
 133   }
 134
 135   private getWorkerNodeVirtualTaskStartTimestamp (
 136     workerNodeKey: number
 137   ): number {
 138     const virtualTaskEndTimestamp =
 139       this.pool.workerNodes[workerNodeKey]?.strategyData
 140         ?.virtualTaskEndTimestamp
 141     const now = performance.now()
 142     return now < (virtualTaskEndTimestamp ?? -Infinity)
 143       ? // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 144       virtualTaskEndTimestamp!
 145       : now
 146   }
 147 }