X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fselection-strategies%2Fselection-strategies.test.js;h=19ac5ea3a9a930c67996efd31fb5a683aa856685;hb=ee9f52959d4996f56712791e73994dfbeb0370f0;hp=7755cc341f6ec97f143ba7253dae95601345de3e;hpb=89b09b2646d240a5a30531b2fc742fdba209fa13;p=poolifier.git diff --git a/tests/pools/selection-strategies/selection-strategies.test.js b/tests/pools/selection-strategies/selection-strategies.test.js index 7755cc34..19ac5ea3 100644 --- a/tests/pools/selection-strategies/selection-strategies.test.js +++ b/tests/pools/selection-strategies/selection-strategies.test.js @@ -4,7 +4,8 @@ const { DynamicThreadPool, FixedThreadPool, FixedClusterPool -} = require('../../../lib/index') +} = require('../../../lib') +const { CircularArray } = require('../../../lib/circular-array') describe('Selection strategies test suite', () => { const min = 0 @@ -12,8 +13,8 @@ describe('Selection strategies test suite', () => { it('Verify that WorkerChoiceStrategies enumeration provides string values', () => { expect(WorkerChoiceStrategies.ROUND_ROBIN).toBe('ROUND_ROBIN') - expect(WorkerChoiceStrategies.LESS_USED).toBe('LESS_USED') - expect(WorkerChoiceStrategies.LESS_BUSY).toBe('LESS_BUSY') + expect(WorkerChoiceStrategies.LEAST_USED).toBe('LEAST_USED') + expect(WorkerChoiceStrategies.LEAST_BUSY).toBe('LEAST_BUSY') expect(WorkerChoiceStrategies.FAIR_SHARE).toBe('FAIR_SHARE') expect(WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN).toBe( 'WEIGHTED_ROUND_ROBIN' @@ -77,20 +78,16 @@ describe('Selection strategies test suite', () => { ).nextWorkerNodeId ).toBe(0) } else if (workerChoiceStrategy === WorkerChoiceStrategies.FAIR_SHARE) { - for (const workerNodeKey of pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.keys()) { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.get(workerNodeKey).start - ).toBe(0) - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.get(workerNodeKey).end - ).toBe(0) - } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp + ).toBeInstanceOf(Array) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(0) } else if ( workerChoiceStrategy === WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN ) { @@ -104,20 +101,11 @@ describe('Selection strategies test suite', () => { workerChoiceStrategy ).defaultWorkerWeight ).toBeGreaterThan(0) - for (const workerNodeKey of pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(workerChoiceStrategy) - .workersTaskRunTime.keys()) { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(workerChoiceStrategy) - .workersTaskRunTime.get(workerNodeKey).weight - ).toBeGreaterThan(0) - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(workerChoiceStrategy) - .workersTaskRunTime.get(workerNodeKey).runTime - ).toBe(0) - } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workerVirtualTaskRunTime + ).toBe(0) } } await pool.destroy() @@ -131,14 +119,15 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: false, + avgRunTime: false, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) await pool.destroy() pool = new DynamicThreadPool( min, @@ -147,14 +136,15 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: false, + avgRunTime: false, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) // We need to clean up the resources after our test await pool.destroy() }) @@ -166,11 +156,29 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN } ) // TODO: Create a better test to cover `RoundRobinWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: 0, + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + } // We need to clean up the resources after our test await pool.destroy() }) @@ -183,11 +191,29 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN } ) // TODO: Create a better test to cover `RoundRobinWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: 0, + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + } // We need to clean up the resources after our test await pool.destroy() }) @@ -201,7 +227,7 @@ describe('Selection strategies test suite', () => { ) let results = new Set() for (let i = 0; i < max; i++) { - results.add(pool.chooseWorkerNode()[1].worker.id) + results.add(pool.workerNodes[pool.chooseWorkerNode()].worker.id) } expect(results.size).toBe(max) await pool.destroy() @@ -212,7 +238,7 @@ describe('Selection strategies test suite', () => { ) results = new Set() for (let i = 0; i < max; i++) { - results.add(pool.chooseWorkerNode()[1].worker.threadId) + results.add(pool.workerNodes[pool.chooseWorkerNode()].worker.threadId) } expect(results.size).toBe(max) await pool.destroy() @@ -258,22 +284,23 @@ describe('Selection strategies test suite', () => { await pool.destroy() }) - it('Verify LESS_USED strategy default tasks usage statistics requirements', async () => { - const workerChoiceStrategy = WorkerChoiceStrategies.LESS_USED + it('Verify LEAST_USED strategy default tasks usage statistics requirements', async () => { + const workerChoiceStrategy = WorkerChoiceStrategies.LEAST_USED let pool = new FixedThreadPool( max, './tests/worker-files/thread/testWorker.js', { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: false, + avgRunTime: false, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) await pool.destroy() pool = new DynamicThreadPool( min, @@ -282,67 +309,105 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: false, + avgRunTime: false, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) // We need to clean up the resources after our test await pool.destroy() }) - it('Verify LESS_USED strategy can be run in a fixed pool', async () => { + it('Verify LEAST_USED strategy can be run in a fixed pool', async () => { const pool = new FixedThreadPool( max, './tests/worker-files/thread/testWorker.js', - { workerChoiceStrategy: WorkerChoiceStrategies.LESS_USED } + { workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED } ) - // TODO: Create a better test to cover `LessUsedWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + // TODO: Create a better test to cover `LeastUsedWorkerChoiceStrategy#choose` + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: 0, + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + } // We need to clean up the resources after our test await pool.destroy() }) - it('Verify LESS_USED strategy can be run in a dynamic pool', async () => { + it('Verify LEAST_USED strategy can be run in a dynamic pool', async () => { const pool = new DynamicThreadPool( min, max, './tests/worker-files/thread/testWorker.js', - { workerChoiceStrategy: WorkerChoiceStrategies.LESS_USED } + { workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED } ) - // TODO: Create a better test to cover `LessUsedWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + // TODO: Create a better test to cover `LeastUsedWorkerChoiceStrategy#choose` + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: 0, + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + } // We need to clean up the resources after our test await pool.destroy() }) - it('Verify LESS_BUSY strategy default tasks usage statistics requirements', async () => { - const workerChoiceStrategy = WorkerChoiceStrategies.LESS_BUSY + it('Verify LEAST_BUSY strategy default tasks usage statistics requirements', async () => { + const workerChoiceStrategy = WorkerChoiceStrategies.LEAST_BUSY let pool = new FixedThreadPool( max, './tests/worker-files/thread/testWorker.js', { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: true, + avgRunTime: false, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) await pool.destroy() pool = new DynamicThreadPool( min, @@ -351,47 +416,86 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(false) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: true, + avgRunTime: false, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) // We need to clean up the resources after our test await pool.destroy() }) - it('Verify LESS_BUSY strategy can be run in a fixed pool', async () => { + it('Verify LEAST_BUSY strategy can be run in a fixed pool', async () => { const pool = new FixedThreadPool( max, './tests/worker-files/thread/testWorker.js', - { workerChoiceStrategy: WorkerChoiceStrategies.LESS_BUSY } + { workerChoiceStrategy: WorkerChoiceStrategies.LEAST_BUSY } ) - // TODO: Create a better test to cover `LessBusyWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + // TODO: Create a better test to cover `LeastBusyWorkerChoiceStrategy#choose` + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + } // We need to clean up the resources after our test await pool.destroy() }) - it('Verify LESS_BUSY strategy can be run in a dynamic pool', async () => { + it('Verify LEAST_BUSY strategy can be run in a dynamic pool', async () => { const pool = new DynamicThreadPool( min, max, './tests/worker-files/thread/testWorker.js', - { workerChoiceStrategy: WorkerChoiceStrategies.LESS_BUSY } + { workerChoiceStrategy: WorkerChoiceStrategies.LEAST_BUSY } ) - // TODO: Create a better test to cover `LessBusyWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + // TODO: Create a better test to cover `LeastBusyWorkerChoiceStrategy#choose` + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + } // We need to clean up the resources after our test await pool.destroy() }) @@ -404,14 +508,15 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: true, + avgRunTime: true, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) await pool.destroy() pool = new DynamicThreadPool( min, @@ -420,14 +525,15 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: true, + avgRunTime: true, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) // We need to clean up the resources after our test await pool.destroy() }) @@ -439,15 +545,35 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE } ) // TODO: Create a better test to cover `FairShareChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: expect.any(Number), + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + expect(workerNode.tasksUsage.avgRunTime).toBeGreaterThan(0) + } expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( pool.workerChoiceStrategyContext.workerChoiceStrategy - ).workerLastVirtualTaskTimestamp.size + ).workersVirtualTaskEndTimestamp.length ).toBe(pool.workerNodes.length) // We need to clean up the resources after our test await pool.destroy() @@ -461,19 +587,83 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE } ) // TODO: Create a better test to cover `FairShareChoiceStrategy#choose` - const promises = [] + const promises = new Set() const maxMultiplier = 2 for (let i = 0; i < max * maxMultiplier; i++) { - promises.push(pool.execute()) + promises.add(pool.execute()) } await Promise.all(promises) - // if (process.platform !== 'win32') { - // expect( - // pool.workerChoiceStrategyContext.workerChoiceStrategies.get( - // pool.workerChoiceStrategyContext.workerChoiceStrategy - // ).workerLastVirtualTaskTimestamp.size - // ).toBe(pool.workerNodes.length) - // } + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: expect.any(Number), + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + expect(workerNode.tasksUsage.avgRunTime).toBeGreaterThan(0) + } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + pool.workerChoiceStrategyContext.workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(pool.workerNodes.length) + // We need to clean up the resources after our test + await pool.destroy() + }) + + it('Verify FAIR_SHARE strategy can be run in a dynamic pool with median runtime statistic', async () => { + const pool = new DynamicThreadPool( + min, + max, + './tests/worker-files/thread/testWorker.js', + { + workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE, + workerChoiceStrategyOptions: { + medRunTime: true + } + } + ) + // TODO: Create a better test to cover `FairShareChoiceStrategy#choose` + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) + } + await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: expect.any(Number), + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + expect(workerNode.tasksUsage.medRunTime).toBeGreaterThan(0) + } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + pool.workerChoiceStrategyContext.workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(pool.workerNodes.length) // We need to clean up the resources after our test await pool.destroy() }) @@ -487,23 +677,32 @@ describe('Selection strategies test suite', () => { expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( workerChoiceStrategy - ).workerLastVirtualTaskTimestamp - ).toBeDefined() + ).workersVirtualTaskEndTimestamp + ).toBeInstanceOf(Array) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(0) + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp[0] = performance.now() + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(1) pool.setWorkerChoiceStrategy(workerChoiceStrategy) - for (const workerNodeKey of pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.keys()) { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.get(workerNodeKey).start - ).toBe(0) - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.get(workerNodeKey).end - ).toBe(0) - } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp + ).toBeInstanceOf(Array) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(0) await pool.destroy() pool = new DynamicThreadPool( min, @@ -513,23 +712,32 @@ describe('Selection strategies test suite', () => { expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( workerChoiceStrategy - ).workerLastVirtualTaskTimestamp - ).toBeDefined() + ).workersVirtualTaskEndTimestamp + ).toBeInstanceOf(Array) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(0) + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp[0] = performance.now() + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(1) pool.setWorkerChoiceStrategy(workerChoiceStrategy) - for (const workerNodeKey of pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.keys()) { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.get(workerNodeKey).start - ).toBe(0) - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workerLastVirtualTaskTimestamp.get(workerNodeKey).end - ).toBe(0) - } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp + ).toBeInstanceOf(Array) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workersVirtualTaskEndTimestamp.length + ).toBe(0) // We need to clean up the resources after our test await pool.destroy() }) @@ -542,14 +750,15 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: true, + avgRunTime: true, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) await pool.destroy() pool = new DynamicThreadPool( min, @@ -558,14 +767,15 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy } ) expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().runTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime - ).toBe(true) - expect( - pool.workerChoiceStrategyContext.getRequiredStatistics().medRunTime - ).toBe(false) + pool.workerChoiceStrategyContext.getRequiredStatistics() + ).toStrictEqual({ + runTime: true, + avgRunTime: true, + medRunTime: false, + waitTime: false, + avgWaitTime: false, + medWaitTime: false + }) // We need to clean up the resources after our test await pool.destroy() }) @@ -577,16 +787,41 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN } ) // TODO: Create a better test to cover `WeightedRoundRobinWorkerChoiceStrategy#choose` - const promises = [] - for (let i = 0; i < max * 2; i++) { - promises.push(pool.execute()) + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) } await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: expect.any(Number), + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThanOrEqual(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThanOrEqual(0) + expect(workerNode.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0) + } expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( pool.workerChoiceStrategyContext.workerChoiceStrategy - ).workersTaskRunTime.size - ).toBe(pool.workerNodes.length) + ).defaultWorkerWeight + ).toBeGreaterThan(0) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + pool.workerChoiceStrategyContext.workerChoiceStrategy + ).workerVirtualTaskRunTime + ).toBeGreaterThanOrEqual(0) // We need to clean up the resources after our test await pool.destroy() }) @@ -599,22 +834,93 @@ describe('Selection strategies test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN } ) // TODO: Create a better test to cover `WeightedRoundRobinWorkerChoiceStrategy#choose` - const promises = [] - const maxMultiplier = + const promises = new Set() + const maxMultiplier = 2 + for (let i = 0; i < max * maxMultiplier; i++) { + promises.add(pool.execute()) + } + await Promise.all(promises) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: expect.any(Number), + medRunTime: 0, + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + expect(workerNode.tasksUsage.avgRunTime).toBeGreaterThan(0) + } + expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( pool.workerChoiceStrategyContext.workerChoiceStrategy - ).defaultWorkerWeight * 50 + ).defaultWorkerWeight + ).toBeGreaterThan(0) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + pool.workerChoiceStrategyContext.workerChoiceStrategy + ).workerVirtualTaskRunTime + ).toBeGreaterThanOrEqual(0) + // We need to clean up the resources after our test + await pool.destroy() + }) + + it('Verify WEIGHTED_ROUND_ROBIN strategy can be run in a dynamic pool with median runtime statistic', async () => { + const pool = new DynamicThreadPool( + min, + max, + './tests/worker-files/thread/testWorker.js', + { + workerChoiceStrategy: WorkerChoiceStrategies.WEIGHTED_ROUND_ROBIN, + workerChoiceStrategyOptions: { + medRunTime: true + } + } + ) + // TODO: Create a better test to cover `WeightedRoundRobinWorkerChoiceStrategy#choose` + const promises = new Set() + const maxMultiplier = 2 for (let i = 0; i < max * maxMultiplier; i++) { - promises.push(pool.execute()) + promises.add(pool.execute()) } await Promise.all(promises) - if (process.platform !== 'win32') { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies.get( - pool.workerChoiceStrategyContext.workerChoiceStrategy - ).workersTaskRunTime.size - ).toBe(pool.workerNodes.length) + for (const workerNode of pool.workerNodes) { + expect(workerNode.tasksUsage).toStrictEqual({ + run: expect.any(Number), + running: 0, + runTime: expect.any(Number), + runTimeHistory: expect.any(CircularArray), + avgRunTime: 0, + medRunTime: expect.any(Number), + waitTime: 0, + waitTimeHistory: expect.any(CircularArray), + avgWaitTime: 0, + medWaitTime: 0, + error: 0 + }) + expect(workerNode.tasksUsage.run).toBeGreaterThan(0) + expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(max * maxMultiplier) + expect(workerNode.tasksUsage.runTime).toBeGreaterThan(0) + expect(workerNode.tasksUsage.medRunTime).toBeGreaterThan(0) } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + pool.workerChoiceStrategyContext.workerChoiceStrategy + ).defaultWorkerWeight + ).toBeGreaterThan(0) + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + pool.workerChoiceStrategyContext.workerChoiceStrategy + ).workerVirtualTaskRunTime + ).toBeGreaterThanOrEqual(0) // We need to clean up the resources after our test await pool.destroy() }) @@ -638,7 +944,7 @@ describe('Selection strategies test suite', () => { expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( workerChoiceStrategy - ).workersTaskRunTime + ).workerVirtualTaskRunTime ).toBeDefined() pool.setWorkerChoiceStrategy(workerChoiceStrategy) expect( @@ -651,15 +957,11 @@ describe('Selection strategies test suite', () => { pool.workerChoiceStrategyContext.workerChoiceStrategy ).defaultWorkerWeight ).toBeGreaterThan(0) - for (const workerNodeKey of pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workersTaskRunTime.keys()) { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workersTaskRunTime.get(workerNodeKey).runTime - ).toBe(0) - } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workerVirtualTaskRunTime + ).toBe(0) await pool.destroy() pool = new DynamicThreadPool( min, @@ -679,7 +981,7 @@ describe('Selection strategies test suite', () => { expect( pool.workerChoiceStrategyContext.workerChoiceStrategies.get( workerChoiceStrategy - ).workersTaskRunTime + ).workerVirtualTaskRunTime ).toBeDefined() pool.setWorkerChoiceStrategy(workerChoiceStrategy) expect( @@ -692,15 +994,11 @@ describe('Selection strategies test suite', () => { pool.workerChoiceStrategyContext.workerChoiceStrategy ).defaultWorkerWeight ).toBeGreaterThan(0) - for (const workerNodeKey of pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workersTaskRunTime.keys()) { - expect( - pool.workerChoiceStrategyContext.workerChoiceStrategies - .get(pool.workerChoiceStrategyContext.workerChoiceStrategy) - .workersTaskRunTime.get(workerNodeKey).runTime - ).toBe(0) - } + expect( + pool.workerChoiceStrategyContext.workerChoiceStrategies.get( + workerChoiceStrategy + ).workerVirtualTaskRunTime + ).toBe(0) // We need to clean up the resources after our test await pool.destroy() }) @@ -714,8 +1012,6 @@ describe('Selection strategies test suite', () => { './tests/worker-files/thread/testWorker.js', { workerChoiceStrategy: 'UNKNOWN_STRATEGY' } ) - ).toThrowError( - new Error("Invalid worker choice strategy 'UNKNOWN_STRATEGY'") - ) + ).toThrowError("Invalid worker choice strategy 'UNKNOWN_STRATEGY'") }) })