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,
  27     Response
  28   >
  29   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  30   implements IWorkerChoiceStrategy {
  31   /** {@inheritDoc} */
  32   public readonly requiredStatistics: RequiredStatistics = {
  33     runTime: true
  34   }
  35
  36   /**
  37    *  Worker last virtual task execution timestamp.
  38    */
  39   private readonly workerLastVirtualTaskTimestamp: Map<
  40   number,
  41   WorkerVirtualTaskTimestamp
  42   > = new Map<number, WorkerVirtualTaskTimestamp>()
  43
  44   /** {@inheritDoc} */
  45   public reset (): boolean {
  46     this.workerLastVirtualTaskTimestamp.clear()
  47     return true
  48   }
  49
  50   /** {@inheritDoc} */
  51   public choose (): number {
  52     let minWorkerVirtualTaskEndTimestamp = Infinity
  53     let chosenWorkerKey!: number
  54     for (const [index] of this.pool.workers.entries()) {
  55       this.computeWorkerLastVirtualTaskTimestamp(index)
  56       const workerLastVirtualTaskEndTimestamp =
  57         this.workerLastVirtualTaskTimestamp.get(index)?.end ?? 0
  58       if (
  59         workerLastVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  60       ) {
  61         minWorkerVirtualTaskEndTimestamp = workerLastVirtualTaskEndTimestamp
  62         chosenWorkerKey = index
  63       }
  64     }
  65     return chosenWorkerKey
  66   }
  67
  68   /** {@inheritDoc} */
  69   public remove (workerKey: number): boolean {
  70     const workerDeleted = this.workerLastVirtualTaskTimestamp.delete(workerKey)
  71     for (const [key, value] of this.workerLastVirtualTaskTimestamp.entries()) {
  72       if (key > workerKey) {
  73         this.workerLastVirtualTaskTimestamp.set(key - 1, value)
  74       }
  75     }
  76     return workerDeleted
  77   }
  78
  79   /**
  80    * Computes worker last virtual task timestamp.
  81    *
  82    * @param workerKey - The worker key.
  83    */
  84   private computeWorkerLastVirtualTaskTimestamp (workerKey: number): void {
  85     const workerVirtualTaskStartTimestamp = Math.max(
  86       Date.now(),
  87       this.workerLastVirtualTaskTimestamp.get(workerKey)?.end ?? -Infinity
  88     )
  89     this.workerLastVirtualTaskTimestamp.set(workerKey, {
  90       start: workerVirtualTaskStartTimestamp,
  91       end:
  92         workerVirtualTaskStartTimestamp +
  93         (this.pool.workers[workerKey].tasksUsage.avgRunTime ?? 0)
  94     })
  95   }
  96 }