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