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

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