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 = 0
  32
  33   /** @inheritDoc */
  34   public readonly strategyPolicy: StrategyPolicy = {
  35     useDynamicWorker: false
  36   }
  37
  38   /** @inheritDoc */
  39   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  40     runTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  41     waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  42     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  43   }
  44
  45   /**
  46    * Constructs a worker choice strategy bound to the pool.
  47    *
  48    * @param pool - The pool instance.
  49    * @param opts - The worker choice strategy options.
  50    */
  51   public constructor (
  52     protected readonly pool: IPool<Worker, Data, Response>,
  53     protected opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  54   ) {
  55     this.choose = this.choose.bind(this)
  56   }
  57
  58   protected setTaskStatisticsRequirements (
  59     opts: WorkerChoiceStrategyOptions
  60   ): void {
  61     this.toggleMedianMeasurementStatisticsRequirements(
  62       this.taskStatisticsRequirements.runTime,
  63       opts.runTime?.median as boolean
  64     )
  65     this.toggleMedianMeasurementStatisticsRequirements(
  66       this.taskStatisticsRequirements.waitTime,
  67       opts.waitTime?.median as boolean
  68     )
  69     this.toggleMedianMeasurementStatisticsRequirements(
  70       this.taskStatisticsRequirements.elu,
  71       opts.elu?.median as boolean
  72     )
  73   }
  74
  75   private toggleMedianMeasurementStatisticsRequirements (
  76     measurementStatisticsRequirements: MeasurementStatisticsRequirements,
  77     toggleMedian: boolean
  78   ): void {
  79     if (measurementStatisticsRequirements.average && toggleMedian) {
  80       measurementStatisticsRequirements.average = false
  81       measurementStatisticsRequirements.median = toggleMedian
  82     }
  83     if (measurementStatisticsRequirements.median && !toggleMedian) {
  84       measurementStatisticsRequirements.average = true
  85       measurementStatisticsRequirements.median = toggleMedian
  86     }
  87   }
  88
  89   /** @inheritDoc */
  90   public abstract reset (): boolean
  91
  92   /** @inheritDoc */
  93   public abstract update (workerNodeKey: number): boolean
  94
  95   /** @inheritDoc */
  96   public abstract choose (): number
  97
  98   /** @inheritDoc */
  99   public abstract remove (workerNodeKey: number): boolean
 100
 101   /** @inheritDoc */
 102   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 103     this.opts = opts ?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
 104     this.setTaskStatisticsRequirements(this.opts)
 105   }
 106
 107   /**
 108    * Whether the worker node is ready or not.
 109    *
 110    * @param workerNodeKey - The worker node key.
 111    * @returns Whether the worker node is ready or not.
 112    */
 113   protected isWorkerNodeReady (workerNodeKey: number): boolean {
 114     return this.pool.workerNodes[workerNodeKey].info.ready
 115   }
 116
 117   /**
 118    * Whether the worker node has back pressure or not (i.e. its tasks queue is full).
 119    *
 120    * @param workerNodeKey - The worker node key.
 121    * @returns `true` if the worker node has back pressure, `false` otherwise.
 122    */
 123   protected hasWorkerNodeBackPressure (workerNodeKey: number): boolean {
 124     return this.pool.hasWorkerNodeBackPressure(workerNodeKey)
 125   }
 126
 127   /**
 128    * Gets the worker task runtime.
 129    * If the task statistics require the average runtime, the average runtime is returned.
 130    * If the task statistics require the median runtime , the median runtime is returned.
 131    *
 132    * @param workerNodeKey - The worker node key.
 133    * @returns The worker task runtime.
 134    */
 135   protected getWorkerTaskRunTime (workerNodeKey: number): number {
 136     return this.taskStatisticsRequirements.runTime.median
 137       ? this.pool.workerNodes[workerNodeKey].usage.runTime?.median ?? 0
 138       : this.pool.workerNodes[workerNodeKey].usage.runTime?.average ?? 0
 139   }
 140
 141   /**
 142    * Gets the worker task wait time.
 143    * If the task statistics require the average wait time, the average wait time is returned.
 144    * If the task statistics require the median wait time, the median wait time is returned.
 145    *
 146    * @param workerNodeKey - The worker node key.
 147    * @returns The worker task wait time.
 148    */
 149   protected getWorkerTaskWaitTime (workerNodeKey: number): number {
 150     return this.taskStatisticsRequirements.waitTime.median
 151       ? this.pool.workerNodes[workerNodeKey].usage.waitTime?.median ?? 0
 152       : this.pool.workerNodes[workerNodeKey].usage.waitTime?.average ?? 0
 153   }
 154
 155   /**
 156    * Gets the worker task ELU.
 157    * If the task statistics require the average ELU, the average ELU is returned.
 158    * If the task statistics require the median ELU, the median ELU is returned.
 159    *
 160    * @param workerNodeKey - The worker node key.
 161    * @returns The worker task ELU.
 162    */
 163   protected getWorkerTaskElu (workerNodeKey: number): number {
 164     return this.taskStatisticsRequirements.elu.median
 165       ? this.pool.workerNodes[workerNodeKey].usage.elu.active?.median ?? 0
 166       : this.pool.workerNodes[workerNodeKey].usage.elu.active?.average ?? 0
 167   }
 168
 169   protected computeDefaultWorkerWeight (): number {
 170     let cpusCycleTimeWeight = 0
 171     for (const cpu of cpus()) {
 172       // CPU estimated cycle time
 173       const numberOfDigits = cpu.speed.toString().length - 1
 174       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 175       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 176     }
 177     return Math.round(cpusCycleTimeWeight / cpus().length)
 178   }
 179 }