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