src/pools/selection-strategies.ts

   1 import { isKillBehavior, KillBehaviors } from '../worker/worker-options'
   2 import type { IWorker } from './abstract-pool'
   3 import type { IPoolInternal } from './pool-internal'
   4
   5 /**
   6  * Enumeration of worker choice strategies.
   7  */
   8 export const WorkerChoiceStrategies = Object.freeze({
   9   /**
  10    * Round robin worker selection strategy.
  11    */
  12   ROUND_ROBIN: 'ROUND_ROBIN',
  13   /**
  14    * Less recently used worker selection strategy.
  15    */
  16   LESS_RECENTLY_USED: 'LESS_RECENTLY_USED'
  17 } as const)
  18
  19 /**
  20  * Worker choice strategy.
  21  */
  22 export type WorkerChoiceStrategy = keyof typeof WorkerChoiceStrategies
  23
  24 /**
  25  * Worker choice strategy interface.
  26  *
  27  * @template Worker Type of worker which manages the strategy.
  28  */
  29 interface IWorkerChoiceStrategy<Worker extends IWorker> {
  30   /**
  31    * Choose a worker in the pool.
  32    */
  33   choose(): Worker
  34 }
  35
  36 /**
  37  * Selects the next worker in a round robin fashion.
  38  *
  39  * @template Worker Type of worker which manages the strategy.
  40  * @template Data Type of data sent to the worker. This can only be serializable data.
  41  * @template Response Type of response of execution. This can only be serializable data.
  42  */
  43 class RoundRobinWorkerChoiceStrategy<Worker extends IWorker, Data, Response>
  44   implements IWorkerChoiceStrategy<Worker> {
  45   /**
  46    * Index for the next worker.
  47    */
  48   private nextWorkerIndex: number = 0
  49
  50   /**
  51    * Constructs a worker choice strategy that selects in a round robin fashion.
  52    *
  53    * @param pool The pool instance.
  54    */
  55   public constructor (
  56     private readonly pool: IPoolInternal<Worker, Data, Response>
  57   ) {}
  58
  59   /** @inheritdoc */
  60   public choose (): Worker {
  61     const chosenWorker = this.pool.workers[this.nextWorkerIndex]
  62     this.nextWorkerIndex =
  63       this.pool.workers.length - 1 === this.nextWorkerIndex
  64         ? 0
  65         : this.nextWorkerIndex + 1
  66     return chosenWorker
  67   }
  68 }
  69
  70 /**
  71  * Selects the less recently used worker.
  72  *
  73  * @template Worker Type of worker which manages the strategy.
  74  * @template Data Type of data sent to the worker. This can only be serializable data.
  75  * @template Response Type of response of execution. This can only be serializable data.
  76  */
  77 class LessRecentlyUsedWorkerChoiceStrategy<
  78   Worker extends IWorker,
  79   Data,
  80   Response
  81 > implements IWorkerChoiceStrategy<Worker> {
  82   /**
  83    * Constructs a worker choice strategy that selects based on less recently used.
  84    *
  85    * @param pool The pool instance.
  86    */
  87   public constructor (
  88     private readonly pool: IPoolInternal<Worker, Data, Response>
  89   ) {}
  90
  91   /** @inheritdoc */
  92   public choose (): Worker {
  93     const isPoolDynamic = this.pool.dynamic
  94     let minNumberOfTasks = Infinity
  95     // A worker is always found because it picks the one with fewer tasks
  96     let lessRecentlyUsedWorker!: Worker
  97     for (const [worker, numberOfTasks] of this.pool.tasks) {
  98       if (!isPoolDynamic && numberOfTasks === 0) {
  99         return worker
 100       } else if (numberOfTasks < minNumberOfTasks) {
 101         minNumberOfTasks = numberOfTasks
 102         lessRecentlyUsedWorker = worker
 103       }
 104     }
 105     return lessRecentlyUsedWorker
 106   }
 107 }
 108
 109 /**
 110  * Dynamically choose a worker.
 111  *
 112  * @template Worker Type of worker which manages the strategy.
 113  * @template Data Type of data sent to the worker. This can only be serializable data.
 114  * @template Response Type of response of execution. This can only be serializable data.
 115  */
 116 class DynamicPoolWorkerChoiceStrategy<Worker extends IWorker, Data, Response>
 117   implements IWorkerChoiceStrategy<Worker> {
 118   private workerChoiceStrategy: IWorkerChoiceStrategy<Worker>
 119
 120   /**
 121    * Constructs a worker choice strategy for dynamical pools.
 122    *
 123    * @param pool The pool instance.
 124    * @param workerChoiceStrategy The worker choice strategy when the pull is full.
 125    */
 126   public constructor (
 127     private readonly pool: IPoolInternal<Worker, Data, Response>,
 128     workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
 129   ) {
 130     this.workerChoiceStrategy = SelectionStrategiesUtils.getWorkerChoiceStrategy(
 131       this.pool,
 132       workerChoiceStrategy
 133     )
 134   }
 135
 136   /** @inheritdoc */
 137   public choose (): Worker {
 138     const freeWorker = SelectionStrategiesUtils.findFreeWorkerBasedOnTasks(
 139       this.pool
 140     )
 141     if (freeWorker) {
 142       return freeWorker
 143     }
 144
 145     if (this.pool.workers.length === this.pool.max) {
 146       this.pool.emitter.emit('FullPool')
 147       return this.workerChoiceStrategy.choose()
 148     }
 149
 150     // All workers are busy, create a new worker
 151     const workerCreated = this.pool.createAndSetupWorker()
 152     this.pool.registerWorkerMessageListener(workerCreated, message => {
 153       const tasksInProgress = this.pool.tasks.get(workerCreated)
 154       if (
 155         isKillBehavior(KillBehaviors.HARD, message.kill) ||
 156         tasksInProgress === 0
 157       ) {
 158         // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
 159         void this.pool.destroyWorker(workerCreated)
 160       }
 161     })
 162     return workerCreated
 163   }
 164 }
 165
 166 /**
 167  * The worker choice strategy context.
 168  *
 169  * @template Worker Type of worker.
 170  * @template Data Type of data sent to the worker. This can only be serializable data.
 171  * @template Response Type of response of execution. This can only be serializable data.
 172  */
 173 export class WorkerChoiceStrategyContext<
 174   Worker extends IWorker,
 175   Data,
 176   Response
 177 > {
 178   // Will be set by setter in constructor
 179   private workerChoiceStrategy!: IWorkerChoiceStrategy<Worker>
 180
 181   /**
 182    * Worker choice strategy context constructor.
 183    *
 184    * @param pool The pool instance.
 185    * @param workerChoiceStrategy The worker choice strategy.
 186    */
 187   public constructor (
 188     private readonly pool: IPoolInternal<Worker, Data, Response>,
 189     workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
 190   ) {
 191     this.setWorkerChoiceStrategy(workerChoiceStrategy)
 192   }
 193
 194   /**
 195    * Get the worker choice strategy instance specific to the pool type.
 196    *
 197    * @param workerChoiceStrategy The worker choice strategy.
 198    * @returns The worker choice strategy instance for the pool type.
 199    */
 200   private getPoolWorkerChoiceStrategy (
 201     workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
 202   ): IWorkerChoiceStrategy<Worker> {
 203     if (this.pool.dynamic) {
 204       return new DynamicPoolWorkerChoiceStrategy(
 205         this.pool,
 206         workerChoiceStrategy
 207       )
 208     }
 209     return SelectionStrategiesUtils.getWorkerChoiceStrategy(
 210       this.pool,
 211       workerChoiceStrategy
 212     )
 213   }
 214
 215   /**
 216    * Set the worker choice strategy to use in the context.
 217    *
 218    * @param workerChoiceStrategy The worker choice strategy to set.
 219    */
 220   public setWorkerChoiceStrategy (
 221     workerChoiceStrategy: WorkerChoiceStrategy
 222   ): void {
 223     this.workerChoiceStrategy = this.getPoolWorkerChoiceStrategy(
 224       workerChoiceStrategy
 225     )
 226   }
 227
 228   /**
 229    * Choose a worker with the underlying selection strategy.
 230    *
 231    * @returns The chosen one.
 232    */
 233   public execute (): Worker {
 234     return this.workerChoiceStrategy.choose()
 235   }
 236 }
 237
 238 /**
 239  * Worker selection strategies helpers.
 240  */
 241 class SelectionStrategiesUtils {
 242   /**
 243    * Find a free worker based on number of tasks the worker has applied.
 244    *
 245    * If a worker was found that has `0` tasks, it is detected as free and will be returned.
 246    *
 247    * If no free worker was found, `null` will be returned.
 248    *
 249    * @param pool The pool instance.
 250    * @returns A free worker if there was one, otherwise `null`.
 251    */
 252   public static findFreeWorkerBasedOnTasks<
 253     Worker extends IWorker,
 254     Data,
 255     Response
 256   > (pool: IPoolInternal<Worker, Data, Response>): Worker | null {
 257     for (const [worker, numberOfTasks] of pool.tasks) {
 258       if (numberOfTasks === 0) {
 259         // A worker is free, use it
 260         return worker
 261       }
 262     }
 263     return null
 264   }
 265
 266   /**
 267    * Get the worker choice strategy instance.
 268    *
 269    * @param pool The pool instance.
 270    * @param workerChoiceStrategy The worker choice strategy.
 271    * @returns The worker choice strategy instance.
 272    */
 273   public static getWorkerChoiceStrategy<
 274     Worker extends IWorker,
 275     Data,
 276     Response
 277   > (
 278     pool: IPoolInternal<Worker, Data, Response>,
 279     workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN
 280   ): IWorkerChoiceStrategy<Worker> {
 281     switch (workerChoiceStrategy) {
 282       case WorkerChoiceStrategies.ROUND_ROBIN:
 283         return new RoundRobinWorkerChoiceStrategy(pool)
 284       case WorkerChoiceStrategies.LESS_RECENTLY_USED:
 285         return new LessRecentlyUsedWorkerChoiceStrategy(pool)
 286       default:
 287         throw new Error(
 288           `Worker choice strategy '${workerChoiceStrategy}' not found`
 289         )
 290     }
 291   }
 292 }