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    * Gets the worker task runtime.
 119    * If the task statistics require the average runtime, the average runtime is returned.
 120    * If the task statistics require the median runtime , the median runtime is returned.
 121    *
 122    * @param workerNodeKey - The worker node key.
 123    * @returns The worker task runtime.
 124    */
 125   protected getWorkerTaskRunTime (workerNodeKey: number): number {
 126     return this.taskStatisticsRequirements.runTime.median
 127       ? this.pool.workerNodes[workerNodeKey].usage.runTime?.median ?? 0
 128       : this.pool.workerNodes[workerNodeKey].usage.runTime?.average ?? 0
 129   }
 130
 131   /**
 132    * Gets the worker task wait time.
 133    * If the task statistics require the average wait time, the average wait time is returned.
 134    * If the task statistics require the median wait time, the median wait time is returned.
 135    *
 136    * @param workerNodeKey - The worker node key.
 137    * @returns The worker task wait time.
 138    */
 139   protected getWorkerTaskWaitTime (workerNodeKey: number): number {
 140     return this.taskStatisticsRequirements.waitTime.median
 141       ? this.pool.workerNodes[workerNodeKey].usage.waitTime?.median ?? 0
 142       : this.pool.workerNodes[workerNodeKey].usage.waitTime?.average ?? 0
 143   }
 144
 145   /**
 146    * Gets the worker task ELU.
 147    * If the task statistics require the average ELU, the average ELU is returned.
 148    * If the task statistics require the median ELU, the median ELU is returned.
 149    *
 150    * @param workerNodeKey - The worker node key.
 151    * @returns The worker task ELU.
 152    */
 153   protected getWorkerTaskElu (workerNodeKey: number): number {
 154     return this.taskStatisticsRequirements.elu.median
 155       ? this.pool.workerNodes[workerNodeKey].usage.elu.active?.median ?? 0
 156       : this.pool.workerNodes[workerNodeKey].usage.elu.active?.average ?? 0
 157   }
 158
 159   protected computeDefaultWorkerWeight (): number {
 160     let cpusCycleTimeWeight = 0
 161     for (const cpu of cpus()) {
 162       // CPU estimated cycle time
 163       const numberOfDigits = cpu.speed.toString().length - 1
 164       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 165       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 166     }
 167     return Math.round(cpusCycleTimeWeight / cpus().length)
 168   }
 169 }