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