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