X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fselection-strategies%2Fweighted-round-robin-worker-choice-strategy.ts;h=a0553fcb81ace9c910b1ecd755b9c123b640f362;hb=f6b641d651fdd5114bcd63bf1c991955d5b7d3eb;hp=c2782e8b2bf844b1a767eafcfe8554b0fc568ec7;hpb=da3098610d6cf6155bbbe53c84b6ac6ccd400ff5;p=poolifier.git diff --git a/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts b/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts index c2782e8b..a0553fcb 100644 --- a/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts @@ -1,6 +1,7 @@ import { cpus } from 'node:os' -import type { IPoolInternal } from '../pool-internal' 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 { IWorkerChoiceStrategy, @@ -8,21 +9,13 @@ import type { WorkerChoiceStrategyOptions } from './selection-strategies-types' -/** - * Virtual task runtime. - */ -interface TaskRunTime { - weight: number - runTime: number -} - /** * Selects the next worker with a weighted round robin scheduling algorithm. * Loosely modeled after the weighted round robin queueing algorithm: https://en.wikipedia.org/wiki/Weighted_round_robin. * * @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 response of execution. This can only be serializable data. + * @typeParam Response - Type of execution response. This can only be serializable data. */ export class WeightedRoundRobinWorkerChoiceStrategy< Worker extends IWorker, @@ -47,63 +40,48 @@ export class WeightedRoundRobinWorkerChoiceStrategy< */ private readonly defaultWorkerWeight: number /** - * Workers' virtual task runtime. + * Worker virtual task runtime. */ - private readonly workersTaskRunTime: Map = new Map< - number, - TaskRunTime - >() + private workerVirtualTaskRunTime: number = 0 - /** - * Constructs a worker choice strategy that selects with a weighted round robin scheduling algorithm. - * - * @param pool - The pool instance. - * @param opts - The worker choice strategy options. - */ + /** @inheritDoc */ public constructor ( - pool: IPoolInternal, - opts?: WorkerChoiceStrategyOptions + pool: IPool, + opts: WorkerChoiceStrategyOptions = DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS ) { super(pool, opts) - this.defaultWorkerWeight = this.computeWorkerWeight() - this.initWorkersTaskRunTime() + this.setRequiredStatistics(this.opts) + this.defaultWorkerWeight = this.computeDefaultWorkerWeight() } /** @inheritDoc */ public reset (): boolean { this.currentWorkerNodeId = 0 - this.workersTaskRunTime.clear() - this.initWorkersTaskRunTime() + this.workerVirtualTaskRunTime = 0 + return true + } + + /** @inheritDoc */ + public update (): boolean { return true } /** @inheritDoc */ public choose (): number { const chosenWorkerNodeKey = this.currentWorkerNodeId - if ( - this.isDynamicPool && - !this.workersTaskRunTime.has(chosenWorkerNodeKey) - ) { - this.initWorkerTaskRunTime(chosenWorkerNodeKey) - } - const workerTaskRunTime = - this.workersTaskRunTime.get(chosenWorkerNodeKey)?.runTime ?? 0 - const workerTaskWeight = - this.workersTaskRunTime.get(chosenWorkerNodeKey)?.weight ?? - this.defaultWorkerWeight - if (workerTaskRunTime < workerTaskWeight) { - this.setWorkerTaskRunTime( - chosenWorkerNodeKey, - workerTaskWeight, - workerTaskRunTime + - (this.getWorkerVirtualTaskRunTime(chosenWorkerNodeKey) ?? 0) - ) + const workerVirtualTaskRunTime = this.workerVirtualTaskRunTime + const workerWeight = + this.opts.weights?.[chosenWorkerNodeKey] ?? this.defaultWorkerWeight + if (workerVirtualTaskRunTime < workerWeight) { + this.workerVirtualTaskRunTime = + workerVirtualTaskRunTime + + this.getWorkerTaskRunTime(chosenWorkerNodeKey) } else { this.currentWorkerNodeId = this.currentWorkerNodeId === this.pool.workerNodes.length - 1 ? 0 : this.currentWorkerNodeId + 1 - this.setWorkerTaskRunTime(this.currentWorkerNodeId, workerTaskWeight, 0) + this.workerVirtualTaskRunTime = 0 } return chosenWorkerNodeKey } @@ -113,50 +91,15 @@ export class WeightedRoundRobinWorkerChoiceStrategy< if (this.currentWorkerNodeId === workerNodeKey) { if (this.pool.workerNodes.length === 0) { this.currentWorkerNodeId = 0 - } else { - this.currentWorkerNodeId = - this.currentWorkerNodeId > this.pool.workerNodes.length - 1 - ? this.pool.workerNodes.length - 1 - : this.currentWorkerNodeId + } else if (this.currentWorkerNodeId > this.pool.workerNodes.length - 1) { + this.currentWorkerNodeId = this.pool.workerNodes.length - 1 } + this.workerVirtualTaskRunTime = 0 } - const deleted = this.workersTaskRunTime.delete(workerNodeKey) - for (const [key, value] of this.workersTaskRunTime) { - if (key > workerNodeKey) { - this.workersTaskRunTime.set(key - 1, value) - } - } - return deleted - } - - private initWorkersTaskRunTime (): void { - for (const [index] of this.pool.workerNodes.entries()) { - this.initWorkerTaskRunTime(index) - } - } - - private initWorkerTaskRunTime (workerNodeKey: number): void { - this.setWorkerTaskRunTime(workerNodeKey, this.defaultWorkerWeight, 0) - } - - private setWorkerTaskRunTime ( - workerNodeKey: number, - weight: number, - runTime: number - ): void { - this.workersTaskRunTime.set(workerNodeKey, { - weight, - runTime - }) - } - - private getWorkerVirtualTaskRunTime (workerNodeKey: number): number { - return this.requiredStatistics.medRunTime - ? this.pool.workerNodes[workerNodeKey].tasksUsage.medRunTime - : this.pool.workerNodes[workerNodeKey].tasksUsage.avgRunTime + return true } - private computeWorkerWeight (): number { + private computeDefaultWorkerWeight (): number { let cpusCycleTimeWeight = 0 for (const cpu of cpus()) { // CPU estimated cycle time