1 // eslint-disable-next-line n/no-unsupported-features/node-builtins
2 import { createHook, executionAsyncId } from 'node:async_hooks'
3 import { EventEmitterAsyncResource } from 'node:events'
4 import { readFileSync } from 'node:fs'
5 import { dirname, join } from 'node:path'
6 import { fileURLToPath } from 'node:url'
8 import { expect } from 'expect'
9 import { restore, stub } from 'sinon'
11 import { CircularBuffer } from '../../lib/circular-buffer.cjs'
19 WorkerChoiceStrategies,
21 } from '../../lib/index.cjs'
22 import { WorkerNode } from '../../lib/pools/worker-node.cjs'
23 import { PriorityQueue } from '../../lib/priority-queue.cjs'
24 import { defaultBucketSize } from '../../lib/utility-types.cjs'
25 import { DEFAULT_TASK_NAME } from '../../lib/utils.cjs'
26 import { waitPoolEvents } from '../test-utils.cjs'
28 describe('Abstract pool test suite', () => {
29 const version = JSON.parse(
31 join(dirname(fileURLToPath(import.meta.url)), '../..', 'package.json'),
35 const numberOfWorkers = 2
36 class StubPoolWithIsMain extends FixedThreadPool {
46 it('Verify that pool can be created and destroyed', async () => {
47 const pool = new FixedThreadPool(
49 './tests/worker-files/thread/testWorker.mjs'
51 expect(pool).toBeInstanceOf(FixedThreadPool)
55 it('Verify that pool cannot be created from a non main thread/process', () => {
58 new StubPoolWithIsMain(
60 './tests/worker-files/thread/testWorker.mjs',
62 errorHandler: e => console.error(e),
67 'Cannot start a pool from a worker with the same type as the pool'
72 it('Verify that pool statuses properties are set', async () => {
73 const pool = new FixedThreadPool(
75 './tests/worker-files/thread/testWorker.mjs'
77 expect(pool.started).toBe(true)
78 expect(pool.starting).toBe(false)
79 expect(pool.destroying).toBe(false)
83 it('Verify that filePath is checked', () => {
84 expect(() => new FixedThreadPool(numberOfWorkers)).toThrow(
85 new TypeError('The worker file path must be specified')
87 expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrow(
88 new TypeError('The worker file path must be a string')
91 () => new FixedThreadPool(numberOfWorkers, './dummyWorker.ts')
92 ).toThrow(new Error("Cannot find the worker file './dummyWorker.ts'"))
95 it('Verify that numberOfWorkers is checked', () => {
100 './tests/worker-files/thread/testWorker.mjs'
104 'Cannot instantiate a pool without specifying the number of workers'
109 it('Verify that a negative number of workers is checked', () => {
112 new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.cjs')
115 'Cannot instantiate a pool with a negative number of workers'
120 it('Verify that a non integer number of workers is checked', () => {
123 new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.mjs')
126 'Cannot instantiate a pool with a non safe integer number of workers'
131 it('Verify that pool arguments number and pool type are checked', () => {
136 './tests/worker-files/thread/testWorker.mjs',
142 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
147 it('Verify that dynamic pool sizing is checked', () => {
150 new DynamicClusterPool(
153 './tests/worker-files/cluster/testWorker.cjs'
157 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
162 new DynamicThreadPool(
165 './tests/worker-files/thread/testWorker.mjs'
169 'Cannot instantiate a pool with a non safe integer number of workers'
174 new DynamicClusterPool(
177 './tests/worker-files/cluster/testWorker.cjs'
181 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
186 new DynamicThreadPool(
189 './tests/worker-files/thread/testWorker.mjs'
193 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
198 new DynamicThreadPool(
201 './tests/worker-files/thread/testWorker.mjs'
205 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
210 new DynamicClusterPool(
213 './tests/worker-files/cluster/testWorker.cjs'
217 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
222 it('Verify that pool options are checked', async () => {
223 let pool = new FixedThreadPool(
225 './tests/worker-files/thread/testWorker.mjs'
227 expect(pool.emitter).toBeInstanceOf(EventEmitterAsyncResource)
228 expect(pool.emitter.eventNames()).toStrictEqual([])
229 expect(pool.opts).toStrictEqual({
232 restartWorkerOnError: true,
233 enableTasksQueue: false,
234 workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
236 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategiesContext
237 .workerChoiceStrategies) {
238 expect(workerChoiceStrategy.opts).toStrictEqual({
239 runTime: { median: false },
240 waitTime: { median: false },
241 elu: { median: false },
242 weights: expect.objectContaining({
243 0: expect.any(Number),
244 [pool.info.maxSize - 1]: expect.any(Number),
249 const testHandler = () => console.info('test handler executed')
250 pool = new FixedThreadPool(
252 './tests/worker-files/thread/testWorker.mjs',
254 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
255 workerChoiceStrategyOptions: {
256 runTime: { median: true },
257 weights: { 0: 300, 1: 200 },
260 restartWorkerOnError: false,
261 enableTasksQueue: true,
262 tasksQueueOptions: { concurrency: 2 },
263 messageHandler: testHandler,
264 errorHandler: testHandler,
265 onlineHandler: testHandler,
266 exitHandler: testHandler,
269 expect(pool.emitter).toBeUndefined()
270 expect(pool.opts).toStrictEqual({
273 restartWorkerOnError: false,
274 enableTasksQueue: true,
277 size: Math.pow(numberOfWorkers, 2),
279 tasksStealingOnBackPressure: false,
280 tasksFinishedTimeout: 2000,
282 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
283 workerChoiceStrategyOptions: {
284 runTime: { median: true },
285 weights: { 0: 300, 1: 200 },
287 onlineHandler: testHandler,
288 messageHandler: testHandler,
289 errorHandler: testHandler,
290 exitHandler: testHandler,
292 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategiesContext
293 .workerChoiceStrategies) {
294 expect(workerChoiceStrategy.opts).toStrictEqual({
295 runTime: { median: true },
296 waitTime: { median: false },
297 elu: { median: false },
298 weights: { 0: 300, 1: 200 },
304 it('Verify that pool options are validated', () => {
309 './tests/worker-files/thread/testWorker.mjs',
311 workerChoiceStrategy: 'invalidStrategy',
314 ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'"))
319 './tests/worker-files/thread/testWorker.mjs',
321 workerChoiceStrategyOptions: { weights: {} },
326 'Invalid worker choice strategy options: must have a weight for each worker node'
333 './tests/worker-files/thread/testWorker.mjs',
335 workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' },
340 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
347 './tests/worker-files/thread/testWorker.mjs',
349 enableTasksQueue: true,
350 tasksQueueOptions: 'invalidTasksQueueOptions',
354 new TypeError('Invalid tasks queue options: must be a plain object')
360 './tests/worker-files/thread/testWorker.mjs',
362 enableTasksQueue: true,
363 tasksQueueOptions: { concurrency: 0 },
368 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
375 './tests/worker-files/thread/testWorker.mjs',
377 enableTasksQueue: true,
378 tasksQueueOptions: { concurrency: -1 },
383 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
390 './tests/worker-files/thread/testWorker.mjs',
392 enableTasksQueue: true,
393 tasksQueueOptions: { concurrency: 0.2 },
397 new TypeError('Invalid worker node tasks concurrency: must be an integer')
403 './tests/worker-files/thread/testWorker.mjs',
405 enableTasksQueue: true,
406 tasksQueueOptions: { size: 0 },
411 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
418 './tests/worker-files/thread/testWorker.mjs',
420 enableTasksQueue: true,
421 tasksQueueOptions: { size: -1 },
426 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
433 './tests/worker-files/thread/testWorker.mjs',
435 enableTasksQueue: true,
436 tasksQueueOptions: { size: 0.2 },
440 new TypeError('Invalid worker node tasks queue size: must be an integer')
444 it('Verify that pool worker choice strategy options can be set', async () => {
445 const pool = new FixedThreadPool(
447 './tests/worker-files/thread/testWorker.mjs',
448 { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
450 expect(pool.opts.workerChoiceStrategyOptions).toBeUndefined()
451 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategiesContext
452 .workerChoiceStrategies) {
453 expect(workerChoiceStrategy.opts).toStrictEqual({
454 runTime: { median: false },
455 waitTime: { median: false },
456 elu: { median: false },
457 weights: expect.objectContaining({
458 0: expect.any(Number),
459 [pool.info.maxSize - 1]: expect.any(Number),
464 pool.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
482 pool.setWorkerChoiceStrategyOptions({
483 runTime: { median: true },
484 elu: { median: true },
486 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
487 runTime: { median: true },
488 elu: { median: true },
490 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategiesContext
491 .workerChoiceStrategies) {
492 expect(workerChoiceStrategy.opts).toStrictEqual({
493 runTime: { median: true },
494 waitTime: { median: false },
495 elu: { median: true },
496 weights: expect.objectContaining({
497 0: expect.any(Number),
498 [pool.info.maxSize - 1]: expect.any(Number),
503 pool.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
521 pool.setWorkerChoiceStrategyOptions({
522 runTime: { median: false },
523 elu: { median: false },
525 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
526 runTime: { median: false },
527 elu: { median: false },
529 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategiesContext
530 .workerChoiceStrategies) {
531 expect(workerChoiceStrategy.opts).toStrictEqual({
532 runTime: { median: false },
533 waitTime: { median: false },
534 elu: { median: false },
535 weights: expect.objectContaining({
536 0: expect.any(Number),
537 [pool.info.maxSize - 1]: expect.any(Number),
542 pool.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
561 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
564 'Invalid worker choice strategy options: must be a plain object'
567 expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow(
569 'Invalid worker choice strategy options: must have a weight for each worker node'
573 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
576 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
582 it('Verify that pool tasks queue can be enabled/disabled', async () => {
583 const pool = new FixedThreadPool(
585 './tests/worker-files/thread/testWorker.mjs'
587 expect(pool.opts.enableTasksQueue).toBe(false)
588 expect(pool.opts.tasksQueueOptions).toBeUndefined()
589 pool.enableTasksQueue(true)
590 expect(pool.opts.enableTasksQueue).toBe(true)
591 expect(pool.opts.tasksQueueOptions).toStrictEqual({
593 size: Math.pow(numberOfWorkers, 2),
595 tasksStealingOnBackPressure: false,
596 tasksFinishedTimeout: 2000,
598 pool.enableTasksQueue(true, { concurrency: 2 })
599 expect(pool.opts.enableTasksQueue).toBe(true)
600 expect(pool.opts.tasksQueueOptions).toStrictEqual({
602 size: Math.pow(numberOfWorkers, 2),
604 tasksStealingOnBackPressure: false,
605 tasksFinishedTimeout: 2000,
607 pool.enableTasksQueue(false)
608 expect(pool.opts.enableTasksQueue).toBe(false)
609 expect(pool.opts.tasksQueueOptions).toBeUndefined()
613 it('Verify that pool tasks queue options can be set', async () => {
614 const pool = new FixedThreadPool(
616 './tests/worker-files/thread/testWorker.mjs',
617 { enableTasksQueue: true }
619 expect(pool.opts.tasksQueueOptions).toStrictEqual({
621 size: Math.pow(numberOfWorkers, 2),
623 tasksStealingOnBackPressure: false,
624 tasksFinishedTimeout: 2000,
626 for (const workerNode of pool.workerNodes) {
627 expect(workerNode.tasksQueueBackPressureSize).toBe(
628 pool.opts.tasksQueueOptions.size
631 pool.setTasksQueueOptions({
635 tasksStealingOnBackPressure: false,
636 tasksFinishedTimeout: 3000,
638 expect(pool.opts.tasksQueueOptions).toStrictEqual({
642 tasksStealingOnBackPressure: false,
643 tasksFinishedTimeout: 3000,
645 for (const workerNode of pool.workerNodes) {
646 expect(workerNode.tasksQueueBackPressureSize).toBe(
647 pool.opts.tasksQueueOptions.size
650 pool.setTasksQueueOptions({
653 tasksStealingOnBackPressure: true,
655 expect(pool.opts.tasksQueueOptions).toStrictEqual({
657 size: Math.pow(numberOfWorkers, 2),
659 tasksStealingOnBackPressure: true,
660 tasksFinishedTimeout: 2000,
662 for (const workerNode of pool.workerNodes) {
663 expect(workerNode.tasksQueueBackPressureSize).toBe(
664 pool.opts.tasksQueueOptions.size
667 expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions')).toThrow(
668 new TypeError('Invalid tasks queue options: must be a plain object')
670 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrow(
672 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
675 expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrow(
677 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
680 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrow(
681 new TypeError('Invalid worker node tasks concurrency: must be an integer')
683 expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrow(
685 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
688 expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrow(
690 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
693 expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrow(
694 new TypeError('Invalid worker node tasks queue size: must be an integer')
699 it('Verify that pool info is set', async () => {
700 let pool = new FixedThreadPool(
702 './tests/worker-files/thread/testWorker.mjs'
704 expect(pool.info).toStrictEqual({
706 type: PoolTypes.fixed,
707 worker: WorkerTypes.thread,
710 defaultStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
712 minSize: numberOfWorkers,
713 maxSize: numberOfWorkers,
714 workerNodes: numberOfWorkers,
715 idleWorkerNodes: numberOfWorkers,
722 pool = new DynamicClusterPool(
723 Math.floor(numberOfWorkers / 2),
725 './tests/worker-files/cluster/testWorker.cjs'
727 expect(pool.info).toStrictEqual({
729 type: PoolTypes.dynamic,
730 worker: WorkerTypes.cluster,
733 defaultStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
735 minSize: Math.floor(numberOfWorkers / 2),
736 maxSize: numberOfWorkers,
737 workerNodes: Math.floor(numberOfWorkers / 2),
738 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
747 it('Verify that pool worker tasks usage are initialized', async () => {
748 const pool = new FixedClusterPool(
750 './tests/worker-files/cluster/testWorker.cjs'
752 for (const workerNode of pool.workerNodes) {
753 expect(workerNode).toBeInstanceOf(WorkerNode)
754 expect(workerNode.usage).toStrictEqual({
760 sequentiallyStolen: 0,
765 history: expect.any(CircularBuffer),
768 history: expect.any(CircularBuffer),
772 history: expect.any(CircularBuffer),
775 history: expect.any(CircularBuffer),
783 it('Verify that pool worker tasks queue are initialized', async () => {
784 let pool = new FixedClusterPool(
786 './tests/worker-files/cluster/testWorker.cjs'
788 for (const workerNode of pool.workerNodes) {
789 expect(workerNode).toBeInstanceOf(WorkerNode)
790 expect(workerNode.tasksQueue).toBeInstanceOf(PriorityQueue)
791 expect(workerNode.tasksQueue.size).toBe(0)
792 expect(workerNode.tasksQueue.maxSize).toBe(0)
793 expect(workerNode.tasksQueue.bucketSize).toBe(defaultBucketSize)
794 expect(workerNode.tasksQueue.enablePriority).toBe(false)
797 pool = new DynamicThreadPool(
798 Math.floor(numberOfWorkers / 2),
800 './tests/worker-files/thread/testWorker.mjs'
802 for (const workerNode of pool.workerNodes) {
803 expect(workerNode).toBeInstanceOf(WorkerNode)
804 expect(workerNode.tasksQueue).toBeInstanceOf(PriorityQueue)
805 expect(workerNode.tasksQueue.size).toBe(0)
806 expect(workerNode.tasksQueue.maxSize).toBe(0)
807 expect(workerNode.tasksQueue.bucketSize).toBe(defaultBucketSize)
808 expect(workerNode.tasksQueue.enablePriority).toBe(false)
813 it('Verify that pool worker info are initialized', async () => {
814 let pool = new FixedClusterPool(
816 './tests/worker-files/cluster/testWorker.cjs'
818 for (const workerNode of pool.workerNodes) {
819 expect(workerNode).toBeInstanceOf(WorkerNode)
820 expect(workerNode.info).toStrictEqual({
821 id: expect.any(Number),
822 type: WorkerTypes.cluster,
830 pool = new DynamicThreadPool(
831 Math.floor(numberOfWorkers / 2),
833 './tests/worker-files/thread/testWorker.mjs'
835 for (const workerNode of pool.workerNodes) {
836 expect(workerNode).toBeInstanceOf(WorkerNode)
837 expect(workerNode.info).toStrictEqual({
838 id: expect.any(Number),
839 type: WorkerTypes.thread,
849 it('Verify that pool statuses are checked at start or destroy', async () => {
850 const pool = new FixedThreadPool(
852 './tests/worker-files/thread/testWorker.mjs'
854 expect(pool.info.started).toBe(true)
855 expect(pool.info.ready).toBe(true)
856 expect(() => pool.start()).toThrow(
857 new Error('Cannot start an already started pool')
860 expect(pool.info.started).toBe(false)
861 expect(pool.info.ready).toBe(false)
862 await expect(pool.destroy()).rejects.toThrow(
863 new Error('Cannot destroy an already destroyed pool')
867 it('Verify that pool can be started after initialization', async () => {
868 const pool = new FixedClusterPool(
870 './tests/worker-files/cluster/testWorker.cjs',
875 expect(pool.info.started).toBe(false)
876 expect(pool.info.ready).toBe(false)
877 expect(pool.workerNodes).toStrictEqual([])
878 expect(pool.readyEventEmitted).toBe(false)
879 await expect(pool.execute()).rejects.toThrow(
880 new Error('Cannot execute a task on not started pool')
883 expect(pool.info.started).toBe(true)
884 expect(pool.info.ready).toBe(true)
885 await waitPoolEvents(pool, PoolEvents.ready, 1)
886 expect(pool.readyEventEmitted).toBe(true)
887 expect(pool.workerNodes.length).toBe(numberOfWorkers)
888 for (const workerNode of pool.workerNodes) {
889 expect(workerNode).toBeInstanceOf(WorkerNode)
894 it('Verify that pool execute() arguments are checked', async () => {
895 const pool = new FixedClusterPool(
897 './tests/worker-files/cluster/testWorker.cjs'
899 await expect(pool.execute(undefined, 0)).rejects.toThrow(
900 new TypeError('name argument must be a string')
902 await expect(pool.execute(undefined, '')).rejects.toThrow(
903 new TypeError('name argument must not be an empty string')
905 await expect(pool.execute(undefined, undefined, {})).rejects.toThrow(
906 new TypeError('transferList argument must be an array')
908 await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
909 "Task function 'unknown' not found"
912 await expect(pool.execute()).rejects.toThrow(
913 new Error('Cannot execute a task on not started pool')
917 it('Verify that pool worker tasks usage are computed', async () => {
918 const pool = new FixedClusterPool(
920 './tests/worker-files/cluster/testWorker.cjs'
922 const promises = new Set()
923 const maxMultiplier = 2
924 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
925 promises.add(pool.execute())
927 for (const workerNode of pool.workerNodes) {
928 expect(workerNode.usage).toStrictEqual({
931 executing: maxMultiplier,
934 sequentiallyStolen: 0,
939 history: expect.any(CircularBuffer),
942 history: expect.any(CircularBuffer),
946 history: expect.any(CircularBuffer),
949 history: expect.any(CircularBuffer),
954 await Promise.all(promises)
955 for (const workerNode of pool.workerNodes) {
956 expect(workerNode.usage).toStrictEqual({
958 executed: maxMultiplier,
962 sequentiallyStolen: 0,
967 history: expect.any(CircularBuffer),
970 history: expect.any(CircularBuffer),
974 history: expect.any(CircularBuffer),
977 history: expect.any(CircularBuffer),
985 it("Verify that pool worker tasks usage aren't reset at worker choice strategy change", async () => {
986 const pool = new DynamicThreadPool(
987 Math.floor(numberOfWorkers / 2),
989 './tests/worker-files/thread/testWorker.mjs'
991 const promises = new Set()
992 const maxMultiplier = 2
993 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
994 promises.add(pool.execute())
996 await Promise.all(promises)
997 for (const workerNode of pool.workerNodes) {
998 expect(workerNode.usage).toStrictEqual({
1000 executed: expect.any(Number),
1004 sequentiallyStolen: 0,
1009 history: expect.any(CircularBuffer),
1012 history: expect.any(CircularBuffer),
1016 history: expect.any(CircularBuffer),
1019 history: expect.any(CircularBuffer),
1023 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1024 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1025 numberOfWorkers * maxMultiplier
1028 pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
1029 for (const workerNode of pool.workerNodes) {
1030 expect(workerNode.usage).toStrictEqual({
1032 executed: expect.any(Number),
1036 sequentiallyStolen: 0,
1041 history: expect.any(CircularBuffer),
1044 history: expect.any(CircularBuffer),
1048 history: expect.any(CircularBuffer),
1051 history: expect.any(CircularBuffer),
1055 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1056 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1057 numberOfWorkers * maxMultiplier
1060 await pool.destroy()
1063 it("Verify that pool event emitter 'ready' event can register a callback", async () => {
1064 const pool = new DynamicClusterPool(
1065 Math.floor(numberOfWorkers / 2),
1067 './tests/worker-files/cluster/testWorker.cjs'
1069 expect(pool.emitter.eventNames()).toStrictEqual([])
1072 pool.emitter.on(PoolEvents.ready, info => {
1076 await waitPoolEvents(pool, PoolEvents.ready, 1)
1077 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.ready])
1078 expect(poolReady).toBe(1)
1079 expect(poolInfo).toStrictEqual({
1081 type: PoolTypes.dynamic,
1082 worker: WorkerTypes.cluster,
1085 defaultStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
1086 strategyRetries: expect.any(Number),
1087 minSize: expect.any(Number),
1088 maxSize: expect.any(Number),
1089 workerNodes: expect.any(Number),
1090 idleWorkerNodes: expect.any(Number),
1091 busyWorkerNodes: expect.any(Number),
1092 executedTasks: expect.any(Number),
1093 executingTasks: expect.any(Number),
1094 failedTasks: expect.any(Number),
1096 await pool.destroy()
1099 it("Verify that pool event emitter 'busy' event can register a callback", async () => {
1100 const pool = new FixedThreadPool(
1102 './tests/worker-files/thread/testWorker.mjs'
1104 expect(pool.emitter.eventNames()).toStrictEqual([])
1105 const promises = new Set()
1108 pool.emitter.on(PoolEvents.busy, info => {
1112 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
1113 for (let i = 0; i < numberOfWorkers * 2; i++) {
1114 promises.add(pool.execute())
1116 await Promise.all(promises)
1117 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
1118 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
1119 expect(poolBusy).toBe(numberOfWorkers + 1)
1120 expect(poolInfo).toStrictEqual({
1122 type: PoolTypes.fixed,
1123 worker: WorkerTypes.thread,
1126 defaultStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
1127 strategyRetries: expect.any(Number),
1128 minSize: expect.any(Number),
1129 maxSize: expect.any(Number),
1130 workerNodes: expect.any(Number),
1131 idleWorkerNodes: expect.any(Number),
1132 busyWorkerNodes: expect.any(Number),
1133 executedTasks: expect.any(Number),
1134 executingTasks: expect.any(Number),
1135 failedTasks: expect.any(Number),
1137 await pool.destroy()
1140 it("Verify that pool event emitter 'full' event can register a callback", async () => {
1141 const pool = new DynamicThreadPool(
1142 Math.floor(numberOfWorkers / 2),
1144 './tests/worker-files/thread/testWorker.mjs'
1146 expect(pool.emitter.eventNames()).toStrictEqual([])
1147 const promises = new Set()
1150 pool.emitter.on(PoolEvents.full, info => {
1154 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.full])
1155 for (let i = 0; i < numberOfWorkers * 2; i++) {
1156 promises.add(pool.execute())
1158 await Promise.all(promises)
1159 expect(poolFull).toBe(1)
1160 expect(poolInfo).toStrictEqual({
1162 type: PoolTypes.dynamic,
1163 worker: WorkerTypes.thread,
1166 defaultStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
1167 strategyRetries: expect.any(Number),
1168 minSize: expect.any(Number),
1169 maxSize: expect.any(Number),
1170 workerNodes: expect.any(Number),
1171 idleWorkerNodes: expect.any(Number),
1172 busyWorkerNodes: expect.any(Number),
1173 executedTasks: expect.any(Number),
1174 executingTasks: expect.any(Number),
1175 failedTasks: expect.any(Number),
1177 await pool.destroy()
1180 it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
1181 const pool = new FixedThreadPool(
1183 './tests/worker-files/thread/testWorker.mjs',
1185 enableTasksQueue: true,
1188 stub(pool, 'hasBackPressure').returns(true)
1189 expect(pool.emitter.eventNames()).toStrictEqual([])
1190 const promises = new Set()
1191 let poolBackPressure = 0
1193 pool.emitter.on(PoolEvents.backPressure, info => {
1197 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.backPressure])
1198 for (let i = 0; i < numberOfWorkers + 1; i++) {
1199 promises.add(pool.execute())
1201 await Promise.all(promises)
1202 expect(poolBackPressure).toBe(1)
1203 expect(poolInfo).toStrictEqual({
1205 type: PoolTypes.fixed,
1206 worker: WorkerTypes.thread,
1209 defaultStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
1210 strategyRetries: expect.any(Number),
1211 minSize: expect.any(Number),
1212 maxSize: expect.any(Number),
1213 workerNodes: expect.any(Number),
1214 idleWorkerNodes: expect.any(Number),
1215 stealingWorkerNodes: expect.any(Number),
1216 busyWorkerNodes: expect.any(Number),
1217 executedTasks: expect.any(Number),
1218 executingTasks: expect.any(Number),
1219 maxQueuedTasks: expect.any(Number),
1220 queuedTasks: expect.any(Number),
1222 stolenTasks: expect.any(Number),
1223 failedTasks: expect.any(Number),
1225 expect(pool.hasBackPressure.callCount).toBeGreaterThanOrEqual(7)
1226 await pool.destroy()
1229 it('Verify that destroy() waits for queued tasks to finish', async () => {
1230 const tasksFinishedTimeout = 2500
1231 const pool = new FixedThreadPool(
1233 './tests/worker-files/thread/asyncWorker.mjs',
1235 enableTasksQueue: true,
1236 tasksQueueOptions: { tasksFinishedTimeout },
1239 const maxMultiplier = 4
1240 let tasksFinished = 0
1241 for (const workerNode of pool.workerNodes) {
1242 workerNode.on('taskFinished', () => {
1246 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1249 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1250 const startTime = performance.now()
1251 await pool.destroy()
1252 const elapsedTime = performance.now() - startTime
1253 expect(tasksFinished).toBeLessThanOrEqual(numberOfWorkers * maxMultiplier)
1254 expect(elapsedTime).toBeGreaterThanOrEqual(2000)
1255 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 800)
1258 it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
1259 const tasksFinishedTimeout = 1000
1260 const pool = new FixedThreadPool(
1262 './tests/worker-files/thread/asyncWorker.mjs',
1264 enableTasksQueue: true,
1265 tasksQueueOptions: { tasksFinishedTimeout },
1268 const maxMultiplier = 4
1269 let tasksFinished = 0
1270 for (const workerNode of pool.workerNodes) {
1271 workerNode.on('taskFinished', () => {
1275 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1278 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1279 const startTime = performance.now()
1280 await pool.destroy()
1281 const elapsedTime = performance.now() - startTime
1282 expect(tasksFinished).toBe(0)
1283 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 800)
1286 it('Verify that pool asynchronous resource track tasks execution', async () => {
1291 let resolveCalls = 0
1292 const hook = createHook({
1293 init (asyncId, type) {
1294 if (type === 'poolifier:task') {
1296 taskAsyncId = asyncId
1300 if (asyncId === taskAsyncId) beforeCalls++
1303 if (asyncId === taskAsyncId) afterCalls++
1306 if (executionAsyncId() === taskAsyncId) resolveCalls++
1309 const pool = new FixedThreadPool(
1311 './tests/worker-files/thread/testWorker.mjs'
1314 await pool.execute()
1316 expect(initCalls).toBe(1)
1317 expect(beforeCalls).toBe(1)
1318 expect(afterCalls).toBe(1)
1319 expect(resolveCalls).toBe(1)
1320 await pool.destroy()
1323 it('Verify that hasTaskFunction() is working', async () => {
1324 const dynamicThreadPool = new DynamicThreadPool(
1325 Math.floor(numberOfWorkers / 2),
1327 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1329 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1330 expect(dynamicThreadPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1331 expect(dynamicThreadPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1334 expect(dynamicThreadPool.hasTaskFunction('factorial')).toBe(true)
1335 expect(dynamicThreadPool.hasTaskFunction('fibonacci')).toBe(true)
1336 expect(dynamicThreadPool.hasTaskFunction('unknown')).toBe(false)
1337 await dynamicThreadPool.destroy()
1338 const fixedClusterPool = new FixedClusterPool(
1340 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
1342 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1343 expect(fixedClusterPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1344 expect(fixedClusterPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1347 expect(fixedClusterPool.hasTaskFunction('factorial')).toBe(true)
1348 expect(fixedClusterPool.hasTaskFunction('fibonacci')).toBe(true)
1349 expect(fixedClusterPool.hasTaskFunction('unknown')).toBe(false)
1350 await fixedClusterPool.destroy()
1353 it('Verify that addTaskFunction() is working', async () => {
1354 const dynamicThreadPool = new DynamicThreadPool(
1355 Math.floor(numberOfWorkers / 2),
1357 './tests/worker-files/thread/testWorker.mjs'
1359 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1361 dynamicThreadPool.addTaskFunction(0, () => {})
1362 ).rejects.toThrow(new TypeError('name argument must be a string'))
1364 dynamicThreadPool.addTaskFunction('', () => {})
1366 new TypeError('name argument must not be an empty string')
1368 await expect(dynamicThreadPool.addTaskFunction('test', 0)).rejects.toThrow(
1369 new TypeError('taskFunction property must be a function')
1371 await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow(
1372 new TypeError('taskFunction property must be a function')
1375 dynamicThreadPool.addTaskFunction('test', { taskFunction: 0 })
1376 ).rejects.toThrow(new TypeError('taskFunction property must be a function'))
1378 dynamicThreadPool.addTaskFunction('test', { taskFunction: '' })
1379 ).rejects.toThrow(new TypeError('taskFunction property must be a function'))
1381 dynamicThreadPool.addTaskFunction('test', {
1382 taskFunction: () => {},
1386 new RangeError("Property 'priority' must be between -20 and 19")
1389 dynamicThreadPool.addTaskFunction('test', {
1390 taskFunction: () => {},
1394 new RangeError("Property 'priority' must be between -20 and 19")
1397 dynamicThreadPool.addTaskFunction('test', {
1398 taskFunction: () => {},
1399 strategy: 'invalidStrategy',
1402 new Error("Invalid worker choice strategy 'invalidStrategy'")
1404 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1405 { name: DEFAULT_TASK_NAME },
1409 ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(),
1410 ]).toStrictEqual([WorkerChoiceStrategies.ROUND_ROBIN])
1411 const echoTaskFunction = data => {
1415 dynamicThreadPool.addTaskFunction('echo', {
1416 taskFunction: echoTaskFunction,
1417 strategy: WorkerChoiceStrategies.LEAST_ELU,
1419 ).resolves.toBe(true)
1420 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1421 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual({
1422 taskFunction: echoTaskFunction,
1423 strategy: WorkerChoiceStrategies.LEAST_ELU,
1426 ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(),
1428 WorkerChoiceStrategies.ROUND_ROBIN,
1429 WorkerChoiceStrategies.LEAST_ELU,
1431 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1432 { name: DEFAULT_TASK_NAME },
1434 { name: 'echo', strategy: WorkerChoiceStrategies.LEAST_ELU },
1436 const taskFunctionData = { test: 'test' }
1437 const echoResult = await dynamicThreadPool.execute(taskFunctionData, 'echo')
1438 expect(echoResult).toStrictEqual(taskFunctionData)
1439 for (const workerNode of dynamicThreadPool.workerNodes) {
1440 expect(workerNode.getTaskFunctionWorkerUsage('echo')).toStrictEqual({
1442 executed: expect.any(Number),
1445 sequentiallyStolen: 0,
1450 history: expect.any(CircularBuffer),
1453 history: expect.any(CircularBuffer),
1455 elu: expect.objectContaining({
1456 idle: expect.objectContaining({
1457 history: expect.any(CircularBuffer),
1459 active: expect.objectContaining({
1460 history: expect.any(CircularBuffer),
1465 workerNode.getTaskFunctionWorkerUsage('echo').tasks.executed
1466 ).toBeGreaterThan(0)
1468 workerNode.getTaskFunctionWorkerUsage('echo').elu.active.aggregate ==
1472 workerNode.getTaskFunctionWorkerUsage('echo').elu.active.aggregate
1476 workerNode.getTaskFunctionWorkerUsage('echo').elu.active.aggregate
1477 ).toBeGreaterThan(0)
1480 workerNode.getTaskFunctionWorkerUsage('echo').elu.idle.aggregate == null
1483 workerNode.getTaskFunctionWorkerUsage('echo').elu.idle.aggregate
1487 workerNode.getTaskFunctionWorkerUsage('echo').elu.idle.aggregate
1488 ).toBeGreaterThanOrEqual(0)
1491 workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization == null
1494 workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization
1498 workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization
1499 ).toBeGreaterThanOrEqual(0)
1501 workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization
1502 ).toBeLessThanOrEqual(1)
1505 await dynamicThreadPool.destroy()
1508 it('Verify that removeTaskFunction() is working', async () => {
1509 const dynamicThreadPool = new DynamicThreadPool(
1510 Math.floor(numberOfWorkers / 2),
1512 './tests/worker-files/thread/testWorker.mjs'
1514 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1515 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1516 { name: DEFAULT_TASK_NAME },
1519 await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow(
1520 new Error('Cannot remove a task function not handled on the pool side')
1522 const echoTaskFunction = data => {
1525 await dynamicThreadPool.addTaskFunction('echo', {
1526 taskFunction: echoTaskFunction,
1527 strategy: WorkerChoiceStrategies.LEAST_ELU,
1529 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1530 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual({
1531 taskFunction: echoTaskFunction,
1532 strategy: WorkerChoiceStrategies.LEAST_ELU,
1535 ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(),
1537 WorkerChoiceStrategies.ROUND_ROBIN,
1538 WorkerChoiceStrategies.LEAST_ELU,
1540 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1541 { name: DEFAULT_TASK_NAME },
1543 { name: 'echo', strategy: WorkerChoiceStrategies.LEAST_ELU },
1545 await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe(
1548 expect(dynamicThreadPool.taskFunctions.size).toBe(0)
1549 expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined()
1551 ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys(),
1552 ]).toStrictEqual([WorkerChoiceStrategies.ROUND_ROBIN])
1553 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1554 { name: DEFAULT_TASK_NAME },
1557 await dynamicThreadPool.destroy()
1560 it('Verify that listTaskFunctionsProperties() is working', async () => {
1561 const dynamicThreadPool = new DynamicThreadPool(
1562 Math.floor(numberOfWorkers / 2),
1564 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1566 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1567 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1568 { name: DEFAULT_TASK_NAME },
1569 { name: 'jsonIntegerSerialization' },
1570 { name: 'factorial' },
1571 { name: 'fibonacci' },
1573 await dynamicThreadPool.destroy()
1574 const fixedClusterPool = new FixedClusterPool(
1576 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
1578 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1579 expect(fixedClusterPool.listTaskFunctionsProperties()).toStrictEqual([
1580 { name: DEFAULT_TASK_NAME },
1581 { name: 'jsonIntegerSerialization' },
1582 { name: 'factorial' },
1583 { name: 'fibonacci' },
1585 await fixedClusterPool.destroy()
1588 it('Verify that setDefaultTaskFunction() is working', async () => {
1589 const dynamicThreadPool = new DynamicThreadPool(
1590 Math.floor(numberOfWorkers / 2),
1592 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1594 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1595 const workerId = dynamicThreadPool.workerNodes[0].info.id
1596 await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow(
1598 `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'`
1602 dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME)
1605 `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function reserved name as the default task function'`
1609 dynamicThreadPool.setDefaultTaskFunction('unknown')
1612 `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function to a non-existing task function'`
1615 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1616 { name: DEFAULT_TASK_NAME },
1617 { name: 'jsonIntegerSerialization' },
1618 { name: 'factorial' },
1619 { name: 'fibonacci' },
1622 dynamicThreadPool.setDefaultTaskFunction('factorial')
1623 ).resolves.toBe(true)
1624 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1625 { name: DEFAULT_TASK_NAME },
1626 { name: 'factorial' },
1627 { name: 'jsonIntegerSerialization' },
1628 { name: 'fibonacci' },
1631 dynamicThreadPool.setDefaultTaskFunction('fibonacci')
1632 ).resolves.toBe(true)
1633 expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1634 { name: DEFAULT_TASK_NAME },
1635 { name: 'fibonacci' },
1636 { name: 'jsonIntegerSerialization' },
1637 { name: 'factorial' },
1639 await dynamicThreadPool.destroy()
1642 it('Verify that multiple task functions worker is working', async () => {
1643 const pool = new DynamicClusterPool(
1644 Math.floor(numberOfWorkers / 2),
1646 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
1648 const data = { n: 10 }
1649 const result0 = await pool.execute(data)
1650 expect(result0).toStrictEqual({ ok: 1 })
1651 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1652 expect(result1).toStrictEqual({ ok: 1 })
1653 const result2 = await pool.execute(data, 'factorial')
1654 expect(result2).toBe(3628800)
1655 const result3 = await pool.execute(data, 'fibonacci')
1656 expect(result3).toBe(55)
1657 expect(pool.info.executingTasks).toBe(0)
1658 expect(pool.info.executedTasks).toBe(4)
1659 for (const workerNode of pool.workerNodes) {
1660 expect(workerNode.info.taskFunctionsProperties).toStrictEqual([
1661 { name: DEFAULT_TASK_NAME },
1662 { name: 'jsonIntegerSerialization' },
1663 { name: 'factorial' },
1664 { name: 'fibonacci' },
1666 expect(workerNode.taskFunctionsUsage.size).toBe(3)
1667 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1668 expect(workerNode.tasksQueue.enablePriority).toBe(false)
1669 for (const taskFunctionProperties of pool.listTaskFunctionsProperties()) {
1671 workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1674 executed: expect.any(Number),
1678 sequentiallyStolen: 0,
1682 history: expect.any(CircularBuffer),
1685 history: expect.any(CircularBuffer),
1689 history: expect.any(CircularBuffer),
1692 history: expect.any(CircularBuffer),
1697 workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1699 ).toBeGreaterThan(0)
1702 workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1704 workerNode.getTaskFunctionWorkerUsage(
1705 workerNode.info.taskFunctionsProperties[1].name
1709 await pool.destroy()
1712 it('Verify that mapExecute() is working', async () => {
1713 const pool = new DynamicThreadPool(
1714 Math.floor(numberOfWorkers / 2),
1716 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1718 expect(() => pool.mapExecute()).toThrow(
1719 new TypeError('data argument must be a defined iterable')
1721 expect(() => pool.mapExecute(0)).toThrow(
1722 new TypeError('data argument must be an iterable')
1724 let results = await pool.mapExecute([{}, {}, {}, {}])
1725 expect(results).toStrictEqual([{ ok: 1 }, { ok: 1 }, { ok: 1 }, { ok: 1 }])
1726 expect(pool.info.executingTasks).toBe(0)
1727 expect(pool.info.executedTasks).toBe(4)
1728 results = await pool.mapExecute(
1729 [{ n: 10 }, { n: 20 }, { n: 30 }, { n: 40 }],
1732 expect(results).toStrictEqual([
1733 3628800, 2432902008176640000, 2.6525285981219103e32, 8.159152832478977e47,
1735 expect(pool.info.executingTasks).toBe(0)
1736 expect(pool.info.executedTasks).toBe(8)
1737 results = await pool.mapExecute(
1738 new Set([{ n: 10 }, { n: 20 }, { n: 30 }, { n: 40 }]),
1741 expect(results).toStrictEqual([
1742 3628800, 2432902008176640000, 2.6525285981219103e32, 8.159152832478977e47,
1744 expect(pool.info.executingTasks).toBe(0)
1745 expect(pool.info.executedTasks).toBe(12)
1746 await pool.destroy()
1749 it('Verify that task function objects worker is working', async () => {
1750 const pool = new DynamicThreadPool(
1751 Math.floor(numberOfWorkers / 2),
1753 './tests/worker-files/thread/testTaskFunctionObjectsWorker.mjs'
1755 const data = { n: 10 }
1756 const result0 = await pool.execute(data)
1757 expect(result0).toStrictEqual({ ok: 1 })
1758 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1759 expect(result1).toStrictEqual({ ok: 1 })
1760 const result2 = await pool.execute(data, 'factorial')
1761 expect(result2).toBe(3628800)
1762 const result3 = await pool.execute(data, 'fibonacci')
1763 expect(result3).toBe(55)
1764 expect(pool.info.executingTasks).toBe(0)
1765 expect(pool.info.executedTasks).toBe(4)
1766 for (const workerNode of pool.workerNodes) {
1767 expect(workerNode.info.taskFunctionsProperties).toStrictEqual([
1768 { name: DEFAULT_TASK_NAME },
1769 { name: 'jsonIntegerSerialization' },
1770 { name: 'factorial' },
1771 { name: 'fibonacci', priority: -5 },
1773 expect(workerNode.taskFunctionsUsage.size).toBe(3)
1774 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1775 expect(workerNode.tasksQueue.enablePriority).toBe(true)
1776 for (const taskFunctionProperties of pool.listTaskFunctionsProperties()) {
1778 workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1781 executed: expect.any(Number),
1785 sequentiallyStolen: 0,
1789 history: expect.any(CircularBuffer),
1792 history: expect.any(CircularBuffer),
1796 history: expect.any(CircularBuffer),
1799 history: expect.any(CircularBuffer),
1804 workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1806 ).toBeGreaterThan(0)
1809 workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1811 workerNode.getTaskFunctionWorkerUsage(
1812 workerNode.info.taskFunctionsProperties[1].name
1816 await pool.destroy()
1819 it('Verify sendKillMessageToWorker()', async () => {
1820 const pool = new DynamicClusterPool(
1821 Math.floor(numberOfWorkers / 2),
1823 './tests/worker-files/cluster/testWorker.cjs'
1825 const workerNodeKey = 0
1827 pool.sendKillMessageToWorker(workerNodeKey)
1828 ).resolves.toBeUndefined()
1829 await pool.destroy()
1832 it('Verify sendTaskFunctionOperationToWorker()', async () => {
1833 const pool = new DynamicClusterPool(
1834 Math.floor(numberOfWorkers / 2),
1836 './tests/worker-files/cluster/testWorker.cjs'
1838 const workerNodeKey = 0
1840 pool.sendTaskFunctionOperationToWorker(workerNodeKey, {
1841 taskFunctionOperation: 'add',
1842 taskFunctionProperties: { name: 'empty' },
1843 taskFunction: (() => {}).toString(),
1845 ).resolves.toBe(true)
1847 pool.workerNodes[workerNodeKey].info.taskFunctionsProperties
1849 { name: DEFAULT_TASK_NAME },
1853 await pool.destroy()
1856 it('Verify sendTaskFunctionOperationToWorkers()', async () => {
1857 const pool = new DynamicClusterPool(
1858 Math.floor(numberOfWorkers / 2),
1860 './tests/worker-files/cluster/testWorker.cjs'
1863 pool.sendTaskFunctionOperationToWorkers({
1864 taskFunctionOperation: 'add',
1865 taskFunctionProperties: { name: 'empty' },
1866 taskFunction: (() => {}).toString(),
1868 ).resolves.toBe(true)
1869 for (const workerNode of pool.workerNodes) {
1870 expect(workerNode.info.taskFunctionsProperties).toStrictEqual([
1871 { name: DEFAULT_TASK_NAME },
1876 await pool.destroy()