X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fselection-strategies%2Finterleaved-weighted-round-robin-worker-choice-strategy.ts;h=b1abe98772d88ce295b17aa6be653a332dd11b04;hb=b358c8aca5934f7d5dc4bd234d1909ba13850ea9;hp=12a8055ecdd8691bc8cc29cef4c23a1804d66fda;hpb=5a5fc090d6f7eb9248df1ba5c0ff4d001461b6d4;p=poolifier.git diff --git a/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts b/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts index 12a8055e..b1abe987 100644 --- a/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts @@ -1,19 +1,19 @@ -import { cpus } from 'node:os' -import type { IWorker } from '../worker' -import type { IPool } from '../pool' -import { DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS } from '../../utils' -import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy' +import type { IPool } from '../pool.js' +import { DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS } from '../utils.js' +import type { IWorker } from '../worker.js' +import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy.js' import type { IWorkerChoiceStrategy, + TaskStatisticsRequirements, WorkerChoiceStrategyOptions -} from './selection-strategies-types' +} from './selection-strategies-types.js' /** * Selects the next worker with an interleaved weighted round robin scheduling algorithm. * * @typeParam Worker - Type of worker which manages the strategy. - * @typeParam Data - Type of data sent to the worker. This can only be serializable data. - * @typeParam Response - Type of execution response. This can only be serializable data. + * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. + * @typeParam Response - Type of execution response. This can only be structured-cloneable data. */ export class InterleavedWeightedRoundRobinWorkerChoiceStrategy< Worker extends IWorker, @@ -22,39 +22,50 @@ export class InterleavedWeightedRoundRobinWorkerChoiceStrategy< > extends AbstractWorkerChoiceStrategy implements IWorkerChoiceStrategy { + /** @inheritDoc */ + public readonly taskStatisticsRequirements: TaskStatisticsRequirements = { + runTime: { + aggregate: true, + average: true, + median: false + }, + waitTime: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS, + elu: DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS + } + /** - * Worker node id where the current task will be submitted. - */ - private currentWorkerNodeId: number = 0 - /** - * Current round id. - * This is used to determine the current round weight. + * Round id. */ - private currentRoundId: number = 0 + private roundId = 0 /** * Round weights. */ private roundWeights: number[] /** - * Default worker weight. + * Worker node id. */ - private readonly defaultWorkerWeight: number + private workerNodeId = 0 + /** + * Worker node virtual task runtime. + */ + private workerNodeVirtualTaskRunTime = 0 /** @inheritDoc */ public constructor ( pool: IPool, - opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS + opts?: WorkerChoiceStrategyOptions ) { super(pool, opts) - this.setRequiredStatistics(this.opts) - this.defaultWorkerWeight = this.computeDefaultWorkerWeight() + this.setTaskStatisticsRequirements(this.opts) this.roundWeights = this.getRoundWeights() } /** @inheritDoc */ public reset (): boolean { - this.currentWorkerNodeId = 0 - this.currentRoundId = 0 + this.resetWorkerNodeKeyProperties() + this.roundId = 0 + this.workerNodeId = 0 + this.workerNodeVirtualTaskRunTime = 0 return true } @@ -64,83 +75,95 @@ export class InterleavedWeightedRoundRobinWorkerChoiceStrategy< } /** @inheritDoc */ - public choose (): number { - let roundId: number | undefined - let workerNodeId: number | undefined + public choose (): number | undefined { for ( - let roundIndex = this.currentRoundId; + let roundIndex = this.roundId; roundIndex < this.roundWeights.length; roundIndex++ ) { + this.roundId = roundIndex for ( - let workerNodeKey = this.currentWorkerNodeId; + let workerNodeKey = this.workerNodeId; workerNodeKey < this.pool.workerNodes.length; workerNodeKey++ ) { - const workerWeight = - this.opts.weights?.[workerNodeKey] ?? this.defaultWorkerWeight - if (workerWeight >= this.roundWeights[roundIndex]) { - roundId = roundIndex - workerNodeId = workerNodeKey - break + this.workerNodeId = workerNodeKey + if ( + this.workerNodeId !== this.nextWorkerNodeKey && + this.workerNodeVirtualTaskRunTime !== 0 + ) { + this.workerNodeVirtualTaskRunTime = 0 + } + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const workerWeight = this.opts!.weights![workerNodeKey] + if ( + this.isWorkerNodeReady(workerNodeKey) && + workerWeight >= this.roundWeights[roundIndex] && + this.workerNodeVirtualTaskRunTime < workerWeight + ) { + this.workerNodeVirtualTaskRunTime = + this.workerNodeVirtualTaskRunTime + + this.getWorkerNodeTaskRunTime(workerNodeKey) + this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey) + this.nextWorkerNodeKey = workerNodeKey + return this.nextWorkerNodeKey } } } - this.currentRoundId = roundId ?? 0 - this.currentWorkerNodeId = workerNodeId ?? 0 - const chosenWorkerNodeKey = this.currentWorkerNodeId - if (this.currentWorkerNodeId === this.pool.workerNodes.length - 1) { - this.currentWorkerNodeId = 0 - this.currentRoundId = - this.currentRoundId === this.roundWeights.length - 1 - ? 0 - : this.currentRoundId + 1 + this.interleavedWeightedRoundRobinNextWorkerNodeId() + } + + private interleavedWeightedRoundRobinNextWorkerNodeId (): void { + if (this.pool.workerNodes.length === 0) { + this.workerNodeId = 0 + } else if ( + this.roundId === this.roundWeights.length - 1 && + this.workerNodeId === this.pool.workerNodes.length - 1 + ) { + this.roundId = 0 + this.workerNodeId = 0 + } else if (this.workerNodeId === this.pool.workerNodes.length - 1) { + this.roundId = this.roundId + 1 + this.workerNodeId = 0 } else { - this.currentWorkerNodeId = this.currentWorkerNodeId + 1 + this.workerNodeId = this.workerNodeId + 1 } - return chosenWorkerNodeKey } /** @inheritDoc */ public remove (workerNodeKey: number): boolean { - if (this.currentWorkerNodeId === workerNodeKey) { - if (this.pool.workerNodes.length === 0) { - this.currentWorkerNodeId = 0 - } else if (this.currentWorkerNodeId > this.pool.workerNodes.length - 1) { - this.currentWorkerNodeId = this.pool.workerNodes.length - 1 - this.currentRoundId = - this.currentRoundId === this.roundWeights.length - 1 - ? 0 - : this.currentRoundId + 1 - } + if (this.pool.workerNodes.length === 0) { + this.resetWorkerNodeKeyProperties() + this.workerNodeId = 0 + this.workerNodeVirtualTaskRunTime = 0 + return true + } + if ( + this.workerNodeId === workerNodeKey && + this.workerNodeId > this.pool.workerNodes.length - 1 + ) { + this.workerNodeId = this.pool.workerNodes.length - 1 + } + if ( + this.previousWorkerNodeKey === workerNodeKey && + this.previousWorkerNodeKey > this.pool.workerNodes.length - 1 + ) { + this.previousWorkerNodeKey = this.pool.workerNodes.length - 1 } return true } /** @inheritDoc */ - public setOptions (opts: WorkerChoiceStrategyOptions): void { + public setOptions (opts: WorkerChoiceStrategyOptions | undefined): void { super.setOptions(opts) this.roundWeights = this.getRoundWeights() } - private computeDefaultWorkerWeight (): number { - let cpusCycleTimeWeight = 0 - for (const cpu of cpus()) { - // CPU estimated cycle time - const numberOfDigits = cpu.speed.toString().length - 1 - const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigits)) - cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigits) - } - return Math.round(cpusCycleTimeWeight / cpus().length) - } - private getRoundWeights (): number[] { - if (this.opts.weights == null) { - return [this.defaultWorkerWeight] - } return [ ...new Set( - Object.values(this.opts.weights) + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + Object.values(this.opts!.weights!) .slice() .sort((a, b) => a - b) )