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   /** @inheritDoc */
  97   public abstract reset (): boolean
  98
  99   /** @inheritDoc */
 100   public abstract update (workerNodeKey: number): boolean
 101
 102   /** @inheritDoc */
 103   public abstract choose (): number | undefined
 104
 105   /** @inheritDoc */
 106   public abstract remove (workerNodeKey: number): boolean
 107
 108   /** @inheritDoc */
 109   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 110     this.opts = { ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, ...opts }
 111     this.setTaskStatisticsRequirements(this.opts)
 112   }
 113
 114   /**
 115    * Whether the worker node is ready or not.
 116    *
 117    * @param workerNodeKey - The worker node key.
 118    * @returns Whether the worker node is ready or not.
 119    */
 120   private isWorkerNodeReady (workerNodeKey: number): boolean {
 121     return this.pool.workerNodes[workerNodeKey].info.ready
 122   }
 123
 124   /**
 125    * Whether the worker node has back pressure or not (i.e. its tasks queue is full).
 126    *
 127    * @param workerNodeKey - The worker node key.
 128    * @returns `true` if the worker node has back pressure, `false` otherwise.
 129    */
 130   private hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
 131     return this.pool.hasWorkerNodeBackPressure(workerNodeKey)
 132   }
 133
 134   /**
 135    * Whether the worker node is eligible or not.
 136    * A worker node is eligible if it is ready and does not have back pressure.
 137    *
 138    * @param workerNodeKey - The worker node key.
 139    * @returns `true` if the worker node is eligible, `false` otherwise.
 140    * @see {@link isWorkerNodeReady}
 141    * @see {@link hasWorkerNodeBackPressure}
 142    */
 143   protected isWorkerNodeEligible (workerNodeKey: number): boolean {
 144     return (
 145       this.isWorkerNodeReady(workerNodeKey) &&
 146       !this.hasWorkerNodeBackPressure(workerNodeKey)
 147     )
 148   }
 149
 150   /**
 151    * Gets the worker task runtime.
 152    * If the task statistics require the average runtime, the average runtime is returned.
 153    * If the task statistics require the median runtime , the median runtime is returned.
 154    *
 155    * @param workerNodeKey - The worker node key.
 156    * @returns The worker task runtime.
 157    */
 158   protected getWorkerTaskRunTime (workerNodeKey: number): number {
 159     return this.taskStatisticsRequirements.runTime.median
 160       ? this.pool.workerNodes[workerNodeKey].usage.runTime?.median ?? 0
 161       : this.pool.workerNodes[workerNodeKey].usage.runTime?.average ?? 0
 162   }
 163
 164   /**
 165    * Gets the worker task wait time.
 166    * If the task statistics require the average wait time, the average wait time is returned.
 167    * If the task statistics require the median wait time, the median wait time is returned.
 168    *
 169    * @param workerNodeKey - The worker node key.
 170    * @returns The worker task wait time.
 171    */
 172   protected getWorkerTaskWaitTime (workerNodeKey: number): number {
 173     return this.taskStatisticsRequirements.waitTime.median
 174       ? this.pool.workerNodes[workerNodeKey].usage.waitTime?.median ?? 0
 175       : this.pool.workerNodes[workerNodeKey].usage.waitTime?.average ?? 0
 176   }
 177
 178   /**
 179    * Gets the worker task ELU.
 180    * If the task statistics require the average ELU, the average ELU is returned.
 181    * If the task statistics require the median ELU, the median ELU is returned.
 182    *
 183    * @param workerNodeKey - The worker node key.
 184    * @returns The worker task ELU.
 185    */
 186   protected getWorkerTaskElu (workerNodeKey: number): number {
 187     return this.taskStatisticsRequirements.elu.median
 188       ? this.pool.workerNodes[workerNodeKey].usage.elu.active?.median ?? 0
 189       : this.pool.workerNodes[workerNodeKey].usage.elu.active?.average ?? 0
 190   }
 191
 192   /**
 193    * Assign to nextWorkerNodeKey property the chosen worker node key.
 194    *
 195    * @param chosenWorkerNodeKey - The chosen worker node key.
 196    */
 197   protected assignChosenWorkerNodeKey (
 198     chosenWorkerNodeKey: number | undefined
 199   ): void {
 200     if (chosenWorkerNodeKey != null) {
 201       this.nextWorkerNodeKey = chosenWorkerNodeKey
 202     } else {
 203       this.nextWorkerNodeKey = undefined
 204     }
 205   }
 206
 207   protected checkNextWorkerNodeEligibility (
 208     chosenWorkerNodeKey: number | undefined
 209   ): void {
 210     if (!this.isWorkerNodeEligible(this.nextWorkerNodeKey as number)) {
 211       this.nextWorkerNodeKey = undefined
 212       this.previousWorkerNodeKey =
 213         chosenWorkerNodeKey ?? this.previousWorkerNodeKey
 214     }
 215   }
 216
 217   protected computeDefaultWorkerWeight (): number {
 218     let cpusCycleTimeWeight = 0
 219     for (const cpu of cpus()) {
 220       // CPU estimated cycle time
 221       const numberOfDigits = cpu.speed.toString().length - 1
 222       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 223       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 224     }
 225     return Math.round(cpusCycleTimeWeight / cpus().length)
 226   }
 227 }