X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fabstract-pool.test.mjs;h=f26efdd846b5e86ae026306105f34ca030f9d4ca;hb=146eaa19f0f0ad8b0aecfef0ad3d8552dd064f33;hp=84212a3f4f81f953fe50a718cac4f5f0edba1cdf;hpb=229e9e73566ed1960ba5be0534fb3aa2eeb115eb;p=poolifier.git diff --git a/tests/pools/abstract-pool.test.mjs b/tests/pools/abstract-pool.test.mjs index 84212a3f..f26efdd8 100644 --- a/tests/pools/abstract-pool.test.mjs +++ b/tests/pools/abstract-pool.test.mjs @@ -1,5 +1,8 @@ import { EventEmitterAsyncResource } from 'node:events' +import { dirname, join } from 'node:path' import { readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' +import { createHook, executionAsyncId } from 'node:async_hooks' import { expect } from 'expect' import { restore, stub } from 'sinon' import { @@ -19,7 +22,12 @@ import { waitPoolEvents } from '../test-utils.js' import { WorkerNode } from '../../lib/pools/worker-node.js' describe('Abstract pool test suite', () => { - const version = JSON.parse(readFileSync('./package.json', 'utf8')).version + const version = JSON.parse( + readFileSync( + join(dirname(fileURLToPath(import.meta.url)), '../..', 'package.json'), + 'utf8' + ) + ).version const numberOfWorkers = 2 class StubPoolWithIsMain extends FixedThreadPool { isMain () { @@ -31,7 +39,16 @@ describe('Abstract pool test suite', () => { restore() }) - it('Simulate pool creation from a non main thread/process', () => { + it('Verify that pool can be created and destroyed', async () => { + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.mjs' + ) + expect(pool).toBeInstanceOf(FixedThreadPool) + await pool.destroy() + }) + + it('Verify that pool cannot be created from a non main thread/process', () => { expect( () => new StubPoolWithIsMain( @@ -41,7 +58,7 @@ describe('Abstract pool test suite', () => { errorHandler: e => console.error(e) } ) - ).toThrowError( + ).toThrow( new Error( 'Cannot start a pool from a worker with the same type as the pool' ) @@ -53,30 +70,22 @@ describe('Abstract pool test suite', () => { numberOfWorkers, './tests/worker-files/thread/testWorker.mjs' ) - expect(pool.starting).toBe(false) expect(pool.started).toBe(true) + expect(pool.starting).toBe(false) + expect(pool.destroying).toBe(false) await pool.destroy() }) it('Verify that filePath is checked', () => { - const expectedError = new Error( - 'Please specify a file with a worker implementation' - ) - expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError( - expectedError - ) - expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError( - expectedError + expect(() => new FixedThreadPool(numberOfWorkers)).toThrow( + new TypeError('The worker file path must be specified') ) - expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrowError( - expectedError - ) - expect(() => new FixedThreadPool(numberOfWorkers, true)).toThrowError( - expectedError + expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrow( + new TypeError('The worker file path must be a string') ) expect( () => new FixedThreadPool(numberOfWorkers, './dummyWorker.ts') - ).toThrowError(new Error("Cannot find the worker file './dummyWorker.ts'")) + ).toThrow(new Error("Cannot find the worker file './dummyWorker.ts'")) }) it('Verify that numberOfWorkers is checked', () => { @@ -86,7 +95,7 @@ describe('Abstract pool test suite', () => { undefined, './tests/worker-files/thread/testWorker.mjs' ) - ).toThrowError( + ).toThrow( new Error( 'Cannot instantiate a pool without specifying the number of workers' ) @@ -97,7 +106,7 @@ describe('Abstract pool test suite', () => { expect( () => new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js') - ).toThrowError( + ).toThrow( new RangeError( 'Cannot instantiate a pool with a negative number of workers' ) @@ -108,13 +117,29 @@ describe('Abstract pool test suite', () => { expect( () => new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.mjs') - ).toThrowError( + ).toThrow( new TypeError( 'Cannot instantiate a pool with a non safe integer number of workers' ) ) }) + it('Verify that pool arguments number and pool type are checked', () => { + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.mjs', + undefined, + numberOfWorkers * 2 + ) + ).toThrow( + new Error( + 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization' + ) + ) + }) + it('Verify that dynamic pool sizing is checked', () => { expect( () => @@ -123,7 +148,7 @@ describe('Abstract pool test suite', () => { undefined, './tests/worker-files/cluster/testWorker.js' ) - ).toThrowError( + ).toThrow( new TypeError( 'Cannot instantiate a dynamic pool without specifying the maximum pool size' ) @@ -135,7 +160,7 @@ describe('Abstract pool test suite', () => { 1, './tests/worker-files/thread/testWorker.mjs' ) - ).toThrowError( + ).toThrow( new TypeError( 'Cannot instantiate a pool with a non safe integer number of workers' ) @@ -147,7 +172,7 @@ describe('Abstract pool test suite', () => { 0.5, './tests/worker-files/cluster/testWorker.js' ) - ).toThrowError( + ).toThrow( new TypeError( 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size' ) @@ -159,7 +184,7 @@ describe('Abstract pool test suite', () => { 1, './tests/worker-files/thread/testWorker.mjs' ) - ).toThrowError( + ).toThrow( new RangeError( 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size' ) @@ -171,7 +196,7 @@ describe('Abstract pool test suite', () => { 0, './tests/worker-files/thread/testWorker.mjs' ) - ).toThrowError( + ).toThrow( new RangeError( 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero' ) @@ -183,7 +208,7 @@ describe('Abstract pool test suite', () => { 1, './tests/worker-files/cluster/testWorker.js' ) - ).toThrowError( + ).toThrow( new RangeError( 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead' ) @@ -201,28 +226,32 @@ describe('Abstract pool test suite', () => { enableEvents: true, restartWorkerOnError: true, enableTasksQueue: false, - workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, - workerChoiceStrategyOptions: { - retries: 6, - runTime: { median: false }, - waitTime: { median: false }, - elu: { median: false } - } + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - retries: 6, + retries: + pool.info.maxSize + + Object.keys(pool.workerChoiceStrategyContext.opts.weights).length, runTime: { median: false }, waitTime: { median: false }, - elu: { median: false } + elu: { median: false }, + weights: expect.objectContaining({ + 0: expect.any(Number), + [pool.info.maxSize - 1]: expect.any(Number) + }) }) for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { - expect(workerChoiceStrategy.opts).toStrictEqual({ - retries: 6, - runTime: { median: false }, - waitTime: { median: false }, - elu: { median: false } - }) + expect(workerChoiceStrategy.opts).toStrictEqual( + expect.objectContaining({ + retries: + pool.info.maxSize + + Object.keys(workerChoiceStrategy.opts.weights).length, + runTime: { median: false }, + waitTime: { median: false }, + elu: { median: false } + }) + ) } await pool.destroy() const testHandler = () => console.info('test handler executed') @@ -255,14 +284,12 @@ describe('Abstract pool test suite', () => { concurrency: 2, size: Math.pow(numberOfWorkers, 2), taskStealing: true, - tasksStealingOnBackPressure: true + tasksStealingOnBackPressure: true, + tasksFinishedTimeout: 2000 }, workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED, workerChoiceStrategyOptions: { - retries: 6, runTime: { median: true }, - waitTime: { median: false }, - elu: { median: false }, weights: { 0: 300, 1: 200 } }, onlineHandler: testHandler, @@ -271,7 +298,9 @@ describe('Abstract pool test suite', () => { exitHandler: testHandler }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - retries: 6, + retries: + pool.info.maxSize + + Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length, runTime: { median: true }, waitTime: { median: false }, elu: { median: false }, @@ -280,7 +309,9 @@ describe('Abstract pool test suite', () => { for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { expect(workerChoiceStrategy.opts).toStrictEqual({ - retries: 6, + retries: + pool.info.maxSize + + Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length, runTime: { median: true }, waitTime: { median: false }, elu: { median: false }, @@ -290,7 +321,7 @@ describe('Abstract pool test suite', () => { await pool.destroy() }) - it('Verify that pool options are validated', async () => { + it('Verify that pool options are validated', () => { expect( () => new FixedThreadPool( @@ -300,41 +331,7 @@ describe('Abstract pool test suite', () => { workerChoiceStrategy: 'invalidStrategy' } ) - ).toThrowError( - new Error("Invalid worker choice strategy 'invalidStrategy'") - ) - expect( - () => - new FixedThreadPool( - numberOfWorkers, - './tests/worker-files/thread/testWorker.mjs', - { - workerChoiceStrategyOptions: { - retries: 'invalidChoiceRetries' - } - } - ) - ).toThrowError( - new TypeError( - 'Invalid worker choice strategy options: retries must be an integer' - ) - ) - expect( - () => - new FixedThreadPool( - numberOfWorkers, - './tests/worker-files/thread/testWorker.mjs', - { - workerChoiceStrategyOptions: { - retries: -1 - } - } - ) - ).toThrowError( - new RangeError( - "Invalid worker choice strategy options: retries '-1' must be greater or equal than zero" - ) - ) + ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'")) expect( () => new FixedThreadPool( @@ -344,7 +341,7 @@ describe('Abstract pool test suite', () => { workerChoiceStrategyOptions: { weights: {} } } ) - ).toThrowError( + ).toThrow( new Error( 'Invalid worker choice strategy options: must have a weight for each worker node' ) @@ -358,7 +355,7 @@ describe('Abstract pool test suite', () => { workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' } } ) - ).toThrowError( + ).toThrow( new Error( "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" ) @@ -373,7 +370,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: 'invalidTasksQueueOptions' } ) - ).toThrowError( + ).toThrow( new TypeError('Invalid tasks queue options: must be a plain object') ) expect( @@ -386,7 +383,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { concurrency: 0 } } ) - ).toThrowError( + ).toThrow( new RangeError( 'Invalid worker node tasks concurrency: 0 is a negative integer or zero' ) @@ -401,7 +398,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { concurrency: -1 } } ) - ).toThrowError( + ).toThrow( new RangeError( 'Invalid worker node tasks concurrency: -1 is a negative integer or zero' ) @@ -416,7 +413,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { concurrency: 0.2 } } ) - ).toThrowError( + ).toThrow( new TypeError('Invalid worker node tasks concurrency: must be an integer') ) expect( @@ -429,7 +426,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { size: 0 } } ) - ).toThrowError( + ).toThrow( new RangeError( 'Invalid worker node tasks queue size: 0 is a negative integer or zero' ) @@ -444,7 +441,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { size: -1 } } ) - ).toThrowError( + ).toThrow( new RangeError( 'Invalid worker node tasks queue size: -1 is a negative integer or zero' ) @@ -459,7 +456,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { size: 0.2 } } ) - ).toThrowError( + ).toThrow( new TypeError('Invalid worker node tasks queue size: must be an integer') ) }) @@ -470,26 +467,31 @@ describe('Abstract pool test suite', () => { './tests/worker-files/thread/testWorker.mjs', { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE } ) - expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - retries: 6, - runTime: { median: false }, - waitTime: { median: false }, - elu: { median: false } - }) + expect(pool.opts.workerChoiceStrategyOptions).toBeUndefined() expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - retries: 6, + retries: + pool.info.maxSize + + Object.keys(pool.workerChoiceStrategyContext.opts.weights).length, runTime: { median: false }, waitTime: { median: false }, - elu: { median: false } + elu: { median: false }, + weights: expect.objectContaining({ + 0: expect.any(Number), + [pool.info.maxSize - 1]: expect.any(Number) + }) }) for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { - expect(workerChoiceStrategy.opts).toStrictEqual({ - retries: 6, - runTime: { median: false }, - waitTime: { median: false }, - elu: { median: false } - }) + expect(workerChoiceStrategy.opts).toStrictEqual( + expect.objectContaining({ + retries: + pool.info.maxSize + + Object.keys(workerChoiceStrategy.opts.weights).length, + runTime: { median: false }, + waitTime: { median: false }, + elu: { median: false } + }) + ) } expect( pool.workerChoiceStrategyContext.getTaskStatisticsRequirements() @@ -515,25 +517,33 @@ describe('Abstract pool test suite', () => { elu: { median: true } }) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - retries: 6, runTime: { median: true }, - waitTime: { median: false }, elu: { median: true } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - retries: 6, + retries: + pool.info.maxSize + + Object.keys(pool.workerChoiceStrategyContext.opts.weights).length, runTime: { median: true }, waitTime: { median: false }, - elu: { median: true } + elu: { median: true }, + weights: expect.objectContaining({ + 0: expect.any(Number), + [pool.info.maxSize - 1]: expect.any(Number) + }) }) for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { - expect(workerChoiceStrategy.opts).toStrictEqual({ - retries: 6, - runTime: { median: true }, - waitTime: { median: false }, - elu: { median: true } - }) + expect(workerChoiceStrategy.opts).toStrictEqual( + expect.objectContaining({ + retries: + pool.info.maxSize + + Object.keys(workerChoiceStrategy.opts.weights).length, + runTime: { median: true }, + waitTime: { median: false }, + elu: { median: true } + }) + ) } expect( pool.workerChoiceStrategyContext.getTaskStatisticsRequirements() @@ -559,25 +569,33 @@ describe('Abstract pool test suite', () => { elu: { median: false } }) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - retries: 6, runTime: { median: false }, - waitTime: { median: false }, elu: { median: false } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - retries: 6, + retries: + pool.info.maxSize + + Object.keys(pool.workerChoiceStrategyContext.opts.weights).length, runTime: { median: false }, waitTime: { median: false }, - elu: { median: false } + elu: { median: false }, + weights: expect.objectContaining({ + 0: expect.any(Number), + [pool.info.maxSize - 1]: expect.any(Number) + }) }) for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { - expect(workerChoiceStrategy.opts).toStrictEqual({ - retries: 6, - runTime: { median: false }, - waitTime: { median: false }, - elu: { median: false } - }) + expect(workerChoiceStrategy.opts).toStrictEqual( + expect.objectContaining({ + retries: + pool.info.maxSize + + Object.keys(workerChoiceStrategy.opts.weights).length, + runTime: { median: false }, + waitTime: { median: false }, + elu: { median: false } + }) + ) } expect( pool.workerChoiceStrategyContext.getTaskStatisticsRequirements() @@ -600,37 +618,19 @@ describe('Abstract pool test suite', () => { }) expect(() => pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions') - ).toThrowError( + ).toThrow( new TypeError( 'Invalid worker choice strategy options: must be a plain object' ) ) - expect(() => - pool.setWorkerChoiceStrategyOptions({ - retries: 'invalidChoiceRetries' - }) - ).toThrowError( - new TypeError( - 'Invalid worker choice strategy options: retries must be an integer' - ) - ) - expect(() => - pool.setWorkerChoiceStrategyOptions({ retries: -1 }) - ).toThrowError( - new RangeError( - "Invalid worker choice strategy options: retries '-1' must be greater or equal than zero" - ) - ) - expect(() => - pool.setWorkerChoiceStrategyOptions({ weights: {} }) - ).toThrowError( + expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow( new Error( 'Invalid worker choice strategy options: must have a weight for each worker node' ) ) expect(() => pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' }) - ).toThrowError( + ).toThrow( new Error( "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" ) @@ -645,41 +645,27 @@ describe('Abstract pool test suite', () => { ) expect(pool.opts.enableTasksQueue).toBe(false) expect(pool.opts.tasksQueueOptions).toBeUndefined() - for (const workerNode of pool.workerNodes) { - expect(workerNode.onEmptyQueue).toBeUndefined() - expect(workerNode.onBackPressure).toBeUndefined() - } pool.enableTasksQueue(true) expect(pool.opts.enableTasksQueue).toBe(true) expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1, size: Math.pow(numberOfWorkers, 2), taskStealing: true, - tasksStealingOnBackPressure: true + tasksStealingOnBackPressure: true, + tasksFinishedTimeout: 2000 }) - for (const workerNode of pool.workerNodes) { - expect(workerNode.onEmptyQueue).toBeInstanceOf(Function) - expect(workerNode.onBackPressure).toBeInstanceOf(Function) - } pool.enableTasksQueue(true, { concurrency: 2 }) expect(pool.opts.enableTasksQueue).toBe(true) expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2, size: Math.pow(numberOfWorkers, 2), taskStealing: true, - tasksStealingOnBackPressure: true + tasksStealingOnBackPressure: true, + tasksFinishedTimeout: 2000 }) - for (const workerNode of pool.workerNodes) { - expect(workerNode.onEmptyQueue).toBeInstanceOf(Function) - expect(workerNode.onBackPressure).toBeInstanceOf(Function) - } pool.enableTasksQueue(false) expect(pool.opts.enableTasksQueue).toBe(false) expect(pool.opts.tasksQueueOptions).toBeUndefined() - for (const workerNode of pool.workerNodes) { - expect(workerNode.onEmptyQueue).toBeUndefined() - expect(workerNode.onBackPressure).toBeUndefined() - } await pool.destroy() }) @@ -693,33 +679,32 @@ describe('Abstract pool test suite', () => { concurrency: 1, size: Math.pow(numberOfWorkers, 2), taskStealing: true, - tasksStealingOnBackPressure: true + tasksStealingOnBackPressure: true, + tasksFinishedTimeout: 2000 }) for (const workerNode of pool.workerNodes) { expect(workerNode.tasksQueueBackPressureSize).toBe( pool.opts.tasksQueueOptions.size ) - expect(workerNode.onEmptyQueue).toBeInstanceOf(Function) - expect(workerNode.onBackPressure).toBeInstanceOf(Function) } pool.setTasksQueueOptions({ concurrency: 2, size: 2, taskStealing: false, - tasksStealingOnBackPressure: false + tasksStealingOnBackPressure: false, + tasksFinishedTimeout: 3000 }) expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2, size: 2, taskStealing: false, - tasksStealingOnBackPressure: false + tasksStealingOnBackPressure: false, + tasksFinishedTimeout: 3000 }) for (const workerNode of pool.workerNodes) { expect(workerNode.tasksQueueBackPressureSize).toBe( pool.opts.tasksQueueOptions.size ) - expect(workerNode.onEmptyQueue).toBeUndefined() - expect(workerNode.onBackPressure).toBeUndefined() } pool.setTasksQueueOptions({ concurrency: 1, @@ -730,44 +715,41 @@ describe('Abstract pool test suite', () => { concurrency: 1, size: Math.pow(numberOfWorkers, 2), taskStealing: true, - tasksStealingOnBackPressure: true + tasksStealingOnBackPressure: true, + tasksFinishedTimeout: 2000 }) for (const workerNode of pool.workerNodes) { expect(workerNode.tasksQueueBackPressureSize).toBe( pool.opts.tasksQueueOptions.size ) - expect(workerNode.onEmptyQueue).toBeInstanceOf(Function) - expect(workerNode.onBackPressure).toBeInstanceOf(Function) } - expect(() => - pool.setTasksQueueOptions('invalidTasksQueueOptions') - ).toThrowError( + expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions')).toThrow( new TypeError('Invalid tasks queue options: must be a plain object') ) - expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError( + expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrow( new RangeError( 'Invalid worker node tasks concurrency: 0 is a negative integer or zero' ) ) - expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError( + expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrow( new RangeError( 'Invalid worker node tasks concurrency: -1 is a negative integer or zero' ) ) - expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError( + expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrow( new TypeError('Invalid worker node tasks concurrency: must be an integer') ) - expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError( + expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrow( new RangeError( 'Invalid worker node tasks queue size: 0 is a negative integer or zero' ) ) - expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError( + expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrow( new RangeError( 'Invalid worker node tasks queue size: -1 is a negative integer or zero' ) ) - expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError( + expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrow( new TypeError('Invalid worker node tasks queue size: must be an integer') ) await pool.destroy() @@ -832,6 +814,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -912,6 +895,24 @@ describe('Abstract pool test suite', () => { await pool.destroy() }) + it('Verify that pool statuses are checked at start or destroy', async () => { + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.mjs' + ) + expect(pool.info.started).toBe(true) + expect(pool.info.ready).toBe(true) + expect(() => pool.start()).toThrow( + new Error('Cannot start an already started pool') + ) + await pool.destroy() + expect(pool.info.started).toBe(false) + expect(pool.info.ready).toBe(false) + await expect(pool.destroy()).rejects.toThrow( + new Error('Cannot destroy an already destroyed pool') + ) + }) + it('Verify that pool can be started after initialization', async () => { const pool = new FixedClusterPool( numberOfWorkers, @@ -922,13 +923,16 @@ describe('Abstract pool test suite', () => { ) expect(pool.info.started).toBe(false) expect(pool.info.ready).toBe(false) + expect(pool.readyEventEmitted).toBe(false) expect(pool.workerNodes).toStrictEqual([]) - await expect(pool.execute()).rejects.toThrowError( + await expect(pool.execute()).rejects.toThrow( new Error('Cannot execute a task on not started pool') ) pool.start() expect(pool.info.started).toBe(true) expect(pool.info.ready).toBe(true) + await waitPoolEvents(pool, PoolEvents.ready, 1) + expect(pool.readyEventEmitted).toBe(true) expect(pool.workerNodes.length).toBe(numberOfWorkers) for (const workerNode of pool.workerNodes) { expect(workerNode).toBeInstanceOf(WorkerNode) @@ -941,20 +945,20 @@ describe('Abstract pool test suite', () => { numberOfWorkers, './tests/worker-files/cluster/testWorker.js' ) - await expect(pool.execute(undefined, 0)).rejects.toThrowError( + await expect(pool.execute(undefined, 0)).rejects.toThrow( new TypeError('name argument must be a string') ) - await expect(pool.execute(undefined, '')).rejects.toThrowError( + await expect(pool.execute(undefined, '')).rejects.toThrow( new TypeError('name argument must not be an empty string') ) - await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError( + await expect(pool.execute(undefined, undefined, {})).rejects.toThrow( new TypeError('transferList argument must be an array') ) await expect(pool.execute(undefined, 'unknown')).rejects.toBe( "Task function 'unknown' not found" ) await pool.destroy() - await expect(pool.execute()).rejects.toThrowError( + await expect(pool.execute()).rejects.toThrow( new Error('Cannot execute a task on not started pool') ) }) @@ -976,6 +980,7 @@ describe('Abstract pool test suite', () => { executing: maxMultiplier, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1003,6 +1008,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1044,6 +1050,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1079,6 +1086,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1262,7 +1270,101 @@ describe('Abstract pool test suite', () => { stolenTasks: expect.any(Number), failedTasks: expect.any(Number) }) - expect(pool.hasBackPressure.called).toBe(true) + expect(pool.hasBackPressure.callCount).toBe(5) + await pool.destroy() + }) + + it('Verify that destroy() waits for queued tasks to finish', async () => { + const tasksFinishedTimeout = 2500 + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/asyncWorker.mjs', + { + enableTasksQueue: true, + tasksQueueOptions: { tasksFinishedTimeout } + } + ) + const maxMultiplier = 4 + let tasksFinished = 0 + for (const workerNode of pool.workerNodes) { + workerNode.on('taskFinished', () => { + ++tasksFinished + }) + } + for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) { + pool.execute() + } + expect(pool.info.queuedTasks).toBeGreaterThan(0) + const startTime = performance.now() + await pool.destroy() + const elapsedTime = performance.now() - startTime + expect(tasksFinished).toBe(numberOfWorkers * maxMultiplier) + expect(elapsedTime).toBeGreaterThanOrEqual(2000) + expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 100) + }) + + it('Verify that destroy() waits until the tasks finished timeout is reached', async () => { + const tasksFinishedTimeout = 1000 + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/asyncWorker.mjs', + { + enableTasksQueue: true, + tasksQueueOptions: { tasksFinishedTimeout } + } + ) + const maxMultiplier = 4 + let tasksFinished = 0 + for (const workerNode of pool.workerNodes) { + workerNode.on('taskFinished', () => { + ++tasksFinished + }) + } + for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) { + pool.execute() + } + expect(pool.info.queuedTasks).toBeGreaterThan(0) + const startTime = performance.now() + await pool.destroy() + const elapsedTime = performance.now() - startTime + expect(tasksFinished).toBe(0) + expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 600) + }) + + it('Verify that pool asynchronous resource track tasks execution', async () => { + let taskAsyncId + let initCalls = 0 + let beforeCalls = 0 + let afterCalls = 0 + let resolveCalls = 0 + const hook = createHook({ + init (asyncId, type) { + if (type === 'poolifier:task') { + initCalls++ + taskAsyncId = asyncId + } + }, + before (asyncId) { + if (asyncId === taskAsyncId) beforeCalls++ + }, + after (asyncId) { + if (asyncId === taskAsyncId) afterCalls++ + }, + promiseResolve () { + if (executionAsyncId() === taskAsyncId) resolveCalls++ + } + }) + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.mjs' + ) + hook.enable() + await pool.execute() + hook.disable() + expect(initCalls).toBe(1) + expect(beforeCalls).toBe(1) + expect(afterCalls).toBe(1) + expect(resolveCalls).toBe(1) await pool.destroy() }) @@ -1305,18 +1407,18 @@ describe('Abstract pool test suite', () => { await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1) await expect( dynamicThreadPool.addTaskFunction(0, () => {}) - ).rejects.toThrowError(new TypeError('name argument must be a string')) + ).rejects.toThrow(new TypeError('name argument must be a string')) await expect( dynamicThreadPool.addTaskFunction('', () => {}) - ).rejects.toThrowError( + ).rejects.toThrow( new TypeError('name argument must not be an empty string') ) - await expect( - dynamicThreadPool.addTaskFunction('test', 0) - ).rejects.toThrowError(new TypeError('fn argument must be a function')) - await expect( - dynamicThreadPool.addTaskFunction('test', '') - ).rejects.toThrowError(new TypeError('fn argument must be a function')) + await expect(dynamicThreadPool.addTaskFunction('test', 0)).rejects.toThrow( + new TypeError('fn argument must be a function') + ) + await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow( + new TypeError('fn argument must be a function') + ) expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([ DEFAULT_TASK_NAME, 'test' @@ -1345,6 +1447,7 @@ describe('Abstract pool test suite', () => { executed: expect.any(Number), executing: 0, queued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1378,9 +1481,7 @@ describe('Abstract pool test suite', () => { DEFAULT_TASK_NAME, 'test' ]) - await expect( - dynamicThreadPool.removeTaskFunction('test') - ).rejects.toThrowError( + await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow( new Error('Cannot remove a task function not handled on the pool side') ) const echoTaskFunction = data => { @@ -1443,25 +1544,24 @@ describe('Abstract pool test suite', () => { './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs' ) await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1) - await expect( - dynamicThreadPool.setDefaultTaskFunction(0) - ).rejects.toThrowError( + const workerId = dynamicThreadPool.workerNodes[0].info.id + await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow( new Error( - "Task function operation 'default' failed on worker 31 with error: 'TypeError: name parameter is not a string'" + `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'` ) ) await expect( dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME) - ).rejects.toThrowError( + ).rejects.toThrow( new Error( - "Task function operation 'default' failed on worker 31 with error: 'Error: Cannot set the default task function reserved name as the default task function'" + `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function reserved name as the default task function'` ) ) await expect( dynamicThreadPool.setDefaultTaskFunction('unknown') - ).rejects.toThrowError( + ).rejects.toThrow( new Error( - "Task function operation 'default' failed on worker 31 with error: 'Error: Cannot set the default task function to a non-existing task function'" + `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function to a non-existing task function'` ) ) expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([ @@ -1488,6 +1588,7 @@ describe('Abstract pool test suite', () => { 'jsonIntegerSerialization', 'factorial' ]) + await dynamicThreadPool.destroy() }) it('Verify that multiple task functions worker is working', async () => { @@ -1522,6 +1623,7 @@ describe('Abstract pool test suite', () => { executing: 0, failed: 0, queued: 0, + sequentiallyStolen: 0, stolen: 0 }, runTime: { @@ -1564,6 +1666,11 @@ describe('Abstract pool test suite', () => { await expect( pool.sendKillMessageToWorker(workerNodeKey) ).resolves.toBeUndefined() + await expect( + pool.sendKillMessageToWorker(numberOfWorkers) + ).rejects.toStrictEqual( + new Error(`Invalid worker node key '${numberOfWorkers}'`) + ) await pool.destroy() })