src/pools/selection-strategies/abstract-worker-choice-strategy.ts

   1 import { cpus } from 'node:os'
   2 import { DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS } from '../../utils'
   3 import type { IPool } from '../pool'
   4 import type { IWorker } from '../worker'
   5 import type {
   6   IWorkerChoiceStrategy,
   7   TaskStatisticsRequirements,
   8   WorkerChoiceStrategyOptions
   9 } from './selection-strategies-types'
  10
  11 /**
  12  * Worker choice strategy abstract base class.
  13  *
  14  * @typeParam Worker - Type of worker which manages the strategy.
  15  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  16  * @typeParam Response - Type of execution response. This can only be serializable data.
  17  */
  18 export abstract class AbstractWorkerChoiceStrategy<
  19   Worker extends IWorker,
  20   Data = unknown,
  21   Response = unknown
  22 > implements IWorkerChoiceStrategy {
  23   /**
  24    * Toggles finding the last free worker node key.
  25    */
  26   private toggleFindLastFreeWorkerNodeKey: boolean = false
  27   /** @inheritDoc */
  28   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  29     runTime: {
  30       aggregate: false,
  31       average: false,
  32       median: false
  33     },
  34     waitTime: {
  35       aggregate: false,
  36       average: false,
  37       median: false
  38     },
  39     elu: {
  40       aggregate: false,
  41       average: false,
  42       median: false
  43     }
  44   }
  45
  46   /**
  47    * Constructs a worker choice strategy bound to the pool.
  48    *
  49    * @param pool - The pool instance.
  50    * @param opts - The worker choice strategy options.
  51    */
  52   public constructor (
  53     protected readonly pool: IPool<Worker, Data, Response>,
  54     protected opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  55   ) {
  56     this.choose = this.choose.bind(this)
  57   }
  58
  59   protected setTaskStatisticsRequirements (
  60     opts: WorkerChoiceStrategyOptions
  61   ): void {
  62     if (
  63       this.taskStatisticsRequirements.runTime.average &&
  64       opts.runTime?.median === true
  65     ) {
  66       this.taskStatisticsRequirements.runTime.average = false
  67       this.taskStatisticsRequirements.runTime.median = opts.runTime
  68         .median as boolean
  69     }
  70     if (
  71       this.taskStatisticsRequirements.runTime.median &&
  72       opts.runTime?.median === false
  73     ) {
  74       this.taskStatisticsRequirements.runTime.average = true
  75       this.taskStatisticsRequirements.runTime.median = opts.runTime
  76         .median as boolean
  77     }
  78     if (
  79       this.taskStatisticsRequirements.waitTime.average &&
  80       opts.waitTime?.median === true
  81     ) {
  82       this.taskStatisticsRequirements.waitTime.average = false
  83       this.taskStatisticsRequirements.waitTime.median = opts.waitTime
  84         .median as boolean
  85     }
  86     if (
  87       this.taskStatisticsRequirements.waitTime.median &&
  88       opts.waitTime?.median === false
  89     ) {
  90       this.taskStatisticsRequirements.waitTime.average = true
  91       this.taskStatisticsRequirements.waitTime.median = opts.waitTime
  92         .median as boolean
  93     }
  94     if (
  95       this.taskStatisticsRequirements.elu.average &&
  96       opts.elu?.median === true
  97     ) {
  98       this.taskStatisticsRequirements.elu.average = false
  99       this.taskStatisticsRequirements.elu.median = opts.elu.median as boolean
 100     }
 101     if (
 102       this.taskStatisticsRequirements.elu.median &&
 103       opts.elu?.median === false
 104     ) {
 105       this.taskStatisticsRequirements.elu.average = true
 106       this.taskStatisticsRequirements.elu.median = opts.elu.median as boolean
 107     }
 108   }
 109
 110   /** @inheritDoc */
 111   public abstract reset (): boolean
 112
 113   /** @inheritDoc */
 114   public abstract update (workerNodeKey: number): boolean
 115
 116   /** @inheritDoc */
 117   public abstract choose (): number
 118
 119   /** @inheritDoc */
 120   public abstract remove (workerNodeKey: number): boolean
 121
 122   /** @inheritDoc */
 123   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 124     opts = opts ?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
 125     this.setTaskStatisticsRequirements(opts)
 126     this.opts = opts
 127   }
 128
 129   /**
 130    * Finds a free worker node key.
 131    *
 132    * @returns The free worker node key or `-1` if there is no free worker node.
 133    */
 134   protected findFreeWorkerNodeKey (): number {
 135     if (this.toggleFindLastFreeWorkerNodeKey) {
 136       this.toggleFindLastFreeWorkerNodeKey = false
 137       return this.findLastFreeWorkerNodeKey()
 138     }
 139     this.toggleFindLastFreeWorkerNodeKey = true
 140     return this.findFirstFreeWorkerNodeKey()
 141   }
 142
 143   /**
 144    * Gets the worker task runtime.
 145    * If the task statistics require the average runtime, the average runtime is returned.
 146    * If the task statistics require the median runtime , the median runtime is returned.
 147    *
 148    * @param workerNodeKey - The worker node key.
 149    * @returns The worker task runtime.
 150    */
 151   protected getWorkerTaskRunTime (workerNodeKey: number): number {
 152     return this.taskStatisticsRequirements.runTime.median
 153       ? this.pool.workerNodes[workerNodeKey].workerUsage.runTime.median
 154       : this.pool.workerNodes[workerNodeKey].workerUsage.runTime.average
 155   }
 156
 157   /**
 158    * Gets the worker task wait time.
 159    * If the task statistics require the average wait time, the average wait time is returned.
 160    * If the task statistics require the median wait time, the median wait time is returned.
 161    *
 162    * @param workerNodeKey - The worker node key.
 163    * @returns The worker task wait time.
 164    */
 165   protected getWorkerTaskWaitTime (workerNodeKey: number): number {
 166     return this.taskStatisticsRequirements.waitTime.median
 167       ? this.pool.workerNodes[workerNodeKey].workerUsage.runTime.median
 168       : this.pool.workerNodes[workerNodeKey].workerUsage.runTime.average
 169   }
 170
 171   /**
 172    * Gets the worker task ELU.
 173    * If the task statistics require the average ELU, the average ELU is returned.
 174    * If the task statistics require the median ELU, the median ELU is returned.
 175    *
 176    * @param workerNodeKey - The worker node key.
 177    * @returns The worker task ELU.
 178    */
 179   protected getWorkerTaskElu (workerNodeKey: number): number {
 180     return this.taskStatisticsRequirements.elu.median
 181       ? this.pool.workerNodes[workerNodeKey].workerUsage.elu.active.median
 182       : this.pool.workerNodes[workerNodeKey].workerUsage.elu.active.average
 183   }
 184
 185   protected computeDefaultWorkerWeight (): number {
 186     let cpusCycleTimeWeight = 0
 187     for (const cpu of cpus()) {
 188       // CPU estimated cycle time
 189       const numberOfDigits = cpu.speed.toString().length - 1
 190       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 191       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 192     }
 193     return Math.round(cpusCycleTimeWeight / cpus().length)
 194   }
 195
 196   /**
 197    * Finds the first free worker node key based on the number of tasks the worker has applied.
 198    *
 199    * If a worker is found with `0` executing tasks, it is detected as free and its worker node key is returned.
 200    *
 201    * If no free worker is found, `-1` is returned.
 202    *
 203    * @returns A worker node key if there is one, `-1` otherwise.
 204    */
 205   private findFirstFreeWorkerNodeKey (): number {
 206     return this.pool.workerNodes.findIndex(workerNode => {
 207       return workerNode.workerUsage.tasks.executing === 0
 208     })
 209   }
 210
 211   /**
 212    * Finds the last free worker node key based on the number of tasks the worker has applied.
 213    *
 214    * If a worker is found with `0` executing tasks, it is detected as free and its worker node key is returned.
 215    *
 216    * If no free worker is found, `-1` is returned.
 217    *
 218    * @returns A worker node key if there is one, `-1` otherwise.
 219    */
 220   private findLastFreeWorkerNodeKey (): number {
 221     // It requires node >= 18.0.0:
 222     // return this.workerNodes.findLastIndex(workerNode => {
 223     //   return workerNode.workerUsage.tasks.executing === 0
 224     // })
 225     for (
 226       let workerNodeKey = this.pool.workerNodes.length - 1;
 227       workerNodeKey >= 0;
 228       workerNodeKey--
 229     ) {
 230       if (
 231         this.pool.workerNodes[workerNodeKey].workerUsage.tasks.executing === 0
 232       ) {
 233         return workerNodeKey
 234       }
 235     }
 236     return -1
 237   }
 238 }