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
   4 /**
   5  * Worker virtual task timestamp.
   6  */
   7 type WorkerVirtualTaskTimestamp = {
   8   start: number
   9   end: number
  10 }
  11
  12 /**
  13  * Selects the next worker with a fair share scheduling algorithm.
  14  * Loosely modeled after the fair queueing algorithm: https://en.wikipedia.org/wiki/Fair_queuing.
  15  *
  16  * @template Worker Type of worker which manages the strategy.
  17  * @template Data Type of data sent to the worker. This can only be serializable data.
  18  * @template Response Type of response of execution. This can only be serializable data.
  19  */
  20 export class FairShareWorkerChoiceStrategy<
  21   Worker extends AbstractPoolWorker,
  22   Data,
  23   Response
  24 > extends AbstractWorkerChoiceStrategy<Worker, Data, Response> {
  25   /**
  26    *  Worker last virtual task execution timestamp.
  27    */
  28   private workerLastVirtualTaskTimestamp: Map<
  29     Worker,
  30     WorkerVirtualTaskTimestamp
  31   > = new Map<Worker, WorkerVirtualTaskTimestamp>()
  32
  33   /** @inheritDoc */
  34   public choose (): Worker {
  35     this.updateWorkerLastVirtualTaskTimestamp()
  36     let minWorkerVirtualTaskEndTimestamp = Infinity
  37     let chosenWorker!: Worker
  38     for (const worker of this.pool.workers) {
  39       const workerLastVirtualTaskEndTimestamp =
  40         this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? 0
  41       if (
  42         workerLastVirtualTaskEndTimestamp < minWorkerVirtualTaskEndTimestamp
  43       ) {
  44         minWorkerVirtualTaskEndTimestamp = workerLastVirtualTaskEndTimestamp
  45         chosenWorker = worker
  46       }
  47     }
  48     return chosenWorker
  49   }
  50
  51   /**
  52    * Compute workers last virtual task timestamp.
  53    */
  54   private updateWorkerLastVirtualTaskTimestamp () {
  55     for (const worker of this.pool.workers) {
  56       const workerVirtualTaskStartTimestamp = Math.max(
  57         Date.now(),
  58         this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? 0
  59       )
  60       const workerVirtualTaskEndTimestamp =
  61         workerVirtualTaskStartTimestamp +
  62         (this.pool.getWorkerAverageTasksRunTime(worker) ?? 0)
  63       this.workerLastVirtualTaskTimestamp.set(worker, {
  64         start: workerVirtualTaskStartTimestamp,
  65         end: workerVirtualTaskEndTimestamp
  66       })
  67     }
  68   }
  69 }