src/pools/selection-strategies/least-busy-worker-choice-strategy.ts

   1 import {
   2   DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS,
   3   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
   4 } from '../../utils'
   5 import type { IPool } from '../pool'
   6 import type { IWorker } from '../worker'
   7 import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy'
   8 import type {
   9   IWorkerChoiceStrategy,
  10   TaskStatisticsRequirements,
  11   WorkerChoiceStrategyOptions
  12 } from './selection-strategies-types'
  13
  14 /**
  15  * Selects the least busy worker.
  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 structured-cloneable data.
  19  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  20  */
  21 export class LeastBusyWorkerChoiceStrategy<
  22     Worker extends IWorker,
  23     Data = unknown,
  24     Response = unknown
  25   >
  26   extends AbstractWorkerChoiceStrategy<Worker, Data, Response>
  27   implements IWorkerChoiceStrategy {
  28   /** @inheritDoc */
  29   public readonly taskStatisticsRequirements: TaskStatisticsRequirements = {
  30     runTime: {
  31       aggregate: true,
  32       average: false,
  33       median: false
  34     },
  35     waitTime: {
  36       aggregate: true,
  37       average: false,
  38       median: false
  39     },
  40     elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS
  41   }
  42
  43   /** @inheritDoc */
  44   public constructor (
  45     pool: IPool<Worker, Data, Response>,
  46     opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS
  47   ) {
  48     super(pool, opts)
  49     this.setTaskStatisticsRequirements(this.opts)
  50   }
  51
  52   /** @inheritDoc */
  53   public reset (): boolean {
  54     return true
  55   }
  56
  57   /** @inheritDoc */
  58   public update (): boolean {
  59     return true
  60   }
  61
  62   /** @inheritDoc */
  63   public choose (): number | undefined {
  64     this.nextWorkerNodeKey = this.leastBusyNextWorkerNodeKey()
  65     return this.nextWorkerNodeKey
  66   }
  67
  68   /** @inheritDoc */
  69   public remove (): boolean {
  70     return true
  71   }
  72
  73   private leastBusyNextWorkerNodeKey (): number | undefined {
  74     let minTime = Infinity
  75     let chosenWorkerNodeKey: number | undefined
  76     for (const [workerNodeKey, workerNode] of this.pool.workerNodes.entries()) {
  77       const workerTime =
  78         (workerNode.usage.runTime?.aggregate ?? 0) +
  79         (workerNode.usage.waitTime?.aggregate ?? 0)
  80       if (workerTime === 0) {
  81         chosenWorkerNodeKey = workerNodeKey
  82         break
  83       } else if (workerTime < minTime) {
  84         minTime = workerTime
  85         chosenWorkerNodeKey = workerNodeKey
  86       }
  87     }
  88     return chosenWorkerNodeKey
  89   }
  90 }