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