X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fpools%2Fselection-strategies%2Fweighted-round-robin-worker-choice-strategy.ts;h=f4bac5d76c4e3b669bb5720b99c03b07eabc577d;hb=c4855468bc26a7ee37d2c8ef34bb1ac864448e77;hp=c2f85145c44dfd4d3eee48fba3fd9d6c64db94d0;hpb=777af0ac351dd4137b6daa612146533085f9fe2e;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 c2f85145..f4bac5d7 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,13 +1,17 @@ -import { cpus } from 'os' -import type { IPoolInternal } from '../pool-internal' -import type { IPoolWorker } from '../pool-worker' +import { cpus } from 'node:os' +import type { IWorker } from '../worker' import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy' -import type { RequiredStatistics } from './selection-strategies-types' +import type { + IWorkerChoiceStrategy, + RequiredStatistics, + WorkerChoiceStrategyOptions +} from './selection-strategies-types' +import type { IPool } from '../pool' /** * Virtual task runtime. */ -type TaskRunTime = { +interface TaskRunTime { weight: number runTime: number } @@ -16,130 +20,149 @@ type TaskRunTime = { * 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. * - * @template Worker Type of worker which manages the strategy. - * @template Data Type of data sent to the worker. This can only be serializable data. - * @template Response Type of response of execution. This can only be serializable data. + * @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. */ export class WeightedRoundRobinWorkerChoiceStrategy< - Worker extends IPoolWorker, - Data, - Response -> extends AbstractWorkerChoiceStrategy { + Worker extends IWorker, + Data = unknown, + Response = unknown + > + extends AbstractWorkerChoiceStrategy + implements IWorkerChoiceStrategy { /** @inheritDoc */ public readonly requiredStatistics: RequiredStatistics = { - runTime: true + runTime: true, + avgRunTime: true, + medRunTime: false } /** - * Worker index where the previous task was submitted. + * Worker node id where the current task will be submitted. */ - private previousWorkerIndex: number = 0 - /** - * Worker index where the current task will be submitted. - */ - private currentWorkerIndex: number = 0 + private currentWorkerNodeId: number = 0 /** * Default worker weight. */ private readonly defaultWorkerWeight: number /** - * Per worker virtual task runtime map. + * Workers' virtual task runtime. */ - private readonly workersTaskRunTime: Map = new Map< - Worker, - TaskRunTime + private readonly workersTaskRunTime: Map = new Map< + number, + TaskRunTime >() /** * Constructs a worker choice strategy that selects with a weighted round robin scheduling algorithm. * - * @param pool The pool instance. + * @param pool - The pool instance. + * @param opts - The worker choice strategy options. */ - public constructor (pool: IPoolInternal) { - super(pool) + public constructor ( + pool: IPool, + opts?: WorkerChoiceStrategyOptions + ) { + super(pool, opts) this.defaultWorkerWeight = this.computeWorkerWeight() this.initWorkersTaskRunTime() } /** @inheritDoc */ public reset (): boolean { - this.previousWorkerIndex = 0 - this.currentWorkerIndex = 0 + this.currentWorkerNodeId = 0 this.workersTaskRunTime.clear() this.initWorkersTaskRunTime() return true } /** @inheritDoc */ - public choose (): Worker { - const currentWorker = this.pool.workers[this.currentWorkerIndex] + public choose (): number { + const chosenWorkerNodeKey = this.currentWorkerNodeId if ( - this.isDynamicPool === true && - this.workersTaskRunTime.has(currentWorker) === false + this.isDynamicPool && + !this.workersTaskRunTime.has(chosenWorkerNodeKey) ) { - this.initWorkerTaskRunTime(currentWorker) + this.initWorkerTaskRunTime(chosenWorkerNodeKey) } - const workerVirtualTaskRunTime = - this.getWorkerVirtualTaskRunTime(currentWorker) ?? 0 + const workerTaskRunTime = + this.workersTaskRunTime.get(chosenWorkerNodeKey)?.runTime ?? 0 const workerTaskWeight = - this.workersTaskRunTime.get(currentWorker)?.weight ?? + this.workersTaskRunTime.get(chosenWorkerNodeKey)?.weight ?? this.defaultWorkerWeight - if (this.currentWorkerIndex === this.previousWorkerIndex) { - const workerTaskRunTime = - (this.workersTaskRunTime.get(currentWorker)?.runTime ?? 0) + - workerVirtualTaskRunTime + if (workerTaskRunTime < workerTaskWeight) { this.setWorkerTaskRunTime( - currentWorker, + chosenWorkerNodeKey, workerTaskWeight, - workerTaskRunTime + workerTaskRunTime + + (this.getWorkerVirtualTaskRunTime(chosenWorkerNodeKey) ?? 0) ) } else { - this.setWorkerTaskRunTime(currentWorker, workerTaskWeight, 0) - } - if (workerVirtualTaskRunTime < workerTaskWeight) { - this.previousWorkerIndex = this.currentWorkerIndex - } else { - this.previousWorkerIndex = this.currentWorkerIndex - this.currentWorkerIndex = - this.pool.workers.length - 1 === this.currentWorkerIndex + this.currentWorkerNodeId = + this.currentWorkerNodeId === this.pool.workerNodes.length - 1 ? 0 - : this.currentWorkerIndex + 1 + : this.currentWorkerNodeId + 1 + this.setWorkerTaskRunTime(this.currentWorkerNodeId, workerTaskWeight, 0) + } + return chosenWorkerNodeKey + } + + /** @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 + } + } + const deleted = this.workersTaskRunTime.delete(workerNodeKey) + for (const [key, value] of this.workersTaskRunTime) { + if (key > workerNodeKey) { + this.workersTaskRunTime.set(key - 1, value) + } } - return this.pool.workers[this.currentWorkerIndex] + return deleted } private initWorkersTaskRunTime (): void { - for (const worker of this.pool.workers) { - this.initWorkerTaskRunTime(worker) + for (const [index] of this.pool.workerNodes.entries()) { + this.initWorkerTaskRunTime(index) } } - private initWorkerTaskRunTime (worker: Worker): void { - this.setWorkerTaskRunTime(worker, this.defaultWorkerWeight, 0) + private initWorkerTaskRunTime (workerNodeKey: number): void { + this.setWorkerTaskRunTime(workerNodeKey, this.defaultWorkerWeight, 0) } private setWorkerTaskRunTime ( - worker: Worker, + workerNodeKey: number, weight: number, runTime: number ): void { - this.workersTaskRunTime.set(worker, { + this.workersTaskRunTime.set(workerNodeKey, { weight, runTime }) } - private getWorkerVirtualTaskRunTime (worker: Worker): number | undefined { - return this.pool.getWorkerAverageTasksRunTime(worker) + 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 numberOfDigit = cpu.speed.toString().length - 1 - const cpuCycleTime = 1 / (cpu.speed / Math.pow(10, numberOfDigit)) - cpusCycleTimeWeight += cpuCycleTime * Math.pow(10, numberOfDigit) + 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) }