src/pools/selection-strategies/fair-share-worker-choice-strategy.ts

   1 import type { IPoolWorker } from '../pool-worker'
   2 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   3 import type {
   4   IWorkerChoiceStrategy,
   5   RequiredStatistics
   6 } from './selection-strategies-types'
   7
   8 /**
   9  * Worker virtual task timestamp.
  10  */
  11 interface WorkerVirtualTaskTimestamp {
  12   start: number
  13   end: number
  14 }
  15
  16 /**
  17  * Selects the next worker with a fair share scheduling algorithm.
  18  * Loosely modeled after the fair queueing algorithm: https://en.wikipedia.org/wiki/Fair_queuing.
  19  *
  20  * @typeParam Worker - Type of worker which manages the strategy.
  21  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
  22  * @typeParam Response - Type of response of execution. This can only be serializable data.
  23  */
  24 export class FairShareWorkerChoiceStrategy<
  25     Worker extends IPoolWorker,
  26     Data = unknown,
  27     Response = unknown
  28   >
  29   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  30   implements IWorkerChoiceStrategy {
  31   /** @inheritDoc */
  32   public readonly requiredStatistics: RequiredStatistics = {
  33     runTime: true,
  34     avgRunTime: true,
  35     medRunTime: false
  36   }
  37
  38   /**
  39    *  Worker last virtual task execution timestamp.
  40    */
  41   private readonly workerLastVirtualTaskTimestamp: Map<
  42   number,
  43   WorkerVirtualTaskTimestamp
  44   > = new Map<number, WorkerVirtualTaskTimestamp>()
  45
  46   /** @inheritDoc */
  47   public reset (): boolean {
  48     this.workerLastVirtualTaskTimestamp.clear()
  49     return true
  50   }
  51
  52   /** @inheritDoc */
  53   public choose (): number {
  54     let minWorkerVirtualTaskEndTimestamp = Infinity
  55     let chosenWorkerKey!: number
  56     for (const [index] of this.pool.workers.entries()) {
  57       this.computeWorkerLastVirtualTaskTimestamp(index)
  58       const workerLastVirtualTaskEndTimestamp =
  59         this.workerLastVirtualTaskTimestamp.get(index)?.end ?? 0
  60       if (
  61         workerLastVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  62       ) {
  63         minWorkerVirtualTaskEndTimestamp = workerLastVirtualTaskEndTimestamp
  64         chosenWorkerKey = index
  65       }
  66     }
  67     return chosenWorkerKey
  68   }
  69
  70   /** @inheritDoc */
  71   public remove (workerKey: number): boolean {
  72     const workerDeleted = this.workerLastVirtualTaskTimestamp.delete(workerKey)
  73     for (const [key, value] of this.workerLastVirtualTaskTimestamp.entries()) {
  74       if (key > workerKey) {
  75         this.workerLastVirtualTaskTimestamp.set(key - 1, value)
  76       }
  77     }
  78     return workerDeleted
  79   }
  80
  81   /**
  82    * Computes worker last virtual task timestamp.
  83    *
  84    * @param workerKey - The worker key.
  85    */
  86   private computeWorkerLastVirtualTaskTimestamp (workerKey: number): void {
  87     const workerVirtualTaskStartTimestamp = Math.max(
  88       Date.now(),
  89       this.workerLastVirtualTaskTimestamp.get(workerKey)?.end ?? -Infinity
  90     )
  91     this.workerLastVirtualTaskTimestamp.set(workerKey, {
  92       start: workerVirtualTaskStartTimestamp,
  93       end:
  94         workerVirtualTaskStartTimestamp +
  95         (this.pool.workers[workerKey].tasksUsage.avgRunTime ?? 0)
  96     })
  97   }
  98 }