src/pools/selection-strategies/abstract-worker-choice-strategy.ts

   1 import { cpus } from 'node:os'
   2 import {
   3   DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
   4   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
   5 } from '../../utils'
   6 import type { IPool } from '../pool'
   7 import type { IWorker } from '../worker'
   8 import type {
   9   IWorkerChoiceStrategy,
  10   MeasurementStatisticsRequirements,
  11   StrategyPolicy,
  12   TaskStatisticsRequirements,
  13   WorkerChoiceStrategyOptions
  14 } from './selection-strategies-types'
  15
  16 /**
  17  * Worker choice strategy abstract base class.
  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 abstract class AbstractWorkerChoiceStrategy<
  24   Worker extends IWorker,
  25   Data = unknown,
  26   Response = unknown
  27 > implements IWorkerChoiceStrategy {
  28   /**
  29    * The next worker node key.
  30    */
  31   protected nextWorkerNodeKey: number | undefined = 0
  32
  33   /**
  34    * The previous worker node key.
  35    */
  36   protected previousWorkerNodeKey: number = 0
  37
  38   /** @inheritDoc */
  39   public readonly strategyPolicy: StrategyPolicy = {
  40     dynamicWorkerUsage: false,
  41     dynamicWorkerReady: true
  42   }
  43
  44   /** @inheritDoc */
  45   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  46     runTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  47     waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  48     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  49   }
  50
  51   /**
  52    * Constructs a worker choice strategy bound to the pool.
  53    *
  54    * @param pool - The pool instance.
  55    * @param opts - The worker choice strategy options.
  56    */
  57   public constructor (
  58     protected readonly pool: IPool<Worker, Data, Response>,
  59     protected opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  60   ) {
  61     this.opts = { ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, ...opts }
  62     this.choose = this.choose.bind(this)
  63   }
  64
  65   protected setTaskStatisticsRequirements (
  66     opts: WorkerChoiceStrategyOptions
  67   ): void {
  68     this.toggleMedianMeasurementStatisticsRequirements(
  69       this.taskStatisticsRequirements.runTime,
  70       opts.runTime?.median as boolean
  71     )
  72     this.toggleMedianMeasurementStatisticsRequirements(
  73       this.taskStatisticsRequirements.waitTime,
  74       opts.waitTime?.median as boolean
  75     )
  76     this.toggleMedianMeasurementStatisticsRequirements(
  77       this.taskStatisticsRequirements.elu,
  78       opts.elu?.median as boolean
  79     )
  80   }
  81
  82   private toggleMedianMeasurementStatisticsRequirements (
  83     measurementStatisticsRequirements: MeasurementStatisticsRequirements,
  84     toggleMedian: boolean
  85   ): void {
  86     if (measurementStatisticsRequirements.average && toggleMedian) {
  87       measurementStatisticsRequirements.average = false
  88       measurementStatisticsRequirements.median = toggleMedian
  89     }
  90     if (measurementStatisticsRequirements.median && !toggleMedian) {
  91       measurementStatisticsRequirements.average = true
  92       measurementStatisticsRequirements.median = toggleMedian
  93     }
  94   }
  95
  96   protected resetWorkerNodeKeyProperties (): void {
  97     this.nextWorkerNodeKey = 0
  98     this.previousWorkerNodeKey = 0
  99   }
 100
 101   /** @inheritDoc */
 102   public abstract reset (): boolean
 103
 104   /** @inheritDoc */
 105   public abstract update (workerNodeKey: number): boolean
 106
 107   /** @inheritDoc */
 108   public abstract choose (): number | undefined
 109
 110   /** @inheritDoc */
 111   public abstract remove (workerNodeKey: number): boolean
 112
 113   /** @inheritDoc */
 114   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 115     this.opts = { ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, ...opts }
 116     this.setTaskStatisticsRequirements(this.opts)
 117   }
 118
 119   /**
 120    * Whether the worker node is ready or not.
 121    *
 122    * @param workerNodeKey - The worker node key.
 123    * @returns Whether the worker node is ready or not.
 124    */
 125   private isWorkerNodeReady (workerNodeKey: number): boolean {
 126     return this.pool.workerNodes[workerNodeKey].info.ready
 127   }
 128
 129   /**
 130    * Whether the worker node has back pressure or not (i.e. its tasks queue is full).
 131    *
 132    * @param workerNodeKey - The worker node key.
 133    * @returns `true` if the worker node has back pressure, `false` otherwise.
 134    */
 135   private hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
 136     return this.pool.hasWorkerNodeBackPressure(workerNodeKey)
 137   }
 138
 139   /**
 140    * Whether the worker node is eligible or not.
 141    * A worker node is eligible if it is ready and does not have back pressure.
 142    *
 143    * @param workerNodeKey - The worker node key.
 144    * @returns `true` if the worker node is eligible, `false` otherwise.
 145    * @see {@link isWorkerNodeReady}
 146    * @see {@link hasWorkerNodeBackPressure}
 147    */
 148   protected isWorkerNodeEligible (workerNodeKey: number): boolean {
 149     return (
 150       this.isWorkerNodeReady(workerNodeKey) &&
 151       !this.hasWorkerNodeBackPressure(workerNodeKey)
 152     )
 153   }
 154
 155   /**
 156    * Gets the worker node task runtime.
 157    * If the task statistics require the average runtime, the average runtime is returned.
 158    * If the task statistics require the median runtime , the median runtime is returned.
 159    *
 160    * @param workerNodeKey - The worker node key.
 161    * @returns The worker node task runtime.
 162    */
 163   protected getWorkerNodeTaskRunTime (workerNodeKey: number): number {
 164     return this.taskStatisticsRequirements.runTime.median
 165       ? this.pool.workerNodes[workerNodeKey]?.usage?.runTime?.median ?? 0
 166       : this.pool.workerNodes[workerNodeKey]?.usage?.runTime?.average ?? 0
 167   }
 168
 169   /**
 170    * Gets the worker node task wait time.
 171    * If the task statistics require the average wait time, the average wait time is returned.
 172    * If the task statistics require the median wait time, the median wait time is returned.
 173    *
 174    * @param workerNodeKey - The worker node key.
 175    * @returns The worker node task wait time.
 176    */
 177   protected getWorkerNodeTaskWaitTime (workerNodeKey: number): number {
 178     return this.taskStatisticsRequirements.waitTime.median
 179       ? this.pool.workerNodes[workerNodeKey]?.usage?.waitTime?.median ?? 0
 180       : this.pool.workerNodes[workerNodeKey]?.usage?.waitTime?.average ?? 0
 181   }
 182
 183   /**
 184    * Gets the worker node task ELU.
 185    * If the task statistics require the average ELU, the average ELU is returned.
 186    * If the task statistics require the median ELU, the median ELU is returned.
 187    *
 188    * @param workerNodeKey - The worker node key.
 189    * @returns The worker node task ELU.
 190    */
 191   protected getWorkerNodeTaskElu (workerNodeKey: number): number {
 192     return this.taskStatisticsRequirements.elu.median
 193       ? this.pool.workerNodes[workerNodeKey]?.usage?.elu?.active?.median ?? 0
 194       : this.pool.workerNodes[workerNodeKey]?.usage?.elu?.active?.average ?? 0
 195   }
 196
 197   /**
 198    * Sets safely the previous worker node key.
 199    *
 200    * @param workerNodeKey - The worker node key.
 201    */
 202   protected setPreviousWorkerNodeKey (workerNodeKey: number | undefined): void {
 203     this.previousWorkerNodeKey = workerNodeKey ?? this.previousWorkerNodeKey
 204   }
 205
 206   /**
 207    * Check the next worker node eligibility.
 208    */
 209   protected checkNextWorkerNodeEligibility (): void {
 210     if (!this.isWorkerNodeEligible(this.nextWorkerNodeKey as number)) {
 211       this.nextWorkerNodeKey = undefined
 212     }
 213   }
 214
 215   protected computeDefaultWorkerWeight (): number {
 216     let cpusCycleTimeWeight = 0
 217     for (const cpu of cpus()) {
 218       // CPU estimated cycle time
 219       const numberOfDigits = cpu.speed.toString().length - 1
 220       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 221       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 222     }
 223     return Math.round(cpusCycleTimeWeight / cpus().length)
 224   }
 225 }