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   protected isWorkerNodeReady (workerNodeKey: number): boolean {
 126     return this.pool.workerNodes[workerNodeKey]?.info.ready ?? false
 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   protected hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
 136     return this.pool.hasWorkerNodeBackPressure(workerNodeKey)
 137   }
 138
 139   /**
 140    * Gets the worker node task runtime.
 141    * If the task statistics require the average runtime, the average runtime is returned.
 142    * If the task statistics require the median runtime , the median runtime is returned.
 143    *
 144    * @param workerNodeKey - The worker node key.
 145    * @returns The worker node task runtime.
 146    */
 147   protected getWorkerNodeTaskRunTime (workerNodeKey: number): number {
 148     return this.taskStatisticsRequirements.runTime.median
 149       ? this.pool.workerNodes[workerNodeKey].usage.runTime.median ?? 0
 150       : this.pool.workerNodes[workerNodeKey].usage.runTime.average ?? 0
 151   }
 152
 153   /**
 154    * Gets the worker node task wait time.
 155    * If the task statistics require the average wait time, the average wait time is returned.
 156    * If the task statistics require the median wait time, the median wait time is returned.
 157    *
 158    * @param workerNodeKey - The worker node key.
 159    * @returns The worker node task wait time.
 160    */
 161   protected getWorkerNodeTaskWaitTime (workerNodeKey: number): number {
 162     return this.taskStatisticsRequirements.waitTime.median
 163       ? this.pool.workerNodes[workerNodeKey].usage.waitTime.median ?? 0
 164       : this.pool.workerNodes[workerNodeKey].usage.waitTime.average ?? 0
 165   }
 166
 167   /**
 168    * Gets the worker node task ELU.
 169    * If the task statistics require the average ELU, the average ELU is returned.
 170    * If the task statistics require the median ELU, the median ELU is returned.
 171    *
 172    * @param workerNodeKey - The worker node key.
 173    * @returns The worker node task ELU.
 174    */
 175   protected getWorkerNodeTaskElu (workerNodeKey: number): number {
 176     return this.taskStatisticsRequirements.elu.median
 177       ? this.pool.workerNodes[workerNodeKey].usage.elu.active.median ?? 0
 178       : this.pool.workerNodes[workerNodeKey].usage.elu.active.average ?? 0
 179   }
 180
 181   /**
 182    * Sets safely the previous worker node key.
 183    *
 184    * @param workerNodeKey - The worker node key.
 185    */
 186   protected setPreviousWorkerNodeKey (workerNodeKey: number | undefined): void {
 187     this.previousWorkerNodeKey = workerNodeKey ?? this.previousWorkerNodeKey
 188   }
 189
 190   protected computeDefaultWorkerWeight (): number {
 191     let cpusCycleTimeWeight = 0
 192     for (const cpu of cpus()) {
 193       // CPU estimated cycle time
 194       const numberOfDigits = cpu.speed.toString().length - 1
 195       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 196       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 197     }
 198     return Math.round(cpusCycleTimeWeight / cpus().length)
 199   }
 200 }