X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fselection-strategies%2Fweighted-round-robin-worker-choice-strategy.ts;h=3709cce409b4ba988a728f81f0f8c6eb65f205f8;hb=9cd39dd47830f0923cd3ebf53b709bf7fb07e788;hp=6e48f82f2bc877bcf396e0631fb02d9f76dab9a0;hpb=fa6f12966fa14931864dcd13596e080d689464b3;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 6e48f82f..3709cce4 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,16 @@ -import { cpus } from 'os' -import type { AbstractPoolWorker } from '../abstract-pool-worker' +import { cpus } from 'node:os' import type { IPoolInternal } from '../pool-internal' +import type { IPoolWorker } from '../pool-worker' import { AbstractWorkerChoiceStrategy } from './abstract-worker-choice-strategy' -import type { RequiredStatistics } from './selection-strategies-types' +import type { + IWorkerChoiceStrategy, + RequiredStatistics +} from './selection-strategies-types' /** - * Task run time. + * Virtual task runtime. */ -type TaskRunTime = { +interface TaskRunTime { weight: number runTime: number } @@ -16,116 +19,136 @@ 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 AbstractPoolWorker, - Data, - Response -> extends AbstractWorkerChoiceStrategy { - /** @inheritDoc */ - public requiredStatistics: RequiredStatistics = { - runTime: true + Worker extends IPoolWorker, + Data, + Response + > + extends AbstractWorkerChoiceStrategy + implements IWorkerChoiceStrategy { + /** {@inheritDoc} */ + public readonly requiredStatistics: RequiredStatistics = { + runTime: true, + avgRunTime: true } /** - * Worker index where the previous task was submitted. + * Worker 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 currentWorkerId: number = 0 /** * Default worker weight. */ - private defaultWorkerWeight: number + private readonly defaultWorkerWeight: number /** - * Per worker task runtime map. + * Per worker virtual task runtime map. */ - private workerTaskRunTime: 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. */ public constructor (pool: IPoolInternal) { super(pool) this.defaultWorkerWeight = this.computeWorkerWeight() - this.initWorkerTaskRunTime() + this.initWorkersTaskRunTime() } - /** @inheritDoc */ - public choose (): Worker { - const currentWorker = this.pool.workers[this.currentWorkerIndex] - if (this.isDynamicPool === true) { - this.workerTaskRunTime.has(currentWorker) === false && - this.workerTaskRunTime.set(currentWorker, { - weight: this.defaultWorkerWeight, - runTime: 0 - }) + /** {@inheritDoc} */ + public reset (): boolean { + this.currentWorkerId = 0 + this.workersTaskRunTime.clear() + this.initWorkersTaskRunTime() + return true + } + + /** {@inheritDoc} */ + public choose (): number { + const chosenWorkerKey = this.currentWorkerId + if (this.isDynamicPool && !this.workersTaskRunTime.has(chosenWorkerKey)) { + this.initWorkerTaskRunTime(chosenWorkerKey) } - const workerVirtualTaskRunTime = - this.getWorkerVirtualTaskRunTime(currentWorker) ?? 0 + const workerTaskRunTime = + this.workersTaskRunTime.get(chosenWorkerKey)?.runTime ?? 0 const workerTaskWeight = - this.workerTaskRunTime.get(currentWorker)?.weight ?? + this.workersTaskRunTime.get(chosenWorkerKey)?.weight ?? this.defaultWorkerWeight - if (this.currentWorkerIndex === this.previousWorkerIndex) { - const workerTaskRunTime = - (this.workerTaskRunTime.get(currentWorker)?.runTime ?? 0) + - workerVirtualTaskRunTime - this.workerTaskRunTime.set(currentWorker, { - weight: workerTaskWeight, - runTime: workerTaskRunTime - }) - } else { - this.workerTaskRunTime.set(currentWorker, { - weight: workerTaskWeight, - runTime: 0 - }) - } - if ( - workerVirtualTaskRunTime < - (this.workerTaskRunTime.get(currentWorker) ?? this.defaultWorkerWeight) - ) { - this.previousWorkerIndex = this.currentWorkerIndex + if (workerTaskRunTime < workerTaskWeight) { + this.setWorkerTaskRunTime( + chosenWorkerKey, + workerTaskWeight, + workerTaskRunTime + + (this.getWorkerVirtualTaskRunTime(chosenWorkerKey) ?? 0) + ) } else { - this.previousWorkerIndex = this.currentWorkerIndex - this.currentWorkerIndex = - this.pool.workers.length - 1 === this.currentWorkerIndex + this.currentWorkerId = + this.currentWorkerId === this.pool.workers.length - 1 ? 0 - : this.currentWorkerIndex + 1 + : this.currentWorkerId + 1 + this.setWorkerTaskRunTime(this.currentWorkerId, workerTaskWeight, 0) } - return this.pool.workers[this.currentWorkerIndex] + return chosenWorkerKey } - private computeWorkerWeight () { - 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) + /** {@inheritDoc} */ + public remove (workerKey: number): boolean { + if (this.currentWorkerId === workerKey) { + this.currentWorkerId = + this.currentWorkerId > this.pool.workers.length - 1 + ? this.pool.workers.length - 1 + : this.currentWorkerId + } + const workerDeleted = this.workersTaskRunTime.delete(workerKey) + for (const [key, value] of this.workersTaskRunTime) { + if (key > workerKey) { + this.workersTaskRunTime.set(key - 1, value) + } } - return cpusCycleTimeWeight / cpus().length + return workerDeleted } - private initWorkerTaskRunTime () { - for (const worker of this.pool.workers) { - this.workerTaskRunTime.set(worker, { - weight: this.defaultWorkerWeight, - runTime: 0 - }) + private initWorkersTaskRunTime (): void { + for (const [index] of this.pool.workers.entries()) { + this.initWorkerTaskRunTime(index) } } - private getWorkerVirtualTaskRunTime (worker: Worker): number | undefined { - return this.pool.getWorkerAverageTasksRunTime(worker) + private initWorkerTaskRunTime (workerKey: number): void { + this.setWorkerTaskRunTime(workerKey, this.defaultWorkerWeight, 0) + } + + private setWorkerTaskRunTime ( + workerKey: number, + weight: number, + runTime: number + ): void { + this.workersTaskRunTime.set(workerKey, { + weight, + runTime + }) + } + + private getWorkerVirtualTaskRunTime (workerKey: number): number { + return this.pool.workers[workerKey].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) + } + return Math.round(cpusCycleTimeWeight / cpus().length) } }