X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fselection-strategies%2Fweighted-round-robin-worker-choice-strategy.ts;h=6bbbbaf64738527cafb2720ae9a5ba73d7544c0e;hb=04d875a3942abe97ce35c177c6e6f41619be974b;hp=f4bac5d76c4e3b669bb5720b99c03b07eabc577d;hpb=c4855468bc26a7ee37d2c8ef34bb1ac864448e77;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 f4bac5d7..6bbbbaf6 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,28 +1,20 @@ -import { cpus } from 'node:os' -import type { IWorker } from '../worker' -import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy' +import type { IWorker } from '../worker.js' +import type { IPool } from '../pool.js' +import { DEFAULT_MEASUREMENT_STATISTICS_REQUIREMENTS } from '../../utils.js' +import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy.js' import type { IWorkerChoiceStrategy, - RequiredStatistics, + TaskStatisticsRequirements, WorkerChoiceStrategyOptions -} from './selection-strategies-types' -import type { IPool } from '../pool' - -/** - * Virtual task runtime. - */ -interface TaskRunTime { - weight: number - runTime: number -} +} from './selection-strategies-types.js' /** * 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 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 WeightedRoundRobinWorkerChoiceStrategy< Worker extends IWorker, @@ -32,138 +24,87 @@ export class WeightedRoundRobinWorkerChoiceStrategy< extends AbstractWorkerChoiceStrategy implements IWorkerChoiceStrategy { /** @inheritDoc */ - public readonly requiredStatistics: RequiredStatistics = { - runTime: true, - avgRunTime: true, - medRunTime: false + 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. + * Worker node virtual task runtime. */ - private currentWorkerNodeId: number = 0 - /** - * Default worker weight. - */ - private readonly defaultWorkerWeight: number - /** - * Workers' virtual task runtime. - */ - private readonly workersTaskRunTime: Map = new Map< - number, - TaskRunTime - >() + private workerNodeVirtualTaskRunTime: 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: IPool, opts?: WorkerChoiceStrategyOptions ) { super(pool, opts) - this.defaultWorkerWeight = this.computeWorkerWeight() - this.initWorkersTaskRunTime() + this.setTaskStatisticsRequirements(this.opts) } /** @inheritDoc */ public reset (): boolean { - this.currentWorkerNodeId = 0 - this.workersTaskRunTime.clear() - this.initWorkersTaskRunTime() + this.resetWorkerNodeKeyProperties() + this.workerNodeVirtualTaskRunTime = 0 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) - ) - } else { - this.currentWorkerNodeId = - this.currentWorkerNodeId === this.pool.workerNodes.length - 1 - ? 0 - : this.currentWorkerNodeId + 1 - this.setWorkerTaskRunTime(this.currentWorkerNodeId, workerTaskWeight, 0) - } - return chosenWorkerNodeKey + public update (): boolean { + return true + } + + /** @inheritDoc */ + public choose (): number | undefined { + this.setPreviousWorkerNodeKey(this.nextWorkerNodeKey) + this.weightedRoundRobinNextWorkerNodeKey() + this.checkNextWorkerNodeReadiness() + return this.nextWorkerNodeKey } /** @inheritDoc */ public remove (workerNodeKey: number): boolean { - 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 - } + if (this.pool.workerNodes.length === 0) { + this.reset() } - const deleted = this.workersTaskRunTime.delete(workerNodeKey) - for (const [key, value] of this.workersTaskRunTime) { - if (key > workerNodeKey) { - this.workersTaskRunTime.set(key - 1, value) + if (this.nextWorkerNodeKey === workerNodeKey) { + this.workerNodeVirtualTaskRunTime = 0 + if (this.nextWorkerNodeKey > this.pool.workerNodes.length - 1) { + this.nextWorkerNodeKey = this.pool.workerNodes.length - 1 } } - return deleted - } - - private initWorkersTaskRunTime (): void { - for (const [index] of this.pool.workerNodes.entries()) { - this.initWorkerTaskRunTime(index) + if ( + this.previousWorkerNodeKey === workerNodeKey && + this.previousWorkerNodeKey > this.pool.workerNodes.length - 1 + ) { + this.previousWorkerNodeKey = this.pool.workerNodes.length - 1 } + return true } - 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 - } - - private computeWorkerWeight (): 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) + private weightedRoundRobinNextWorkerNodeKey (): number | undefined { + const workerWeight = + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.opts!.weights![this.nextWorkerNodeKey ?? this.previousWorkerNodeKey]! + if (this.workerNodeVirtualTaskRunTime < workerWeight) { + this.workerNodeVirtualTaskRunTime = + this.workerNodeVirtualTaskRunTime + + this.getWorkerNodeTaskRunTime( + this.nextWorkerNodeKey ?? this.previousWorkerNodeKey + ) + } else { + this.nextWorkerNodeKey = + this.nextWorkerNodeKey === this.pool.workerNodes.length - 1 + ? 0 + : (this.nextWorkerNodeKey ?? this.previousWorkerNodeKey) + 1 + this.workerNodeVirtualTaskRunTime = 0 } - return Math.round(cpusCycleTimeWeight / cpus().length) + return this.nextWorkerNodeKey } }