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

   1 import type { AbstractPoolWorker } from '../abstract-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 type 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  * @template Worker Type of worker which manages the strategy.
  18  * @template Data Type of data sent to the worker. This can only be serializable data.
  19  * @template Response Type of response of execution. This can only be serializable data.
  20  */
  21 export class FairShareWorkerChoiceStrategy<
  22   Worker extends AbstractPoolWorker,
  23   Data,
  24   Response
  25 > extends AbstractWorkerChoiceStrategy<Worker, Data, Response> {
  26   /** @inheritDoc */
  27   public requiredStatistics: RequiredStatistics = {
  28     runTime: true
  29   }
  30
  31   /**
  32    *  Worker last virtual task execution timestamp.
  33    */
  34   private workerLastVirtualTaskTimestamp: Map<
  35     Worker,
  36     WorkerVirtualTaskTimestamp
  37   > = new Map<Worker, WorkerVirtualTaskTimestamp>()
  38
  39   /** @inheritDoc */
  40   public choose (): Worker {
  41     this.updateWorkerLastVirtualTaskTimestamp()
  42     let minWorkerVirtualTaskEndTimestamp = Infinity
  43     let chosenWorker!: Worker
  44     for (const worker of this.pool.workers) {
  45       const workerLastVirtualTaskEndTimestamp =
  46         this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? 0
  47       if (
  48         workerLastVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  49       ) {
  50         minWorkerVirtualTaskEndTimestamp = workerLastVirtualTaskEndTimestamp
  51         chosenWorker = worker
  52       }
  53     }
  54     return chosenWorker
  55   }
  56
  57   /**
  58    * Compute workers last virtual task timestamp.
  59    */
  60   private updateWorkerLastVirtualTaskTimestamp () {
  61     for (const worker of this.pool.workers) {
  62       const workerVirtualTaskStartTimestamp = Math.max(
  63         Date.now(),
  64         this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? 0
  65       )
  66       const workerVirtualTaskEndTimestamp =
  67         workerVirtualTaskStartTimestamp +
  68         (this.pool.getWorkerAverageTasksRunTime(worker) ?? 0)
  69       this.workerLastVirtualTaskTimestamp.set(worker, {
  70         start: workerVirtualTaskStartTimestamp,
  71         end: workerVirtualTaskEndTimestamp
  72       })
  73     }
  74   }
  75 }