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

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