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: false
  40   }
  41
  42   /**
  43    * Constructs a worker choice strategy bound to the pool.
  44    *
  45    * @param pool - The pool instance.
  46    * @param opts - The worker choice strategy options.
  47    */
  48   public constructor (
  49     protected readonly pool: IPool<Worker, Data, Response>,
  50     protected opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  51   ) {
  52     this.choose = this.choose.bind(this)
  53   }
  54
  55   protected setTaskStatisticsRequirements (
  56     opts: WorkerChoiceStrategyOptions
  57   ): void {
  58     if (
  59       this.taskStatisticsRequirements.runTime.average &&
  60       opts.runTime?.median === true
  61     ) {
  62       this.taskStatisticsRequirements.runTime.average = false
  63       this.taskStatisticsRequirements.runTime.median = opts.runTime
  64         .median as boolean
  65     }
  66     if (
  67       this.taskStatisticsRequirements.runTime.median &&
  68       opts.runTime?.median === false
  69     ) {
  70       this.taskStatisticsRequirements.runTime.average = true
  71       this.taskStatisticsRequirements.runTime.median = opts.runTime
  72         .median as boolean
  73     }
  74     if (
  75       this.taskStatisticsRequirements.waitTime.average &&
  76       opts.waitTime?.median === true
  77     ) {
  78       this.taskStatisticsRequirements.waitTime.average = false
  79       this.taskStatisticsRequirements.waitTime.median = opts.waitTime
  80         .median as boolean
  81     }
  82     if (
  83       this.taskStatisticsRequirements.waitTime.median &&
  84       opts.waitTime?.median === false
  85     ) {
  86       this.taskStatisticsRequirements.waitTime.average = true
  87       this.taskStatisticsRequirements.waitTime.median = opts.waitTime
  88         .median as boolean
  89     }
  90   }
  91
  92   /** @inheritDoc */
  93   public abstract reset (): boolean
  94
  95   /** @inheritDoc */
  96   public abstract update (workerNodeKey: number): boolean
  97
  98   /** @inheritDoc */
  99   public abstract choose (): number
 100
 101   /** @inheritDoc */
 102   public abstract remove (workerNodeKey: number): boolean
 103
 104   /** @inheritDoc */
 105   public setOptions (opts: WorkerChoiceStrategyOptions): void {
 106     opts = opts ?? DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
 107     this.setTaskStatisticsRequirements(opts)
 108     this.opts = opts
 109   }
 110
 111   /**
 112    * Finds a free worker node key.
 113    *
 114    * @returns The free worker node key or `-1` if there is no free worker node.
 115    */
 116   protected findFreeWorkerNodeKey (): number {
 117     if (this.toggleFindLastFreeWorkerNodeKey) {
 118       this.toggleFindLastFreeWorkerNodeKey = false
 119       return this.findLastFreeWorkerNodeKey()
 120     }
 121     this.toggleFindLastFreeWorkerNodeKey = true
 122     return this.findFirstFreeWorkerNodeKey()
 123   }
 124
 125   /**
 126    * Gets the worker task runtime.
 127    * If the task statistics require the average runtime, the average runtime is returned.
 128    * If the task statistics require the median runtime , the median runtime is returned.
 129    *
 130    * @param workerNodeKey - The worker node key.
 131    * @returns The worker task runtime.
 132    */
 133   protected getWorkerTaskRunTime (workerNodeKey: number): number {
 134     return this.taskStatisticsRequirements.runTime.median
 135       ? this.pool.workerNodes[workerNodeKey].workerUsage.runTime.median
 136       : this.pool.workerNodes[workerNodeKey].workerUsage.runTime.average
 137   }
 138
 139   /**
 140    * Gets the worker task wait time.
 141    * If the task statistics require the average wait time, the average wait time is returned.
 142    * If the task statistics require the median wait time, the median wait time is returned.
 143    *
 144    * @param workerNodeKey - The worker node key.
 145    * @returns The worker task wait time.
 146    */
 147   protected getWorkerWaitTime (workerNodeKey: number): number {
 148     return this.taskStatisticsRequirements.waitTime.median
 149       ? this.pool.workerNodes[workerNodeKey].workerUsage.runTime.median
 150       : this.pool.workerNodes[workerNodeKey].workerUsage.runTime.average
 151   }
 152
 153   protected computeDefaultWorkerWeight (): number {
 154     let cpusCycleTimeWeight = 0
 155     for (const cpu of cpus()) {
 156       // CPU estimated cycle time
 157       const numberOfDigits = cpu.speed.toString().length - 1
 158       const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits))
 159       cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits)
 160     }
 161     return Math.round(cpusCycleTimeWeight / cpus().length)
 162   }
 163
 164   /**
 165    * Finds the first free worker node key based on the number of tasks the worker has applied.
 166    *
 167    * If a worker is found with `0` executing tasks, it is detected as free and its worker node key is returned.
 168    *
 169    * If no free worker is found, `-1` is returned.
 170    *
 171    * @returns A worker node key if there is one, `-1` otherwise.
 172    */
 173   private findFirstFreeWorkerNodeKey (): number {
 174     return this.pool.workerNodes.findIndex(workerNode => {
 175       return workerNode.workerUsage.tasks.executing === 0
 176     })
 177   }
 178
 179   /**
 180    * Finds the last free worker node key based on the number of tasks the worker has applied.
 181    *
 182    * If a worker is found with `0` executing tasks, it is detected as free and its worker node key is returned.
 183    *
 184    * If no free worker is found, `-1` is returned.
 185    *
 186    * @returns A worker node key if there is one, `-1` otherwise.
 187    */
 188   private findLastFreeWorkerNodeKey (): number {
 189     // It requires node >= 18.0.0:
 190     // return this.workerNodes.findLastIndex(workerNode => {
 191     //   return workerNode.workerUsage.tasks.executing === 0
 192     // })
 193     for (
 194       let workerNodeKey = this.pool.workerNodes.length - 1;
 195       workerNodeKey >= 0;
 196       workerNodeKey--
 197     ) {
 198       if (
 199         this.pool.workerNodes[workerNodeKey].workerUsage.tasks.executing === 0
 200       ) {
 201         return workerNodeKey
 202       }
 203     }
 204     return -1
 205   }
 206 }