From 3032893add6cc97da7b0600e21df2865ad1f675c Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 2 Apr 2023 17:06:57 +0200 Subject: [PATCH] refactor: remove unneeded encapsulation around tasks usage handling MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- src/pools/abstract-pool.ts | 126 +++--------------- src/pools/pool-internal.ts | 22 +-- .../fair-share-worker-choice-strategy.ts | 2 +- ...ss-recently-used-worker-choice-strategy.ts | 4 +- ...hted-round-robin-worker-choice-strategy.ts | 2 +- tests/pools/abstract/abstract-pool.test.js | 68 +--------- 6 files changed, 33 insertions(+), 191 deletions(-) diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index b3ebc194..2c4559a3 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -100,7 +100,7 @@ export abstract class AbstractPool< this.registerWorkerMessageListener(workerCreated, message => { if ( isKillBehavior(KillBehaviors.HARD, message.kill) || - this.getWorkerRunningTasks(workerCreated) === 0 + this.getWorkerTasksUsage(workerCreated)?.running === 0 ) { // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime) void this.destroyWorker(workerCreated) @@ -163,21 +163,6 @@ export abstract class AbstractPool< return [...this.workers].find(([, value]) => value.worker === worker)?.[0] } - /** {@inheritDoc} */ - public getWorkerRunningTasks (worker: Worker): number | undefined { - return this.getWorkerTasksUsage(worker)?.running - } - - /** {@inheritDoc} */ - public getWorkerRunTasks (worker: Worker): number | undefined { - return this.getWorkerTasksUsage(worker)?.run - } - - /** {@inheritDoc} */ - public getWorkerAverageTasksRunTime (worker: Worker): number | undefined { - return this.getWorkerTasksUsage(worker)?.avgRunTime - } - /** {@inheritDoc} */ public setWorkerChoiceStrategy ( workerChoiceStrategy: WorkerChoiceStrategy @@ -268,7 +253,7 @@ export abstract class AbstractPool< * @param worker - The worker. */ protected beforePromiseWorkerResponseHook (worker: Worker): void { - this.increaseWorkerRunningTasks(worker) + ++(this.getWorkerTasksUsage(worker) as TasksUsage).running } /** @@ -282,9 +267,21 @@ export abstract class AbstractPool< message: MessageValue, promise: PromiseWorkerResponseWrapper ): void { - this.decreaseWorkerRunningTasks(promise.worker) - this.stepWorkerRunTasks(promise.worker, 1) - this.updateWorkerTasksRunTime(promise.worker, message.taskRunTime) + const workerTasksUsage = this.getWorkerTasksUsage( + promise.worker + ) as TasksUsage + --workerTasksUsage.running + ++workerTasksUsage.run + if ( + this.workerChoiceStrategyContext.getWorkerChoiceStrategy() + .requiredStatistics.runTime + ) { + workerTasksUsage.runTime += message.taskRunTime ?? 0 + if (workerTasksUsage.run !== 0) { + workerTasksUsage.avgRunTime = + workerTasksUsage.runTime / workerTasksUsage.run + } + } } /** @@ -410,80 +407,14 @@ export abstract class AbstractPool< } } - /** - * Increases the number of tasks that the given worker has applied. - * - * @param worker - Worker which running tasks is increased. - */ - private increaseWorkerRunningTasks (worker: Worker): void { - this.stepWorkerRunningTasks(worker, 1) - } - - /** - * Decreases the number of tasks that the given worker has applied. - * - * @param worker - Worker which running tasks is decreased. - */ - private decreaseWorkerRunningTasks (worker: Worker): void { - this.stepWorkerRunningTasks(worker, -1) - } - - /** - * Gets tasks usage of the given worker. - * - * @param worker - Worker which tasks usage is returned. - */ - private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined { - if (this.checkWorker(worker)) { - const workerKey = this.getWorkerKey(worker) as number + /** {@inheritDoc} */ + public getWorkerTasksUsage (worker: Worker): TasksUsage | undefined { + const workerKey = this.getWorkerKey(worker) + if (workerKey !== undefined) { const workerEntry = this.workers.get(workerKey) as WorkerType return workerEntry.tasksUsage } - } - - /** - * Steps the number of tasks that the given worker has applied. - * - * @param worker - Worker which running tasks are stepped. - * @param step - Number of running tasks step. - */ - private stepWorkerRunningTasks (worker: Worker, step: number): void { - // prettier-ignore - (this.getWorkerTasksUsage(worker) as TasksUsage).running += step - } - - /** - * Steps the number of tasks that the given worker has run. - * - * @param worker - Worker which has run tasks. - * @param step - Number of run tasks step. - */ - private stepWorkerRunTasks (worker: Worker, step: number): void { - // prettier-ignore - (this.getWorkerTasksUsage(worker) as TasksUsage).run += step - } - - /** - * Updates tasks runtime for the given worker. - * - * @param worker - Worker which run the task. - * @param taskRunTime - Worker task runtime. - */ - private updateWorkerTasksRunTime ( - worker: Worker, - taskRunTime: number | undefined - ): void { - if ( - this.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .requiredStatistics.runTime - ) { - const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage - workerTasksUsage.runTime += taskRunTime ?? 0 - if (workerTasksUsage.run !== 0) { - workerTasksUsage.avgRunTime = - workerTasksUsage.runTime / workerTasksUsage.run - } - } + throw new Error('Worker could not be found in the pool') } /** @@ -503,17 +434,4 @@ export abstract class AbstractPool< tasksUsage }) } - - /** - * Checks if the given worker is registered in the pool. - * - * @param worker - Worker to check. - * @returns `true` if the worker is registered in the pool. - */ - private checkWorker (worker: Worker): boolean { - if (this.getWorkerKey(worker) == null) { - throw new Error('Worker could not be found in the pool') - } - return true - } } diff --git a/src/pools/pool-internal.ts b/src/pools/pool-internal.ts index 9656175f..9da64952 100644 --- a/src/pools/pool-internal.ts +++ b/src/pools/pool-internal.ts @@ -77,26 +77,10 @@ export interface IPoolInternal< findFreeWorker: () => Worker | false /** - * Gets worker running tasks. + * Gets worker tasks usage. * * @param worker - The worker. - * @returns The number of tasks currently running on the worker. + * @returns The tasks usage on the worker. */ - getWorkerRunningTasks: (worker: Worker) => number | undefined - - /** - * Gets worker run tasks. - * - * @param worker - The worker. - * @returns The number of tasks run on the worker. - */ - getWorkerRunTasks: (worker: Worker) => number | undefined - - /** - * Gets worker average tasks runtime. - * - * @param worker - The worker. - * @returns The average tasks runtime on the worker. - */ - getWorkerAverageTasksRunTime: (worker: Worker) => number | undefined + getWorkerTasksUsage: (worker: Worker) => TasksUsage | undefined } diff --git a/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts b/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts index 8b063119..7415cf4e 100644 --- a/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts @@ -75,7 +75,7 @@ export class FairShareWorkerChoiceStrategy< start: workerVirtualTaskStartTimestamp, end: workerVirtualTaskStartTimestamp + - (this.pool.getWorkerAverageTasksRunTime(worker) ?? 0) + (this.pool.getWorkerTasksUsage(worker)?.avgRunTime ?? 0) }) } } diff --git a/src/pools/selection-strategies/less-recently-used-worker-choice-strategy.ts b/src/pools/selection-strategies/less-recently-used-worker-choice-strategy.ts index 9e582a6e..fc64a74a 100644 --- a/src/pools/selection-strategies/less-recently-used-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/less-recently-used-worker-choice-strategy.ts @@ -25,9 +25,9 @@ export class LessRecentlyUsedWorkerChoiceStrategy< let lessRecentlyUsedWorker!: Worker for (const value of this.pool.workers.values()) { const worker = value.worker + const tasksUsage = this.pool.getWorkerTasksUsage(worker) const workerTasks = - (this.pool.getWorkerRunTasks(worker) as number) + - (this.pool.getWorkerRunningTasks(worker) as number) + (tasksUsage?.run as number) + (tasksUsage?.running as number) if (!this.isDynamicPool && workerTasks === 0) { return worker } else if (workerTasks < minNumberOfTasks) { 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 2813023f..1af7e656 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 @@ -120,7 +120,7 @@ export class WeightedRoundRobinWorkerChoiceStrategy< } private getWorkerVirtualTaskRunTime (worker: Worker): number | undefined { - return this.pool.getWorkerAverageTasksRunTime(worker) + return this.pool.getWorkerTasksUsage(worker)?.avgRunTime } private computeWorkerWeight (): number { diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js index e6a7ad36..011456ae 100644 --- a/tests/pools/abstract/abstract-pool.test.js +++ b/tests/pools/abstract/abstract-pool.test.js @@ -8,7 +8,7 @@ const { describe('Abstract pool test suite', () => { const numberOfWorkers = 1 - const workerNotFoundInTasksUsageMapError = new Error( + const workerNotFoundInPoolError = new Error( 'Worker could not be found in the pool' ) class StubPoolWithRemoveAllWorker extends FixedThreadPool { @@ -118,78 +118,18 @@ describe('Abstract pool test suite', () => { await pool.destroy() }) - it('Simulate worker not found during increaseWorkerRunningTasks', async () => { - const pool = new StubPoolWithRemoveAllWorker( - numberOfWorkers, - './tests/worker-files/cluster/testWorker.js' - ) - // Simulate worker not found. - pool.removeAllWorker() - expect(() => pool.increaseWorkerRunningTasks()).toThrowError( - workerNotFoundInTasksUsageMapError - ) - await pool.destroy() - }) - - it('Simulate worker not found during decreaseWorkerRunningTasks', async () => { - const pool = new StubPoolWithRemoveAllWorker( - numberOfWorkers, - './tests/worker-files/cluster/testWorker.js', - { - errorHandler: e => console.error(e) - } - ) - // Simulate worker not found. - pool.removeAllWorker() - expect(() => pool.decreaseWorkerRunningTasks()).toThrowError( - workerNotFoundInTasksUsageMapError - ) - await pool.destroy() - }) - - it('Simulate worker not found during stepWorkerRunTasks', async () => { - const pool = new StubPoolWithRemoveAllWorker( - numberOfWorkers, - './tests/worker-files/cluster/testWorker.js', - { - errorHandler: e => console.error(e) - } - ) - // Simulate worker not found. - pool.removeAllWorker() - expect(() => pool.stepWorkerRunTasks()).toThrowError( - workerNotFoundInTasksUsageMapError - ) - await pool.destroy() - }) - - it('Simulate worker not found during updateWorkerTasksRunTime with strategy not requiring it', async () => { - const pool = new StubPoolWithRemoveAllWorker( - numberOfWorkers, - './tests/worker-files/cluster/testWorker.js', - { - errorHandler: e => console.error(e) - } - ) - // Simulate worker not found. - pool.removeAllWorker() - expect(() => pool.updateWorkerTasksRunTime()).not.toThrowError() - await pool.destroy() - }) - - it('Simulate worker not found during updateWorkerTasksRunTime with strategy requiring it', async () => { + it('Simulate worker not found during getWorkerTasksUsage', async () => { const pool = new StubPoolWithRemoveAllWorker( numberOfWorkers, './tests/worker-files/cluster/testWorker.js', { - workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE, errorHandler: e => console.error(e) } ) // Simulate worker not found. pool.removeAllWorker() - expect(() => pool.updateWorkerTasksRunTime()).toThrowError( - workerNotFoundInTasksUsageMapError + expect(() => pool.getWorkerTasksUsage()).toThrowError( + workerNotFoundInPoolError ) await pool.destroy() }) -- 2.34.1