X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fabstract-pool.test.mjs;h=0f61dbd87d5e2911f9e1a648023ec1d6dade5005;hb=55d7d6002049be09a06b08da26febe2e8bfa494b;hp=765a2b0d1553febb95feb3fd36307e6f2eacda66;hpb=fa015370d8f886ee3dcb348483dfa4187239d4d3;p=poolifier.git diff --git a/tests/pools/abstract-pool.test.mjs b/tests/pools/abstract-pool.test.mjs index 765a2b0d..0f61dbd8 100644 --- a/tests/pools/abstract-pool.test.mjs +++ b/tests/pools/abstract-pool.test.mjs @@ -2,6 +2,7 @@ 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 { @@ -38,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( @@ -48,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' ) @@ -60,18 +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', () => { - expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError( - new Error("Cannot find the worker file 'undefined'") + expect(() => new FixedThreadPool(numberOfWorkers)).toThrow( + new TypeError('The worker file path must be specified') + ) + 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', () => { @@ -81,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' ) @@ -92,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' ) @@ -103,7 +117,7 @@ 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' ) @@ -118,7 +132,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' ) @@ -130,7 +144,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' ) @@ -142,7 +156,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' ) @@ -154,7 +168,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' ) @@ -166,7 +180,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' ) @@ -178,7 +192,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' ) @@ -250,7 +264,8 @@ 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: { @@ -295,9 +310,7 @@ describe('Abstract pool test suite', () => { workerChoiceStrategy: 'invalidStrategy' } ) - ).toThrowError( - new Error("Invalid worker choice strategy 'invalidStrategy'") - ) + ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'")) expect( () => new FixedThreadPool( @@ -309,7 +322,7 @@ describe('Abstract pool test suite', () => { } } ) - ).toThrowError( + ).toThrow( new TypeError( 'Invalid worker choice strategy options: retries must be an integer' ) @@ -325,7 +338,7 @@ describe('Abstract pool test suite', () => { } } ) - ).toThrowError( + ).toThrow( new RangeError( "Invalid worker choice strategy options: retries '-1' must be greater or equal than zero" ) @@ -339,7 +352,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' ) @@ -353,7 +366,7 @@ describe('Abstract pool test suite', () => { workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' } } ) - ).toThrowError( + ).toThrow( new Error( "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'" ) @@ -368,7 +381,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: 'invalidTasksQueueOptions' } ) - ).toThrowError( + ).toThrow( new TypeError('Invalid tasks queue options: must be a plain object') ) expect( @@ -381,7 +394,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' ) @@ -396,7 +409,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' ) @@ -411,7 +424,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { concurrency: 0.2 } } ) - ).toThrowError( + ).toThrow( new TypeError('Invalid worker node tasks concurrency: must be an integer') ) expect( @@ -424,7 +437,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' ) @@ -439,7 +452,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' ) @@ -454,7 +467,7 @@ describe('Abstract pool test suite', () => { tasksQueueOptions: { size: 0.2 } } ) - ).toThrowError( + ).toThrow( new TypeError('Invalid worker node tasks queue size: must be an integer') ) }) @@ -595,7 +608,7 @@ describe('Abstract pool test suite', () => { }) expect(() => pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions') - ).toThrowError( + ).toThrow( new TypeError( 'Invalid worker choice strategy options: must be a plain object' ) @@ -604,28 +617,24 @@ describe('Abstract pool test suite', () => { pool.setWorkerChoiceStrategyOptions({ retries: 'invalidChoiceRetries' }) - ).toThrowError( + ).toThrow( new TypeError( 'Invalid worker choice strategy options: retries must be an integer' ) ) - expect(() => - pool.setWorkerChoiceStrategyOptions({ retries: -1 }) - ).toThrowError( + expect(() => pool.setWorkerChoiceStrategyOptions({ retries: -1 })).toThrow( 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'" ) @@ -640,41 +649,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() }) @@ -688,33 +683,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, @@ -725,44 +719,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() @@ -827,6 +818,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -907,6 +899,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, @@ -917,13 +927,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) @@ -936,20 +949,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') ) }) @@ -971,6 +984,7 @@ describe('Abstract pool test suite', () => { executing: maxMultiplier, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -998,6 +1012,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1039,6 +1054,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1074,6 +1090,7 @@ describe('Abstract pool test suite', () => { executing: 0, queued: 0, maxQueued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1257,7 +1274,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 + 300) + }) + + 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() }) @@ -1300,18 +1411,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' @@ -1340,6 +1451,7 @@ describe('Abstract pool test suite', () => { executed: expect.any(Number), executing: 0, queued: 0, + sequentiallyStolen: 0, stolen: 0, failed: 0 }, @@ -1373,9 +1485,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 => { @@ -1438,25 +1548,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([ @@ -1483,6 +1592,7 @@ describe('Abstract pool test suite', () => { 'jsonIntegerSerialization', 'factorial' ]) + await dynamicThreadPool.destroy() }) it('Verify that multiple task functions worker is working', async () => { @@ -1517,6 +1627,7 @@ describe('Abstract pool test suite', () => { executing: 0, failed: 0, queued: 0, + sequentiallyStolen: 0, stolen: 0 }, runTime: { @@ -1559,6 +1670,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() })