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   Worker,
  36   WorkerVirtualTaskTimestamp
  37   > = new Map<Worker, WorkerVirtualTaskTimestamp>()
  38
  39   /** {@inheritDoc} */
  40   public reset (): boolean {
  41     this.workerLastVirtualTaskTimestamp.clear()
  42     return true
  43   }
  44
  45   /** {@inheritDoc} */
  46   public choose (): Worker {
  47     let minWorkerVirtualTaskEndTimestamp = Infinity
  48     let chosenWorker!: Worker
  49     for (const value of this.pool.workers.values()) {
  50       const worker = value.worker
  51       this.computeWorkerLastVirtualTaskTimestamp(worker)
  52       const workerLastVirtualTaskEndTimestamp =
  53         this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? 0
  54       if (
  55         workerLastVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  56       ) {
  57         minWorkerVirtualTaskEndTimestamp = workerLastVirtualTaskEndTimestamp
  58         chosenWorker = worker
  59       }
  60     }
  61     return chosenWorker
  62   }
  63
  64   /**
  65    * Computes worker last virtual task timestamp.
  66    *
  67    * @param worker - The worker.
  68    */
  69   private computeWorkerLastVirtualTaskTimestamp (worker: Worker): void {
  70     const workerVirtualTaskStartTimestamp = Math.max(
  71       Date.now(),
  72       this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? -Infinity
  73     )
  74     this.workerLastVirtualTaskTimestamp.set(worker, {
  75       start: workerVirtualTaskStartTimestamp,
  76       end:
  77         workerVirtualTaskStartTimestamp +
  78         (this.pool.getWorkerAverageTasksRunTime(worker) ?? 0)
  79     })
  80   }
  81 }