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