src/pools/selection-strategies/worker-choice-strategies-context.ts

   1 import type { IPool } from '../pool.js'
   2 import type { IWorker } from '../worker.js'
   3 import type {
   4   IWorkerChoiceStrategy,
   5   StrategyPolicy,
   6   TaskStatisticsRequirements,
   7   WorkerChoiceStrategy,
   8   WorkerChoiceStrategyOptions
   9 } from './selection-strategies-types.js'
  10 import { WorkerChoiceStrategies } from './selection-strategies-types.js'
  11 import {
  12   getWorkerChoiceStrategiesRetries,
  13   getWorkerChoiceStrategy
  14 } from './selection-strategies-utils.js'
  15
  16 /**
  17  * The worker choice strategies context.
  18  *
  19  * @typeParam Worker - Type of worker.
  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 class WorkerChoiceStrategiesContext<
  24   Worker extends IWorker,
  25   Data = unknown,
  26   Response = unknown
  27 > {
  28   /**
  29    * The number of worker choice strategies execution retries.
  30    */
  31   public retriesCount: number
  32
  33   /**
  34    * The default worker choice strategy in the context.
  35    */
  36   private defaultWorkerChoiceStrategy: WorkerChoiceStrategy
  37
  38   /**
  39    * The worker choice strategies registered in the context.
  40    */
  41   private readonly workerChoiceStrategies: Map<
  42   WorkerChoiceStrategy,
  43   IWorkerChoiceStrategy
  44   >
  45
  46   /**
  47    * The maximum number of worker choice strategies execution retries.
  48    */
  49   private readonly retries: number
  50
  51   /**
  52    * Worker choice strategies context constructor.
  53    *
  54    * @param pool - The pool instance.
  55    * @param workerChoiceStrategies - The worker choice strategies. @defaultValue [WorkerChoiceStrategies.ROUND_ROBIN]
  56    * @param opts - The worker choice strategy options.
  57    */
  58   public constructor (
  59     private readonly pool: IPool<Worker, Data, Response>,
  60     workerChoiceStrategies: WorkerChoiceStrategy[] = [
  61       WorkerChoiceStrategies.ROUND_ROBIN
  62     ],
  63     opts?: WorkerChoiceStrategyOptions
  64   ) {
  65     this.execute = this.execute.bind(this)
  66     this.defaultWorkerChoiceStrategy = workerChoiceStrategies[0]
  67     this.workerChoiceStrategies = new Map<
  68     WorkerChoiceStrategy,
  69     IWorkerChoiceStrategy
  70     >()
  71     for (const workerChoiceStrategy of workerChoiceStrategies) {
  72       this.addWorkerChoiceStrategy(workerChoiceStrategy, this.pool, opts)
  73     }
  74     this.retriesCount = 0
  75     this.retries = getWorkerChoiceStrategiesRetries(this.pool, opts)
  76   }
  77
  78   /**
  79    * Gets the active worker choice strategies policy in the context.
  80    *
  81    * @returns The strategies policy.
  82    */
  83   public getPolicy (): StrategyPolicy {
  84     const policies: StrategyPolicy[] = []
  85     for (const workerChoiceStrategy of this.workerChoiceStrategies.values()) {
  86       policies.push(workerChoiceStrategy.strategyPolicy)
  87     }
  88     return {
  89       dynamicWorkerUsage: policies.some(p => p.dynamicWorkerUsage),
  90       dynamicWorkerReady: policies.some(p => p.dynamicWorkerReady)
  91     }
  92   }
  93
  94   /**
  95    * Gets the active worker choice strategies in the context task statistics requirements.
  96    *
  97    * @returns The task statistics requirements.
  98    */
  99   public getTaskStatisticsRequirements (): TaskStatisticsRequirements {
 100     const taskStatisticsRequirements: TaskStatisticsRequirements[] = []
 101     for (const workerChoiceStrategy of this.workerChoiceStrategies.values()) {
 102       taskStatisticsRequirements.push(
 103         workerChoiceStrategy.taskStatisticsRequirements
 104       )
 105     }
 106     return {
 107       runTime: {
 108         aggregate: taskStatisticsRequirements.some(r => r.runTime.aggregate),
 109         average: taskStatisticsRequirements.some(r => r.runTime.average),
 110         median: taskStatisticsRequirements.some(r => r.runTime.median)
 111       },
 112       waitTime: {
 113         aggregate: taskStatisticsRequirements.some(r => r.waitTime.aggregate),
 114         average: taskStatisticsRequirements.some(r => r.waitTime.average),
 115         median: taskStatisticsRequirements.some(r => r.waitTime.median)
 116       },
 117       elu: {
 118         aggregate: taskStatisticsRequirements.some(r => r.elu.aggregate),
 119         average: taskStatisticsRequirements.some(r => r.elu.average),
 120         median: taskStatisticsRequirements.some(r => r.elu.median)
 121       }
 122     }
 123   }
 124
 125   /**
 126    * Sets the default worker choice strategy to use in the context.
 127    *
 128    * @param workerChoiceStrategy - The default worker choice strategy to set.
 129    * @param opts - The worker choice strategy options.
 130    */
 131   public setDefaultWorkerChoiceStrategy (
 132     workerChoiceStrategy: WorkerChoiceStrategy,
 133     opts?: WorkerChoiceStrategyOptions
 134   ): void {
 135     this.defaultWorkerChoiceStrategy = workerChoiceStrategy
 136     this.addWorkerChoiceStrategy(workerChoiceStrategy, this.pool, opts)
 137   }
 138
 139   /**
 140    * Updates the worker node key in the active worker choice strategies in the context internals.
 141    *
 142    * @returns `true` if the update is successful, `false` otherwise.
 143    */
 144   public update (workerNodeKey: number): boolean {
 145     const res: boolean[] = []
 146     for (const workerChoiceStrategy of this.workerChoiceStrategies.values()) {
 147       res.push(workerChoiceStrategy.update(workerNodeKey))
 148     }
 149     return res.every(r => r)
 150   }
 151
 152   /**
 153    * Executes the worker choice strategy in the context algorithm.
 154    *
 155    * @param workerChoiceStrategy - The worker choice strategy algorithm to execute. @defaultValue this.defaultWorkerChoiceStrategy
 156    * @returns The key of the worker node.
 157    * @throws {@link https://nodejs.org/api/errors.html#class-error} If after computed retries the worker node key is null or undefined.
 158    */
 159   public execute (
 160     workerChoiceStrategy: WorkerChoiceStrategy = this
 161       .defaultWorkerChoiceStrategy
 162   ): number {
 163     return this.executeStrategy(
 164       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 165       this.workerChoiceStrategies.get(workerChoiceStrategy)!
 166     )
 167   }
 168
 169   /**
 170    * Executes the given worker choice strategy.
 171    *
 172    * @param workerChoiceStrategy - The worker choice strategy.
 173    * @returns The key of the worker node.
 174    * @throws {@link https://nodejs.org/api/errors.html#class-error} If after computed retries the worker node key is null or undefined.
 175    */
 176   private executeStrategy (workerChoiceStrategy: IWorkerChoiceStrategy): number {
 177     let workerNodeKey: number | undefined
 178     let chooseCount = 0
 179     let retriesCount = 0
 180     do {
 181       workerNodeKey = workerChoiceStrategy.choose()
 182       if (workerNodeKey == null && chooseCount > 0) {
 183         ++retriesCount
 184         ++this.retriesCount
 185       }
 186       ++chooseCount
 187     } while (workerNodeKey == null && retriesCount < this.retries)
 188     if (workerNodeKey == null) {
 189       throw new Error(
 190         `Worker node key chosen is null or undefined after ${retriesCount} retries`
 191       )
 192     }
 193     return workerNodeKey
 194   }
 195
 196   /**
 197    * Removes the worker node key from the active worker choice strategies in the context.
 198    *
 199    * @param workerNodeKey - The worker node key.
 200    * @returns `true` if the removal is successful, `false` otherwise.
 201    */
 202   public remove (workerNodeKey: number): boolean {
 203     const res: boolean[] = []
 204     for (const workerChoiceStrategy of this.workerChoiceStrategies.values()) {
 205       res.push(workerChoiceStrategy.remove(workerNodeKey))
 206     }
 207     return res.every(r => r)
 208   }
 209
 210   /**
 211    * Sets the active worker choice strategies in the context options.
 212    *
 213    * @param opts - The worker choice strategy options.
 214    */
 215   public setOptions (opts: WorkerChoiceStrategyOptions | undefined): void {
 216     for (const workerChoiceStrategy of this.workerChoiceStrategies.values()) {
 217       workerChoiceStrategy.setOptions(opts)
 218     }
 219   }
 220
 221   private addWorkerChoiceStrategy (
 222     workerChoiceStrategy: WorkerChoiceStrategy,
 223     pool: IPool<Worker, Data, Response>,
 224     opts?: WorkerChoiceStrategyOptions
 225   ): Map<WorkerChoiceStrategy, IWorkerChoiceStrategy> {
 226     if (!this.workerChoiceStrategies.has(workerChoiceStrategy)) {
 227       return this.workerChoiceStrategies.set(
 228         workerChoiceStrategy,
 229         getWorkerChoiceStrategy<Worker, Data, Response>(
 230           workerChoiceStrategy,
 231           pool,
 232           this,
 233           opts
 234         )
 235       )
 236     }
 237     return this.workerChoiceStrategies
 238   }
 239
 240   // private removeWorkerChoiceStrategy (
 241   //   workerChoiceStrategy: WorkerChoiceStrategy
 242   // ): boolean {
 243   //   return this.workerChoiceStrategies.delete(workerChoiceStrategy)
 244   // }
 245 }