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   StrategyPolicy,
  11   TaskStatisticsRequirements,
  12   WorkerChoiceStrategyOptions
  13 } from './selection-strategies-types'
  14
  15 /**
  16  * Worker choice strategy abstract base class.
  17  *
  18  * @typeParam Worker - Type of worker which manages the strategy.
  19  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  20  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  21  */
  22 export abstract class AbstractWorkerChoiceStrategy<
  23   Worker extends IWorker,
  24   Data = unknown,
  25   Response = unknown
  26 > implements IWorkerChoiceStrategy {
  27   /**
  28    * The next worker node key.
  29    */
  30   protected nextWorkerNodeKey: number = 0
  31
  32   /** @inheritDoc */
  33   public readonly strategyPolicy: StrategyPolicy = {
  34     useDynamicWorker: false
  35   }
  36
  37   /** @inheritDoc */
  38   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  39     runTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  40     waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
  41     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  42   }
  43
  44   /**
  45    * Constructs a worker choice strategy bound to the pool.
  46    *
  47    * @param pool - The pool instance.
  48    * @param opts - The worker choice strategy options.
  49    */
  50   public constructor (
  51     protected readonly pool: IPool<Worker, Data, Response>,
  52     protected opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  53   ) {
  54     this.choose = this.choose.bind(this)
  55   }
  56
  57   protected setTaskStatisticsRequirements (
  58     opts: WorkerChoiceStrategyOptions
  59   ): void {
  60     if (
  61       this.taskStatisticsRequirements.runTime.average &&
  62       opts.runTime?.median === true
  63     ) {
  64       this.taskStatisticsRequirements.runTime.average = false
  65       this.taskStatisticsRequirements.runTime.median = opts.runTime
  66         .median as boolean
  67     }
  68     if (
  69       this.taskStatisticsRequirements.runTime.median &&
  70       opts.runTime?.median === false
  71     ) {
  72       this.taskStatisticsRequirements.runTime.average = true
  73       this.taskStatisticsRequirements.runTime.median = opts.runTime
  74         .median as boolean
  75     }
  76     if (
  77       this.taskStatisticsRequirements.waitTime.average &&
  78       opts.waitTime?.median === true
  79     ) {
  80       this.taskStatisticsRequirements.waitTime.average = false
  81       this.taskStatisticsRequirements.waitTime.median = opts.waitTime
  82         .median as boolean
  83     }
  84     if (
  85       this.taskStatisticsRequirements.waitTime.median &&
  86       opts.waitTime?.median === false
  87     ) {
  88       this.taskStatisticsRequirements.waitTime.average = true
  89       this.taskStatisticsRequirements.waitTime.median = opts.waitTime
  90         .median as boolean
  91     }
  92     if (
  93       this.taskStatisticsRequirements.elu.average &&
  94       opts.elu?.median === true
  95     ) {
  96       this.taskStatisticsRequirements.elu.average = false
  97       this.taskStatisticsRequirements.elu.median = opts.elu.median as boolean
  98     }
  99     if (
 100       this.taskStatisticsRequirements.elu.median &&
 101       opts.elu?.median === false
 102     ) {
 103       this.taskStatisticsRequirements.elu.average = true
 104       this.taskStatisticsRequirements.elu.median = opts.elu.median as boolean
 105     }
 106   }
 107
 108   /** @inheritDoc */
 109   public abstract reset (): boolean
 110
 111   /** @inheritDoc */
 112   public abstract update (workerNodeKey: number): boolean
 113
 114   /** @inheritDoc */
 115   public abstract choose (): number
 116
 117   /** @inheritDoc */
 118   public abstract remove (workerNodeKey: number): boolean
 119
 120   /** @inheritDoc */
 121   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 122     this.opts = opts ?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
 123     this.setTaskStatisticsRequirements(this.opts)
 124   }
 125
 126   protected isWorkerNodeReady (workerNodeKey: number): boolean {
 127     return this.pool.workerNodes[workerNodeKey].info.ready
 128   }
 129
 130   /**
 131    * Gets the worker task runtime.
 132    * If the task statistics require the average runtime, the average runtime is returned.
 133    * If the task statistics require the median runtime , the median runtime is returned.
 134    *
 135    * @param workerNodeKey - The worker node key.
 136    * @returns The worker task runtime.
 137    */
 138   protected getWorkerTaskRunTime (workerNodeKey: number): number {
 139     return this.taskStatisticsRequirements.runTime.median
 140       ? this.pool.workerNodes[workerNodeKey].usage.runTime?.median ?? 0
 141       : this.pool.workerNodes[workerNodeKey].usage.runTime?.average ?? 0
 142   }
 143
 144   /**
 145    * Gets the worker task wait time.
 146    * If the task statistics require the average wait time, the average wait time is returned.
 147    * If the task statistics require the median wait time, the median wait time is returned.
 148    *
 149    * @param workerNodeKey - The worker node key.
 150    * @returns The worker task wait time.
 151    */
 152   protected getWorkerTaskWaitTime (workerNodeKey: number): number {
 153     return this.taskStatisticsRequirements.waitTime.median
 154       ? this.pool.workerNodes[workerNodeKey].usage.waitTime?.median ?? 0
 155       : this.pool.workerNodes[workerNodeKey].usage.waitTime?.average ?? 0
 156   }
 157
 158   /**
 159    * Gets the worker task ELU.
 160    * If the task statistics require the average ELU, the average ELU is returned.
 161    * If the task statistics require the median ELU, the median ELU is returned.
 162    *
 163    * @param workerNodeKey - The worker node key.
 164    * @returns The worker task ELU.
 165    */
 166   protected getWorkerTaskElu (workerNodeKey: number): number {
 167     return this.taskStatisticsRequirements.elu.median
 168       ? this.pool.workerNodes[workerNodeKey].usage.elu.active?.median ?? 0
 169       : this.pool.workerNodes[workerNodeKey].usage.elu.active?.average ?? 0
 170   }
 171
 172   protected computeDefaultWorkerWeight (): number {
 173     let cpusCycleTimeWeight = 0
 174     for (const cpu of cpus()) {
 175       // CPU estimated cycle time
 176       const numberOfDigits = cpu.speed.toString().length - 1
 177       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 178       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 179     }
 180     return Math.round(cpusCycleTimeWeight / cpus().length)
 181   }
 182 }