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

   1 import { cpus } from 'node:os'
   2 import {
   3   DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
   4   getDefaultInternalWorkerChoiceStrategyOptions
   5 } from '../../utils'
   6 import type { IPool } from '../pool'
   7 import type { IWorker } from '../worker'
   8 import type {
   9   IWorkerChoiceStrategy,
  10   InternalWorkerChoiceStrategyOptions,
  11   MeasurementStatisticsRequirements,
  12   StrategyPolicy,
  13   TaskStatisticsRequirements
  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: InternalWorkerChoiceStrategyOptions
  60   ) {
  61     this.setOptions(this.opts)
  62     this.choose = this.choose.bind(this)
  63   }
  64
  65   protected setTaskStatisticsRequirements (
  66     opts: InternalWorkerChoiceStrategyOptions
  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: InternalWorkerChoiceStrategyOptions): void {
 115     this.opts = {
 116       ...getDefaultInternalWorkerChoiceStrategyOptions(
 117         this.pool.info.maxSize + Object.keys(opts?.weights ?? {}).length
 118       ),
 119       ...opts
 120     }
 121     this.setTaskStatisticsRequirements(this.opts)
 122   }
 123
 124   /** @inheritDoc */
 125   public hasPoolWorkerNodesReady (): boolean {
 126     return this.pool.workerNodes.some(workerNode => workerNode.info.ready)
 127   }
 128
 129   /**
 130    * Whether the worker node is ready or not.
 131    *
 132    * @param workerNodeKey - The worker node key.
 133    * @returns Whether the worker node is ready or not.
 134    */
 135   protected isWorkerNodeReady (workerNodeKey: number): boolean {
 136     return this.pool.workerNodes[workerNodeKey]?.info?.ready ?? false
 137   }
 138
 139   /**
 140    * Check the next worker node readiness.
 141    */
 142   protected checkNextWorkerNodeReadiness (): void {
 143     if (!this.isWorkerNodeReady(this.nextWorkerNodeKey as number)) {
 144       delete this.nextWorkerNodeKey
 145     }
 146   }
 147
 148   /**
 149    * Gets the worker node task runtime.
 150    * If the task statistics require the average runtime, the average runtime is returned.
 151    * If the task statistics require the median runtime , the median runtime is returned.
 152    *
 153    * @param workerNodeKey - The worker node key.
 154    * @returns The worker node task runtime.
 155    */
 156   protected getWorkerNodeTaskRunTime (workerNodeKey: number): number {
 157     return this.taskStatisticsRequirements.runTime.median
 158       ? this.pool.workerNodes[workerNodeKey].usage.runTime.median ?? 0
 159       : this.pool.workerNodes[workerNodeKey].usage.runTime.average ?? 0
 160   }
 161
 162   /**
 163    * Gets the worker node task wait time.
 164    * If the task statistics require the average wait time, the average wait time is returned.
 165    * If the task statistics require the median wait time, the median wait time is returned.
 166    *
 167    * @param workerNodeKey - The worker node key.
 168    * @returns The worker node task wait time.
 169    */
 170   protected getWorkerNodeTaskWaitTime (workerNodeKey: number): number {
 171     return this.taskStatisticsRequirements.waitTime.median
 172       ? this.pool.workerNodes[workerNodeKey].usage.waitTime.median ?? 0
 173       : this.pool.workerNodes[workerNodeKey].usage.waitTime.average ?? 0
 174   }
 175
 176   /**
 177    * Gets the worker node task ELU.
 178    * If the task statistics require the average ELU, the average ELU is returned.
 179    * If the task statistics require the median ELU, the median ELU is returned.
 180    *
 181    * @param workerNodeKey - The worker node key.
 182    * @returns The worker node task ELU.
 183    */
 184   protected getWorkerNodeTaskElu (workerNodeKey: number): number {
 185     return this.taskStatisticsRequirements.elu.median
 186       ? this.pool.workerNodes[workerNodeKey].usage.elu.active.median ?? 0
 187       : this.pool.workerNodes[workerNodeKey].usage.elu.active.average ?? 0
 188   }
 189
 190   /**
 191    * Sets safely the previous worker node key.
 192    *
 193    * @param workerNodeKey - The worker node key.
 194    */
 195   protected setPreviousWorkerNodeKey (workerNodeKey: number | undefined): void {
 196     this.previousWorkerNodeKey = workerNodeKey ?? this.previousWorkerNodeKey
 197   }
 198
 199   protected computeDefaultWorkerWeight (): number {
 200     let cpusCycleTimeWeight = 0
 201     for (const cpu of cpus()) {
 202       // CPU estimated cycle time
 203       const numberOfDigits = cpu.speed.toString().length - 1
 204       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 205       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 206     }
 207     return Math.round(cpusCycleTimeWeight / cpus().length)
 208   }
 209 }