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