X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fabstract%2Fabstract-pool.test.js;h=bedb1c6c42a9121697990121185c049c26ed87a4;hb=8a11c8fd0b8be823629d0475ee16362b0bb31d25;hp=ec552e986003d859f38fe0557d6d8727c23d7ceb;hpb=a1763c54c962c69b5e02a30c0909724986495fcd;p=poolifier.git diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js index ec552e98..bedb1c6c 100644 --- a/tests/pools/abstract/abstract-pool.test.js +++ b/tests/pools/abstract/abstract-pool.test.js @@ -1,4 +1,6 @@ +const { EventEmitter } = require('node:events') const { expect } = require('expect') +const sinon = require('sinon') const { DynamicClusterPool, DynamicThreadPool, @@ -10,7 +12,8 @@ const { WorkerTypes } = require('../../../lib') const { CircularArray } = require('../../../lib/circular-array') -const { Queue } = require('../../../lib/queue') +const { Deque } = require('../../../lib/deque') +const { DEFAULT_TASK_NAME } = require('../../../lib/utils') const { version } = require('../../../package.json') const { waitPoolEvents } = require('../../test-utils') @@ -22,6 +25,10 @@ describe('Abstract pool test suite', () => { } } + afterEach(() => { + sinon.restore() + }) + it('Simulate pool creation from a non main thread/process', () => { expect( () => @@ -29,12 +36,24 @@ describe('Abstract pool test suite', () => { numberOfWorkers, './tests/worker-files/thread/testWorker.js', { - errorHandler: (e) => console.error(e) + errorHandler: e => console.error(e) } ) ).toThrowError( - 'Cannot start a pool from a worker with the same type as the pool' + new Error( + 'Cannot start a pool from a worker with the same type as the pool' + ) + ) + }) + + it('Verify that pool statuses properties are set', async () => { + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js' ) + expect(pool.starting).toBe(false) + expect(pool.started).toBe(true) + await pool.destroy() }) it('Verify that filePath is checked', () => { @@ -60,7 +79,9 @@ describe('Abstract pool test suite', () => { it('Verify that numberOfWorkers is checked', () => { expect(() => new FixedThreadPool()).toThrowError( - 'Cannot instantiate a pool without specifying the number of workers' + new Error( + 'Cannot instantiate a pool without specifying the number of workers' + ) ) }) @@ -133,22 +154,22 @@ describe('Abstract pool test suite', () => { ) expect( () => - new DynamicClusterPool( - 1, - 1, - './tests/worker-files/cluster/testWorker.js' - ) + new DynamicThreadPool(0, 0, './tests/worker-files/thread/testWorker.js') ).toThrowError( new RangeError( - 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead' + 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero' ) ) expect( () => - new DynamicThreadPool(0, 0, './tests/worker-files/thread/testWorker.js') + new DynamicClusterPool( + 1, + 1, + './tests/worker-files/cluster/testWorker.js' + ) ).toThrowError( new RangeError( - 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero' + 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead' ) ) }) @@ -158,7 +179,7 @@ describe('Abstract pool test suite', () => { numberOfWorkers, './tests/worker-files/thread/testWorker.js' ) - expect(pool.emitter).toBeDefined() + expect(pool.emitter).toBeInstanceOf(EventEmitter) expect(pool.opts.enableEvents).toBe(true) expect(pool.opts.restartWorkerOnError).toBe(true) expect(pool.opts.enableTasksQueue).toBe(false) @@ -167,17 +188,26 @@ describe('Abstract pool test suite', () => { WorkerChoiceStrategies.ROUND_ROBIN ) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } }) + for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext + .workerChoiceStrategies) { + expect(workerChoiceStrategy.opts).toStrictEqual({ + retries: 6, + runTime: { median: false }, + waitTime: { median: false }, + elu: { median: false } + }) + } expect(pool.opts.messageHandler).toBeUndefined() expect(pool.opts.errorHandler).toBeUndefined() expect(pool.opts.onlineHandler).toBeUndefined() @@ -207,24 +237,37 @@ describe('Abstract pool test suite', () => { expect(pool.opts.enableEvents).toBe(false) expect(pool.opts.restartWorkerOnError).toBe(false) expect(pool.opts.enableTasksQueue).toBe(true) - expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 }) + expect(pool.opts.tasksQueueOptions).toStrictEqual({ + concurrency: 2, + size: 4 + }) expect(pool.opts.workerChoiceStrategy).toBe( WorkerChoiceStrategies.LEAST_USED ) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: true }, waitTime: { median: false }, elu: { median: false }, weights: { 0: 300, 1: 200 } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: true }, waitTime: { median: false }, elu: { median: false }, weights: { 0: 300, 1: 200 } }) + for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext + .workerChoiceStrategies) { + expect(workerChoiceStrategy.opts).toStrictEqual({ + retries: 6, + runTime: { median: true }, + waitTime: { median: false }, + elu: { median: false }, + weights: { 0: 300, 1: 200 } + }) + } expect(pool.opts.messageHandler).toStrictEqual(testHandler) expect(pool.opts.errorHandler).toStrictEqual(testHandler) expect(pool.opts.onlineHandler).toStrictEqual(testHandler) @@ -242,7 +285,41 @@ describe('Abstract pool test suite', () => { workerChoiceStrategy: 'invalidStrategy' } ) - ).toThrowError("Invalid worker choice strategy 'invalidStrategy'") + ).toThrowError( + new Error("Invalid worker choice strategy 'invalidStrategy'") + ) + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + 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.js', + { + workerChoiceStrategyOptions: { + retries: -1 + } + } + ) + ).toThrowError( + new RangeError( + "Invalid worker choice strategy options: retries '-1' must be greater or equal than zero" + ) + ) expect( () => new FixedThreadPool( @@ -253,7 +330,9 @@ describe('Abstract pool test suite', () => { } ) ).toThrowError( - 'Invalid worker choice strategy options: must have a weight for each worker node' + new Error( + 'Invalid worker choice strategy options: must have a weight for each worker node' + ) ) expect( () => @@ -265,7 +344,22 @@ describe('Abstract pool test suite', () => { } ) ).toThrowError( - "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" + new Error( + "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" + ) + ) + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + enableTasksQueue: true, + tasksQueueOptions: 'invalidTasksQueueOptions' + } + ) + ).toThrowError( + new TypeError('Invalid tasks queue options: must be a plain object') ) expect( () => @@ -277,7 +371,11 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { concurrency: 0 } } ) - ).toThrowError("Invalid worker tasks concurrency '0'") + ).toThrowError( + new RangeError( + 'Invalid worker node tasks concurrency: 0 is a negative integer or zero' + ) + ) expect( () => new FixedThreadPool( @@ -285,10 +383,14 @@ describe('Abstract pool test suite', () => { './tests/worker-files/thread/testWorker.js', { enableTasksQueue: true, - tasksQueueOptions: 'invalidTasksQueueOptions' + tasksQueueOptions: { concurrency: -1 } } ) - ).toThrowError('Invalid tasks queue options: must be a plain object') + ).toThrowError( + new RangeError( + 'Invalid worker node tasks concurrency: -1 is a negative integer or zero' + ) + ) expect( () => new FixedThreadPool( @@ -299,7 +401,67 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { concurrency: 0.2 } } ) - ).toThrowError('Invalid worker tasks concurrency: must be an integer') + ).toThrowError( + new TypeError('Invalid worker node tasks concurrency: must be an integer') + ) + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + enableTasksQueue: true, + tasksQueueOptions: { queueMaxSize: 2 } + } + ) + ).toThrowError( + new Error( + 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead' + ) + ) + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + enableTasksQueue: true, + tasksQueueOptions: { size: 0 } + } + ) + ).toThrowError( + new RangeError( + 'Invalid worker node tasks queue size: 0 is a negative integer or zero' + ) + ) + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + enableTasksQueue: true, + tasksQueueOptions: { size: -1 } + } + ) + ).toThrowError( + new RangeError( + 'Invalid worker node tasks queue size: -1 is a negative integer or zero' + ) + ) + expect( + () => + new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + enableTasksQueue: true, + tasksQueueOptions: { size: 0.2 } + } + ) + ).toThrowError( + new TypeError('Invalid worker node tasks queue size: must be an integer') + ) }) it('Verify that pool worker choice strategy options can be set', async () => { @@ -309,13 +471,13 @@ describe('Abstract pool test suite', () => { { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE } ) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } @@ -323,7 +485,7 @@ describe('Abstract pool test suite', () => { for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { expect(workerChoiceStrategy.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } @@ -353,13 +515,13 @@ describe('Abstract pool test suite', () => { elu: { median: true } }) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: true }, waitTime: { median: false }, elu: { median: true } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: true }, waitTime: { median: false }, elu: { median: true } @@ -367,7 +529,7 @@ describe('Abstract pool test suite', () => { for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { expect(workerChoiceStrategy.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: true }, waitTime: { median: false }, elu: { median: true } @@ -397,13 +559,13 @@ describe('Abstract pool test suite', () => { elu: { median: false } }) expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } }) expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } @@ -411,7 +573,7 @@ describe('Abstract pool test suite', () => { for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext .workerChoiceStrategies) { expect(workerChoiceStrategy.opts).toStrictEqual({ - choiceRetries: 6, + retries: 6, runTime: { median: false }, waitTime: { median: false }, elu: { median: false } @@ -439,17 +601,39 @@ describe('Abstract pool test suite', () => { expect(() => pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions') ).toThrowError( - 'Invalid worker choice strategy options: must be a plain object' + 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( - 'Invalid worker choice strategy options: must have a weight for each worker node' + new Error( + 'Invalid worker choice strategy options: must have a weight for each worker node' + ) ) expect(() => pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' }) ).toThrowError( - "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" + new Error( + "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" + ) ) await pool.destroy() }) @@ -463,10 +647,16 @@ describe('Abstract pool test suite', () => { expect(pool.opts.tasksQueueOptions).toBeUndefined() pool.enableTasksQueue(true) expect(pool.opts.enableTasksQueue).toBe(true) - expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 }) + expect(pool.opts.tasksQueueOptions).toStrictEqual({ + concurrency: 1, + size: 4 + }) pool.enableTasksQueue(true, { concurrency: 2 }) expect(pool.opts.enableTasksQueue).toBe(true) - expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 }) + expect(pool.opts.tasksQueueOptions).toStrictEqual({ + concurrency: 2, + size: 4 + }) pool.enableTasksQueue(false) expect(pool.opts.enableTasksQueue).toBe(false) expect(pool.opts.tasksQueueOptions).toBeUndefined() @@ -479,17 +669,50 @@ describe('Abstract pool test suite', () => { './tests/worker-files/thread/testWorker.js', { enableTasksQueue: true } ) - expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 }) + expect(pool.opts.tasksQueueOptions).toStrictEqual({ + concurrency: 1, + size: 4 + }) pool.setTasksQueueOptions({ concurrency: 2 }) - expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 }) + expect(pool.opts.tasksQueueOptions).toStrictEqual({ + concurrency: 2, + size: 4 + }) expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions') - ).toThrowError('Invalid tasks queue options: must be a plain object') + ).toThrowError( + new TypeError('Invalid tasks queue options: must be a plain object') + ) expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError( - "Invalid worker tasks concurrency '0'" + new RangeError( + 'Invalid worker node tasks concurrency: 0 is a negative integer or zero' + ) + ) + expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError( + new RangeError( + 'Invalid worker node tasks concurrency: -1 is a negative integer or zero' + ) ) expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError( - 'Invalid worker tasks concurrency: must be an integer' + new TypeError('Invalid worker node tasks concurrency: must be an integer') + ) + expect(() => pool.setTasksQueueOptions({ queueMaxSize: 2 })).toThrowError( + new Error( + 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead' + ) + ) + expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError( + new RangeError( + 'Invalid worker node tasks queue size: 0 is a negative integer or zero' + ) + ) + expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError( + new RangeError( + 'Invalid worker node tasks queue size: -1 is a negative integer or zero' + ) + ) + expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError( + new TypeError('Invalid worker node tasks queue size: must be an integer') ) await pool.destroy() }) @@ -550,20 +773,21 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + stolen: 0, failed: 0 }, runTime: { - history: expect.any(CircularArray) + history: new CircularArray() }, waitTime: { - history: expect.any(CircularArray) + history: new CircularArray() }, elu: { idle: { - history: expect.any(CircularArray) + history: new CircularArray() }, active: { - history: expect.any(CircularArray) + history: new CircularArray() } } }) @@ -578,7 +802,7 @@ describe('Abstract pool test suite', () => { ) for (const workerNode of pool.workerNodes) { expect(workerNode.tasksQueue).toBeDefined() - expect(workerNode.tasksQueue).toBeInstanceOf(Queue) + expect(workerNode.tasksQueue).toBeInstanceOf(Deque) expect(workerNode.tasksQueue.size).toBe(0) expect(workerNode.tasksQueue.maxSize).toBe(0) } @@ -590,10 +814,11 @@ describe('Abstract pool test suite', () => { ) for (const workerNode of pool.workerNodes) { expect(workerNode.tasksQueue).toBeDefined() - expect(workerNode.tasksQueue).toBeInstanceOf(Queue) + expect(workerNode.tasksQueue).toBeInstanceOf(Deque) expect(workerNode.tasksQueue.size).toBe(0) expect(workerNode.tasksQueue.maxSize).toBe(0) } + await pool.destroy() }) it('Verify that pool worker info are initialized', async () => { @@ -623,6 +848,30 @@ describe('Abstract pool test suite', () => { ready: true }) } + await pool.destroy() + }) + + it('Verify that pool execute() arguments are checked', async () => { + const pool = new FixedClusterPool( + numberOfWorkers, + './tests/worker-files/cluster/testWorker.js' + ) + await expect(pool.execute(undefined, 0)).rejects.toThrowError( + new TypeError('name argument must be a string') + ) + await expect(pool.execute(undefined, '')).rejects.toThrowError( + new TypeError('name argument must not be an empty string') + ) + await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError( + 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(undefined, undefined, {})).rejects.toThrowError( + new Error('Cannot execute a task on destroyed pool') + ) }) it('Verify that pool worker tasks usage are computed', async () => { @@ -642,6 +891,7 @@ describe('Abstract pool test suite', () => { executing: maxMultiplier, queued: 0, maxQueued: 0, + stolen: 0, failed: 0 }, runTime: { @@ -668,6 +918,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + stolen: 0, failed: 0 }, runTime: { @@ -708,6 +959,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + stolen: 0, failed: 0 }, runTime: { @@ -726,7 +978,9 @@ describe('Abstract pool test suite', () => { } }) expect(workerNode.usage.tasks.executed).toBeGreaterThan(0) - expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(maxMultiplier) + expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual( + numberOfWorkers * maxMultiplier + ) expect(workerNode.usage.runTime.history.length).toBe(0) expect(workerNode.usage.waitTime.history.length).toBe(0) expect(workerNode.usage.elu.idle.history.length).toBe(0) @@ -740,6 +994,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + stolen: 0, failed: 0 }, runTime: { @@ -773,7 +1028,7 @@ describe('Abstract pool test suite', () => { ) let poolInfo let poolReady = 0 - pool.emitter.on(PoolEvents.ready, (info) => { + pool.emitter.on(PoolEvents.ready, info => { ++poolReady poolInfo = info }) @@ -805,7 +1060,7 @@ describe('Abstract pool test suite', () => { const promises = new Set() let poolBusy = 0 let poolInfo - pool.emitter.on(PoolEvents.busy, (info) => { + pool.emitter.on(PoolEvents.busy, info => { ++poolBusy poolInfo = info }) @@ -843,7 +1098,7 @@ describe('Abstract pool test suite', () => { const promises = new Set() let poolFull = 0 let poolInfo - pool.emitter.on(PoolEvents.full, (info) => { + pool.emitter.on(PoolEvents.full, info => { ++poolFull poolInfo = info }) @@ -851,9 +1106,7 @@ describe('Abstract pool test suite', () => { promises.add(pool.execute()) } await Promise.all(promises) - // The `full` event is triggered when the number of submitted tasks at once reach the maximum number of workers in the dynamic pool. - // So in total numberOfWorkers * 2 - 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool with min = (max = numberOfWorkers) / 2. - expect(poolFull).toBe(numberOfWorkers * 2 - 1) + expect(poolFull).toBe(1) expect(poolInfo).toStrictEqual({ version, type: PoolTypes.dynamic, @@ -872,6 +1125,50 @@ describe('Abstract pool test suite', () => { await pool.destroy() }) + it("Verify that pool event emitter 'backPressure' event can register a callback", async () => { + const pool = new FixedThreadPool( + numberOfWorkers, + './tests/worker-files/thread/testWorker.js', + { + enableTasksQueue: true + } + ) + sinon.stub(pool, 'hasBackPressure').returns(true) + const promises = new Set() + let poolBackPressure = 0 + let poolInfo + pool.emitter.on(PoolEvents.backPressure, info => { + ++poolBackPressure + poolInfo = info + }) + for (let i = 0; i < numberOfWorkers + 1; i++) { + promises.add(pool.execute()) + } + await Promise.all(promises) + expect(poolBackPressure).toBe(1) + expect(poolInfo).toStrictEqual({ + version, + type: PoolTypes.fixed, + worker: WorkerTypes.thread, + ready: expect.any(Boolean), + strategy: WorkerChoiceStrategies.ROUND_ROBIN, + minSize: expect.any(Number), + maxSize: expect.any(Number), + workerNodes: expect.any(Number), + idleWorkerNodes: expect.any(Number), + busyWorkerNodes: expect.any(Number), + executedTasks: expect.any(Number), + executingTasks: expect.any(Number), + maxQueuedTasks: expect.any(Number), + queuedTasks: expect.any(Number), + backPressure: true, + stolenTasks: expect.any(Number), + failedTasks: expect.any(Number) + }) + expect(pool.hasBackPressure.called).toBe(true) + await pool.destroy() + }) + it('Verify that listTaskFunctions() is working', async () => { const dynamicThreadPool = new DynamicThreadPool( Math.floor(numberOfWorkers / 2), @@ -880,7 +1177,7 @@ describe('Abstract pool test suite', () => { ) await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1) expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([ - 'default', + DEFAULT_TASK_NAME, 'jsonIntegerSerialization', 'factorial', 'fibonacci' @@ -891,11 +1188,13 @@ describe('Abstract pool test suite', () => { ) await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1) expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([ - 'default', + DEFAULT_TASK_NAME, 'jsonIntegerSerialization', 'factorial', 'fibonacci' ]) + await dynamicThreadPool.destroy() + await fixedClusterPool.destroy() }) it('Verify that multiple task functions worker is working', async () => { @@ -917,7 +1216,7 @@ describe('Abstract pool test suite', () => { expect(pool.info.executedTasks).toBe(4) for (const workerNode of pool.workerNodes) { expect(workerNode.info.taskFunctions).toStrictEqual([ - 'default', + DEFAULT_TASK_NAME, 'jsonIntegerSerialization', 'factorial', 'fibonacci' @@ -927,9 +1226,10 @@ describe('Abstract pool test suite', () => { expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({ tasks: { executed: expect.any(Number), - executing: expect.any(Number), + executing: 0, failed: 0, - queued: 0 + queued: 0, + stolen: 0 }, runTime: { history: expect.any(CircularArray) @@ -947,9 +1247,31 @@ describe('Abstract pool test suite', () => { } }) expect( - workerNode.getTaskFunctionWorkerUsage(name).tasks.executing - ).toBeGreaterThanOrEqual(0) + workerNode.getTaskFunctionWorkerUsage(name).tasks.executed + ).toBeGreaterThan(0) } + expect( + workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME) + ).toStrictEqual( + workerNode.getTaskFunctionWorkerUsage(workerNode.info.taskFunctions[1]) + ) } + await pool.destroy() + }) + + it('Verify sendKillMessageToWorker()', async () => { + const pool = new DynamicClusterPool( + Math.floor(numberOfWorkers / 2), + numberOfWorkers, + './tests/worker-files/cluster/testWorker.js' + ) + const workerNodeKey = 0 + await expect( + pool.sendKillMessageToWorker( + workerNodeKey, + pool.workerNodes[workerNodeKey].info.id + ) + ).resolves.toBeUndefined() + await pool.destroy() }) })