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