From ffcbbad84f63b8a77f2b1a08f82deef5430f646e Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Sun, 2 Apr 2023 13:55:59 +0200 Subject: [PATCH] perf: use a single map to store pool workers and their related data MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- CHANGELOG.md | 5 + src/pools/abstract-pool.ts | 169 +++++++++--------- src/pools/cluster/dynamic.ts | 2 +- src/pools/pool-internal.ts | 30 ++-- .../fair-share-worker-choice-strategy.ts | 3 +- ...ss-recently-used-worker-choice-strategy.ts | 3 +- .../round-robin-worker-choice-strategy.ts | 15 +- ...hted-round-robin-worker-choice-strategy.ts | 21 +-- src/pools/thread/dynamic.ts | 2 +- tests/pools/abstract/abstract-pool.test.js | 65 ++++--- tests/pools/cluster/dynamic.test.js | 28 +-- tests/pools/cluster/fixed.test.js | 2 +- .../selection-strategies.test.js | 23 ++- ...round-robin-worker-choice-strategy.test.js | 4 +- tests/pools/thread/dynamic.test.js | 28 +-- tests/pools/thread/fixed.test.js | 2 +- tests/test-utils.js | 6 +- tests/worker/thread-worker.test.js | 2 +- 18 files changed, 209 insertions(+), 201 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dc393afa..36db7f82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Optimize worker storage in pool. - Optimize worker alive status check. +### Fixed + +- Ensure trimmable characters are checked at pool initialization. + ## [2.3.10] - 2023-03-18 ### Fixed diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 122e9fa5..1d4e7ae0 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -6,7 +6,7 @@ import { EMPTY_FUNCTION } from '../utils' import { KillBehaviors, isKillBehavior } from '../worker/worker-options' import type { PoolOptions } from './pool' import { PoolEmitter } from './pool' -import type { IPoolInternal, TasksUsage } from './pool-internal' +import type { IPoolInternal, TasksUsage, WorkerType } from './pool-internal' import { PoolType } from './pool-internal' import type { IPoolWorker } from './pool-worker' import { @@ -28,21 +28,23 @@ export abstract class AbstractPool< Response = unknown > implements IPoolInternal { /** {@inheritDoc} */ - public readonly workers: Worker[] = [] - - /** {@inheritDoc} */ - public readonly workersTasksUsage: Map = new Map< - Worker, - TasksUsage + public readonly workers: Map> = new Map< + number, + WorkerType >() /** {@inheritDoc} */ public readonly emitter?: PoolEmitter + /** + * Id of the next worker. + */ + protected nextWorkerId: number = 0 + /** * The promise map. * - * - `key`: This is the message Id of each submitted task. + * - `key`: This is the message id of each submitted task. * - `value`: An object that contains the worker, the resolve function and the reject function. * * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message. @@ -115,7 +117,10 @@ export abstract class AbstractPool< } private checkFilePath (filePath: string): void { - if (filePath == null || filePath.length === 0) { + if ( + filePath == null || + (typeof filePath === 'string' && filePath.trim().length === 0) + ) { throw new Error('Please specify a file with a worker implementation') } } @@ -152,19 +157,26 @@ export abstract class AbstractPool< return this.promiseMap.size } - /** {@inheritDoc} */ - public getWorkerIndex (worker: Worker): number { - return this.workers.indexOf(worker) + /** + * Gets worker key. + * + * @param worker - The worker. + * @returns The worker key. + */ + private getWorkerKey (worker: Worker): number | undefined { + return [...this.workers].find(([, value]) => value.worker === worker)?.[0] } /** {@inheritDoc} */ public getWorkerRunningTasks (worker: Worker): number | undefined { - return this.workersTasksUsage.get(worker)?.running + return this.workers.get(this.getWorkerKey(worker) as number)?.tasksUsage + ?.running } /** {@inheritDoc} */ public getWorkerAverageTasksRunTime (worker: Worker): number | undefined { - return this.workersTasksUsage.get(worker)?.avgRunTime + return this.workers.get(this.getWorkerKey(worker) as number)?.tasksUsage + ?.avgRunTime } /** {@inheritDoc} */ @@ -172,8 +184,13 @@ export abstract class AbstractPool< workerChoiceStrategy: WorkerChoiceStrategy ): void { this.opts.workerChoiceStrategy = workerChoiceStrategy - for (const worker of this.workers) { - this.resetWorkerTasksUsage(worker) + for (const [key, value] of this.workers) { + this.setWorker(key, value.worker, { + run: 0, + running: 0, + runTime: 0, + avgRunTime: 0 + }) } this.workerChoiceStrategyContext.setWorkerChoiceStrategy( workerChoiceStrategy @@ -192,10 +209,10 @@ export abstract class AbstractPool< /** {@inheritDoc} */ public findFreeWorker (): Worker | false { - for (const worker of this.workers) { - if (this.getWorkerRunningTasks(worker) === 0) { + for (const value of this.workers.values()) { + if (value.tasksUsage.running === 0) { // A worker is free, return the matching worker - return worker + return value.worker } } return false @@ -220,8 +237,8 @@ export abstract class AbstractPool< /** {@inheritDoc} */ public async destroy (): Promise { await Promise.all( - this.workers.map(async worker => { - await this.destroyWorker(worker) + [...this.workers].map(async ([, value]) => { + await this.destroyWorker(value.worker) }) ) } @@ -278,9 +295,8 @@ export abstract class AbstractPool< * @param worker - The worker that will be removed. */ protected removeWorker (worker: Worker): void { - // Clean worker from data structure - this.workers.splice(this.getWorkerIndex(worker), 1) - this.removeWorkerTasksUsage(worker) + this.workers.delete(this.getWorkerKey(worker) as number) + --this.nextWorkerId } /** @@ -345,10 +361,13 @@ export abstract class AbstractPool< this.removeWorker(worker) }) - this.workers.push(worker) - - // Init worker tasks usage map - this.initWorkerTasksUsage(worker) + this.setWorker(this.nextWorkerId, worker, { + run: 0, + running: 0, + runTime: 0, + avgRunTime: 0 + }) + ++this.nextWorkerId this.afterWorkerSetup(worker) @@ -411,6 +430,19 @@ export abstract class AbstractPool< this.stepWorkerRunningTasks(worker, -1) } + /** + * Get 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 + const workerEntry = this.workers.get(workerKey) as WorkerType + return workerEntry.tasksUsage + } + } + /** * Steps the number of tasks that the given worker has applied. * @@ -418,11 +450,8 @@ export abstract class AbstractPool< * @param step - Number of running tasks step. */ private stepWorkerRunningTasks (worker: Worker, step: number): void { - if (this.checkWorkerTasksUsage(worker)) { - const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage - tasksUsage.running = tasksUsage.running + step - this.workersTasksUsage.set(worker, tasksUsage) - } + // prettier-ignore + (this.getWorkerTasksUsage(worker) as TasksUsage).running += step } /** @@ -432,11 +461,8 @@ export abstract class AbstractPool< * @param step - Number of run tasks step. */ private stepWorkerRunTasks (worker: Worker, step: number): void { - if (this.checkWorkerTasksUsage(worker)) { - const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage - tasksUsage.run = tasksUsage.run + step - this.workersTasksUsage.set(worker, tasksUsage) - } + // prettier-ignore + (this.getWorkerTasksUsage(worker) as TasksUsage).run += step } /** @@ -451,62 +477,45 @@ export abstract class AbstractPool< ): void { if ( this.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .requiredStatistics.runTime && - this.checkWorkerTasksUsage(worker) + .requiredStatistics.runTime ) { - const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage - tasksUsage.runTime += taskRunTime ?? 0 - if (tasksUsage.run !== 0) { - tasksUsage.avgRunTime = tasksUsage.runTime / tasksUsage.run + const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage + workerTasksUsage.runTime += taskRunTime ?? 0 + if (workerTasksUsage.run !== 0) { + workerTasksUsage.avgRunTime = + workerTasksUsage.runTime / workerTasksUsage.run } - this.workersTasksUsage.set(worker, tasksUsage) } } /** - * Checks if the given worker is registered in the workers tasks usage map. - * - * @param worker - Worker to check. - * @returns `true` if the worker is registered in the workers tasks usage map. `false` otherwise. - */ - private checkWorkerTasksUsage (worker: Worker): boolean { - const hasTasksUsage = this.workersTasksUsage.has(worker) - if (!hasTasksUsage) { - throw new Error('Worker could not be found in workers tasks usage map') - } - return hasTasksUsage - } - - /** - * Initializes tasks usage statistics. + * Sets the given worker. * + * @param workerKey - The worker key. * @param worker - The worker. + * @param tasksUsage - The worker tasks usage. */ - private initWorkerTasksUsage (worker: Worker): void { - this.workersTasksUsage.set(worker, { - run: 0, - running: 0, - runTime: 0, - avgRunTime: 0 + private setWorker ( + workerKey: number, + worker: Worker, + tasksUsage: TasksUsage + ): void { + this.workers.set(workerKey, { + worker, + tasksUsage }) } /** - * Removes worker tasks usage statistics. - * - * @param worker - The worker. - */ - private removeWorkerTasksUsage (worker: Worker): void { - this.workersTasksUsage.delete(worker) - } - - /** - * Resets worker tasks usage statistics. + * Checks if the given worker is registered in the pool. * - * @param worker - The worker. + * @param worker - Worker to check. + * @returns `true` if the worker is registered in the pool. */ - private resetWorkerTasksUsage (worker: Worker): void { - this.removeWorkerTasksUsage(worker) - this.initWorkerTasksUsage(worker) + 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/cluster/dynamic.ts b/src/pools/cluster/dynamic.ts index 8b8c5185..0a2d0f38 100644 --- a/src/pools/cluster/dynamic.ts +++ b/src/pools/cluster/dynamic.ts @@ -41,6 +41,6 @@ export class DynamicClusterPool< /** {@inheritDoc} */ public get busy (): boolean { - return this.workers.length === this.max + return this.workers.size === this.max } } diff --git a/src/pools/pool-internal.ts b/src/pools/pool-internal.ts index e0d0534f..97e5d5d2 100644 --- a/src/pools/pool-internal.ts +++ b/src/pools/pool-internal.ts @@ -19,6 +19,16 @@ export interface TasksUsage { avgRunTime: number } +/** + * Internal worker type. + * + * @typeParam Worker - Type of worker which manages this pool. + */ +export interface WorkerType { + worker: Worker + tasksUsage: TasksUsage +} + /** * Internal contract definition for a poolifier pool. * @@ -32,17 +42,9 @@ export interface IPoolInternal< Response = unknown > extends IPool { /** - * List of currently available workers. + * Map of workers. */ - readonly workers: Worker[] - - /** - * The workers tasks usage map. - * - * `key`: The `Worker` - * `value`: Worker tasks usage statistics. - */ - readonly workersTasksUsage: Map + readonly workers: Map> /** * Pool type. @@ -74,14 +76,6 @@ export interface IPoolInternal< */ findFreeWorker: () => Worker | false - /** - * Gets worker index. - * - * @param worker - The worker. - * @returns The worker index. - */ - getWorkerIndex: (worker: Worker) => number - /** * Gets worker running tasks. * 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 00ed1c90..8b063119 100644 --- a/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts @@ -46,7 +46,8 @@ export class FairShareWorkerChoiceStrategy< public choose (): Worker { let minWorkerVirtualTaskEndTimestamp = Infinity let chosenWorker!: Worker - for (const worker of this.pool.workers) { + for (const value of this.pool.workers.values()) { + const worker = value.worker this.computeWorkerLastVirtualTaskTimestamp(worker) const workerLastVirtualTaskEndTimestamp = this.workerLastVirtualTaskTimestamp.get(worker)?.end ?? 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 e62d9cdf..467cf2c6 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 @@ -23,7 +23,8 @@ export class LessRecentlyUsedWorkerChoiceStrategy< let minNumberOfRunningTasks = Infinity // A worker is always found because it picks the one with fewer tasks let lessRecentlyUsedWorker!: Worker - for (const worker of this.pool.workers) { + for (const value of this.pool.workers.values()) { + const worker = value.worker const workerRunningTasks = this.pool.getWorkerRunningTasks( worker ) as number diff --git a/src/pools/selection-strategies/round-robin-worker-choice-strategy.ts b/src/pools/selection-strategies/round-robin-worker-choice-strategy.ts index 76cdd4ab..44dfa4b7 100644 --- a/src/pools/selection-strategies/round-robin-worker-choice-strategy.ts +++ b/src/pools/selection-strategies/round-robin-worker-choice-strategy.ts @@ -14,23 +14,24 @@ export class RoundRobinWorkerChoiceStrategy< Response > extends AbstractWorkerChoiceStrategy { /** - * Index for the next worker. + * Id of the next worker. */ - private nextWorkerIndex: number = 0 + private nextWorkerId: number = 0 /** {@inheritDoc} */ public reset (): boolean { - this.nextWorkerIndex = 0 + this.nextWorkerId = 0 return true } /** {@inheritDoc} */ public choose (): Worker { - const chosenWorker = this.pool.workers[this.nextWorkerIndex] - this.nextWorkerIndex = - this.nextWorkerIndex === this.pool.workers.length - 1 + const chosenWorker = this.pool.workers.get(this.nextWorkerId) + ?.worker as Worker + this.nextWorkerId = + this.nextWorkerId === this.pool.workers.size - 1 ? 0 - : this.nextWorkerIndex + 1 + : this.nextWorkerId + 1 return chosenWorker } } 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 2b7cd942..fc9b0695 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 @@ -31,9 +31,9 @@ export class WeightedRoundRobinWorkerChoiceStrategy< } /** - * Worker index where the current task will be submitted. + * Worker id where the current task will be submitted. */ - private currentWorkerIndex: number = 0 + private currentWorkerId: number = 0 /** * Default worker weight. */ @@ -59,7 +59,7 @@ export class WeightedRoundRobinWorkerChoiceStrategy< /** {@inheritDoc} */ public reset (): boolean { - this.currentWorkerIndex = 0 + this.currentWorkerId = 0 this.workersTaskRunTime.clear() this.initWorkersTaskRunTime() return true @@ -67,7 +67,8 @@ export class WeightedRoundRobinWorkerChoiceStrategy< /** {@inheritDoc} */ public choose (): Worker { - const chosenWorker = this.pool.workers[this.currentWorkerIndex] + const chosenWorker = this.pool.workers.get(this.currentWorkerId) + ?.worker as Worker if (this.isDynamicPool && !this.workersTaskRunTime.has(chosenWorker)) { this.initWorkerTaskRunTime(chosenWorker) } @@ -84,12 +85,12 @@ export class WeightedRoundRobinWorkerChoiceStrategy< (this.getWorkerVirtualTaskRunTime(chosenWorker) ?? 0) ) } else { - this.currentWorkerIndex = - this.currentWorkerIndex === this.pool.workers.length - 1 + this.currentWorkerId = + this.currentWorkerId === this.pool.workers.size - 1 ? 0 - : this.currentWorkerIndex + 1 + : this.currentWorkerId + 1 this.setWorkerTaskRunTime( - this.pool.workers[this.currentWorkerIndex], + this.pool.workers.get(this.currentWorkerId)?.worker as Worker, workerTaskWeight, 0 ) @@ -98,8 +99,8 @@ export class WeightedRoundRobinWorkerChoiceStrategy< } private initWorkersTaskRunTime (): void { - for (const worker of this.pool.workers) { - this.initWorkerTaskRunTime(worker) + for (const value of this.pool.workers.values()) { + this.initWorkerTaskRunTime(value.worker) } } diff --git a/src/pools/thread/dynamic.ts b/src/pools/thread/dynamic.ts index 19cb9fbf..d5be43f1 100644 --- a/src/pools/thread/dynamic.ts +++ b/src/pools/thread/dynamic.ts @@ -42,6 +42,6 @@ export class DynamicThreadPool< /** {@inheritDoc} */ public get busy (): boolean { - return this.workers.length === this.max + return this.workers.size === this.max } } diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js index 80ebdb9e..e6a7ad36 100644 --- a/tests/pools/abstract/abstract-pool.test.js +++ b/tests/pools/abstract/abstract-pool.test.js @@ -9,12 +9,11 @@ const { describe('Abstract pool test suite', () => { const numberOfWorkers = 1 const workerNotFoundInTasksUsageMapError = new Error( - 'Worker could not be found in workers tasks usage map' + 'Worker could not be found in the pool' ) class StubPoolWithRemoveAllWorker extends FixedThreadPool { removeAllWorker () { - this.workers = [] - this.workersTasksUsage.clear() + this.workers = new Map() this.promiseMap.clear() } } @@ -200,12 +199,12 @@ describe('Abstract pool test suite', () => { numberOfWorkers, './tests/worker-files/cluster/testWorker.js' ) - for (const tasksUsage of pool.workersTasksUsage.values()) { - expect(tasksUsage).toBeDefined() - expect(tasksUsage.run).toBe(0) - expect(tasksUsage.running).toBe(0) - expect(tasksUsage.runTime).toBe(0) - expect(tasksUsage.avgRunTime).toBe(0) + for (const value of pool.workers.values()) { + expect(value.tasksUsage).toBeDefined() + expect(value.tasksUsage.run).toBe(0) + expect(value.tasksUsage.running).toBe(0) + expect(value.tasksUsage.runTime).toBe(0) + expect(value.tasksUsage.avgRunTime).toBe(0) } await pool.destroy() }) @@ -219,20 +218,20 @@ describe('Abstract pool test suite', () => { for (let i = 0; i < numberOfWorkers * 2; i++) { promises.push(pool.execute()) } - for (const tasksUsage of pool.workersTasksUsage.values()) { - expect(tasksUsage).toBeDefined() - expect(tasksUsage.run).toBe(0) - expect(tasksUsage.running).toBe(numberOfWorkers * 2) - expect(tasksUsage.runTime).toBe(0) - expect(tasksUsage.avgRunTime).toBe(0) + for (const value of pool.workers.values()) { + expect(value.tasksUsage).toBeDefined() + expect(value.tasksUsage.run).toBe(0) + expect(value.tasksUsage.running).toBe(numberOfWorkers * 2) + expect(value.tasksUsage.runTime).toBe(0) + expect(value.tasksUsage.avgRunTime).toBe(0) } await Promise.all(promises) - for (const tasksUsage of pool.workersTasksUsage.values()) { - expect(tasksUsage).toBeDefined() - expect(tasksUsage.run).toBe(numberOfWorkers * 2) - expect(tasksUsage.running).toBe(0) - expect(tasksUsage.runTime).toBeGreaterThanOrEqual(0) - expect(tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0) + for (const value of pool.workers.values()) { + expect(value.tasksUsage).toBeDefined() + expect(value.tasksUsage.run).toBe(numberOfWorkers * 2) + expect(value.tasksUsage.running).toBe(0) + expect(value.tasksUsage.runTime).toBeGreaterThanOrEqual(0) + expect(value.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0) } await pool.destroy() }) @@ -248,20 +247,20 @@ describe('Abstract pool test suite', () => { promises.push(pool.execute()) } await Promise.all(promises) - for (const tasksUsage of pool.workersTasksUsage.values()) { - expect(tasksUsage).toBeDefined() - expect(tasksUsage.run).toBe(numberOfWorkers * 2) - expect(tasksUsage.running).toBe(0) - expect(tasksUsage.runTime).toBeGreaterThanOrEqual(0) - expect(tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0) + for (const value of pool.workers.values()) { + expect(value.tasksUsage).toBeDefined() + expect(value.tasksUsage.run).toBe(numberOfWorkers * 2) + expect(value.tasksUsage.running).toBe(0) + expect(value.tasksUsage.runTime).toBeGreaterThanOrEqual(0) + expect(value.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0) } pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE) - for (const tasksUsage of pool.workersTasksUsage.values()) { - expect(tasksUsage).toBeDefined() - expect(tasksUsage.run).toBe(0) - expect(tasksUsage.running).toBe(0) - expect(tasksUsage.runTime).toBe(0) - expect(tasksUsage.avgRunTime).toBe(0) + for (const value of pool.workers.values()) { + expect(value.tasksUsage).toBeDefined() + expect(value.tasksUsage.run).toBe(0) + expect(value.tasksUsage.running).toBe(0) + expect(value.tasksUsage.runTime).toBe(0) + expect(value.tasksUsage.avgRunTime).toBe(0) } await pool.destroy() }) diff --git a/tests/pools/cluster/dynamic.test.js b/tests/pools/cluster/dynamic.test.js index b1fd166e..7a834ca2 100644 --- a/tests/pools/cluster/dynamic.test.js +++ b/tests/pools/cluster/dynamic.test.js @@ -28,12 +28,12 @@ describe('Dynamic cluster pool test suite', () => { it('Verify that new workers are created when required, max size is not exceeded and that after a while new workers will die', async () => { let poolBusy = 0 - pool.emitter.on('busy', () => poolBusy++) + pool.emitter.on('busy', () => ++poolBusy) for (let i = 0; i < max * 2; i++) { pool.execute() } - expect(pool.workers.length).toBeLessThanOrEqual(max) - expect(pool.workers.length).toBeGreaterThan(min) + expect(pool.workers.size).toBeLessThanOrEqual(max) + expect(pool.workers.size).toBeGreaterThan(min) // The `busy` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool. // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool. expect(poolBusy).toBe(max + 1) @@ -42,19 +42,19 @@ describe('Dynamic cluster pool test suite', () => { }) it('Verify scale worker up and down is working', async () => { - expect(pool.workers.length).toBe(min) + expect(pool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { pool.execute() } - expect(pool.workers.length).toBeGreaterThan(min) + expect(pool.workers.size).toBeGreaterThan(min) await TestUtils.waitExits(pool, max - min) - expect(pool.workers.length).toBe(min) + expect(pool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { pool.execute() } - expect(pool.workers.length).toBeGreaterThan(min) + expect(pool.workers.size).toBeGreaterThan(min) await TestUtils.waitExits(pool, max - min) - expect(pool.workers.length).toBe(min) + expect(pool.workers.size).toBe(min) }) it('Shutdown test', async () => { @@ -93,14 +93,14 @@ describe('Dynamic cluster pool test suite', () => { exitHandler: () => console.log('long running worker exited') } ) - expect(longRunningPool.workers.length).toBe(min) + expect(longRunningPool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { longRunningPool.execute() } - expect(longRunningPool.workers.length).toBe(max) + expect(longRunningPool.workers.size).toBe(max) await TestUtils.waitExits(longRunningPool, max - min) // Here we expect the workers to be at the max size since that the task is still running - expect(longRunningPool.workers.length).toBe(min) + expect(longRunningPool.workers.size).toBe(min) // We need to clean up the resources after our test await longRunningPool.destroy() }) @@ -116,14 +116,14 @@ describe('Dynamic cluster pool test suite', () => { exitHandler: () => console.log('long running worker exited') } ) - expect(longRunningPool.workers.length).toBe(min) + expect(longRunningPool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { longRunningPool.execute() } - expect(longRunningPool.workers.length).toBe(max) + expect(longRunningPool.workers.size).toBe(max) await TestUtils.sleep(1500) // Here we expect the workers to be at the max size since that the task is still running - expect(longRunningPool.workers.length).toBe(max) + expect(longRunningPool.workers.size).toBe(max) // We need to clean up the resources after our test await longRunningPool.destroy() }) diff --git a/tests/pools/cluster/fixed.test.js b/tests/pools/cluster/fixed.test.js index b0e39c30..ae899625 100644 --- a/tests/pools/cluster/fixed.test.js +++ b/tests/pools/cluster/fixed.test.js @@ -67,7 +67,7 @@ describe('Fixed cluster pool test suite', () => { it('Verify that busy event is emitted', async () => { let poolBusy = 0 - pool.emitter.on('busy', () => poolBusy++) + pool.emitter.on('busy', () => ++poolBusy) for (let i = 0; i < numberOfWorkers * 2; i++) { pool.execute() } diff --git a/tests/pools/selection-strategies/selection-strategies.test.js b/tests/pools/selection-strategies/selection-strategies.test.js index 93d85ea6..c55b971e 100644 --- a/tests/pools/selection-strategies/selection-strategies.test.js +++ b/tests/pools/selection-strategies/selection-strategies.test.js @@ -42,7 +42,7 @@ describe('Selection strategies test suite', () => { WorkerChoiceStrategies.ROUND_ROBIN ) expect( - pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().nextWorkerIndex + pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().nextWorkerId ).toBe(0) // We need to clean up the resources after our test await pool.destroy() @@ -153,11 +153,11 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN } ) expect( - pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().nextWorkerIndex + pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().nextWorkerId ).toBeUndefined() pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.ROUND_ROBIN) expect( - pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().nextWorkerIndex + pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().nextWorkerId ).toBe(0) await pool.destroy() pool = new DynamicThreadPool( @@ -168,12 +168,12 @@ describe('Selection strategies test suite', () => { ) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .workerChoiceStrategy.nextWorkerIndex + .workerChoiceStrategy.nextWorkerId ).toBeUndefined() pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.ROUND_ROBIN) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .workerChoiceStrategy.nextWorkerIndex + .workerChoiceStrategy.nextWorkerId ).toBe(0) // We need to clean up the resources after our test await pool.destroy() @@ -424,8 +424,7 @@ describe('Selection strategies test suite', () => { WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN ) expect( - pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .currentWorkerIndex + pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().currentWorkerId ).toBe(0) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() @@ -526,8 +525,7 @@ describe('Selection strategies test suite', () => { './tests/worker-files/thread/testWorker.js' ) expect( - pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .currentWorkerIndex + pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().currentWorkerId ).toBeUndefined() expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() @@ -539,8 +537,7 @@ describe('Selection strategies test suite', () => { ).toBeUndefined() pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN) expect( - pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .currentWorkerIndex + pool.workerChoiceStrategyContext.getWorkerChoiceStrategy().currentWorkerId ).toBe(0) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() @@ -563,7 +560,7 @@ describe('Selection strategies test suite', () => { ) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .workerChoiceStrategy.currentWorkerIndex + .workerChoiceStrategy.currentWorkerId ).toBeUndefined() expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() @@ -576,7 +573,7 @@ describe('Selection strategies test suite', () => { pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .workerChoiceStrategy.currentWorkerIndex + .workerChoiceStrategy.currentWorkerId ).toBe(0) expect( pool.workerChoiceStrategyContext.getWorkerChoiceStrategy() diff --git a/tests/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.test.js b/tests/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.test.js index d8ae8299..cc3e66a4 100644 --- a/tests/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.test.js +++ b/tests/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.test.js @@ -25,7 +25,7 @@ describe('Weighted round robin strategy worker choice strategy test suite', () = it('Verify that reset() resets internals', () => { const strategy = new WeightedRoundRobinWorkerChoiceStrategy(pool) - strategy.currentWorkerIndex = TestUtils.generateRandomInteger() + strategy.currentWorkerId = TestUtils.generateRandomInteger() const workersTaskRunTimeClearStub = sinon .stub(strategy.workersTaskRunTime, 'clear') .returns() @@ -34,7 +34,7 @@ describe('Weighted round robin strategy worker choice strategy test suite', () = .returns() const resetResult = strategy.reset() expect(resetResult).toBe(true) - expect(strategy.currentWorkerIndex).toBe(0) + expect(strategy.currentWorkerId).toBe(0) expect(workersTaskRunTimeClearStub.calledOnce).toBe(true) expect(initWorkersTaskRunTimeStub.calledOnce).toBe(true) }) diff --git a/tests/pools/thread/dynamic.test.js b/tests/pools/thread/dynamic.test.js index fad9dab8..b544452f 100644 --- a/tests/pools/thread/dynamic.test.js +++ b/tests/pools/thread/dynamic.test.js @@ -28,12 +28,12 @@ describe('Dynamic thread pool test suite', () => { it('Verify that new workers are created when required, max size is not exceeded and that after a while new workers will die', async () => { let poolBusy = 0 - pool.emitter.on('busy', () => poolBusy++) + pool.emitter.on('busy', () => ++poolBusy) for (let i = 0; i < max * 2; i++) { pool.execute() } - expect(pool.workers.length).toBeLessThanOrEqual(max) - expect(pool.workers.length).toBeGreaterThan(min) + expect(pool.workers.size).toBeLessThanOrEqual(max) + expect(pool.workers.size).toBeGreaterThan(min) // The `busy` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool. // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool. expect(poolBusy).toBe(max + 1) @@ -42,19 +42,19 @@ describe('Dynamic thread pool test suite', () => { }) it('Verify scale thread up and down is working', async () => { - expect(pool.workers.length).toBe(min) + expect(pool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { pool.execute() } - expect(pool.workers.length).toBe(max) + expect(pool.workers.size).toBe(max) await TestUtils.waitExits(pool, max - min) - expect(pool.workers.length).toBe(min) + expect(pool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { pool.execute() } - expect(pool.workers.length).toBe(max) + expect(pool.workers.size).toBe(max) await TestUtils.waitExits(pool, max - min) - expect(pool.workers.length).toBe(min) + expect(pool.workers.size).toBe(min) }) it('Shutdown test', async () => { @@ -93,13 +93,13 @@ describe('Dynamic thread pool test suite', () => { exitHandler: () => console.log('long running worker exited') } ) - expect(longRunningPool.workers.length).toBe(min) + expect(longRunningPool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { longRunningPool.execute() } - expect(longRunningPool.workers.length).toBe(max) + expect(longRunningPool.workers.size).toBe(max) await TestUtils.waitExits(longRunningPool, max - min) - expect(longRunningPool.workers.length).toBe(min) + expect(longRunningPool.workers.size).toBe(min) // We need to clean up the resources after our test await longRunningPool.destroy() }) @@ -115,14 +115,14 @@ describe('Dynamic thread pool test suite', () => { exitHandler: () => console.log('long running worker exited') } ) - expect(longRunningPool.workers.length).toBe(min) + expect(longRunningPool.workers.size).toBe(min) for (let i = 0; i < max * 10; i++) { longRunningPool.execute() } - expect(longRunningPool.workers.length).toBe(max) + expect(longRunningPool.workers.size).toBe(max) await TestUtils.sleep(1500) // Here we expect the workers to be at the max size since that the task is still running - expect(longRunningPool.workers.length).toBe(max) + expect(longRunningPool.workers.size).toBe(max) // We need to clean up the resources after our test await longRunningPool.destroy() }) diff --git a/tests/pools/thread/fixed.test.js b/tests/pools/thread/fixed.test.js index 116a17da..fd9ce5df 100644 --- a/tests/pools/thread/fixed.test.js +++ b/tests/pools/thread/fixed.test.js @@ -67,7 +67,7 @@ describe('Fixed thread pool test suite', () => { it('Verify that busy event is emitted', async () => { let poolBusy = 0 - pool.emitter.on('busy', () => poolBusy++) + pool.emitter.on('busy', () => ++poolBusy) for (let i = 0; i < numberOfThreads * 2; i++) { pool.execute() } diff --git a/tests/test-utils.js b/tests/test-utils.js index 84bec7e5..7314f261 100644 --- a/tests/test-utils.js +++ b/tests/test-utils.js @@ -4,9 +4,9 @@ class TestUtils { static async waitExits (pool, numberOfExitEventsToWait) { return new Promise(resolve => { let exitEvents = 0 - for (const worker of pool.workers) { - worker.on('exit', () => { - exitEvents++ + for (const value of pool.workers.values()) { + value.worker.on('exit', () => { + ++exitEvents if (exitEvents === numberOfExitEventsToWait) { resolve(exitEvents) } diff --git a/tests/worker/thread-worker.test.js b/tests/worker/thread-worker.test.js index c04d3b2f..624e395c 100644 --- a/tests/worker/thread-worker.test.js +++ b/tests/worker/thread-worker.test.js @@ -4,7 +4,7 @@ const { ThreadWorker } = require('../../lib') describe('Thread worker test suite', () => { let numberOfMessagesPosted = 0 const postMessage = function () { - numberOfMessagesPosted++ + ++numberOfMessagesPosted } class SpyWorker extends ThreadWorker { getMainWorker () { -- 2.34.1