1 const { expect
} = require('expect')
2 const { FixedClusterPool
, PoolEvents
} = require('../../../lib')
3 const { WorkerFunctions
} = require('../../test-types')
4 const TestUtils
= require('../../test-utils')
6 describe('Fixed cluster pool test suite', () => {
7 const numberOfWorkers
= 6
8 const pool
= new FixedClusterPool(
10 './tests/worker-files/cluster/testWorker.js',
12 errorHandler
: e
=> console
.error(e
)
15 const queuePool
= new FixedClusterPool(
17 './tests/worker-files/cluster/testWorker.js',
19 enableTasksQueue
: true,
23 errorHandler
: e
=> console
.error(e
)
26 const emptyPool
= new FixedClusterPool(
28 './tests/worker-files/cluster/emptyWorker.js',
29 { exitHandler
: () => console
.log('empty pool worker exited') }
31 const echoPool
= new FixedClusterPool(
33 './tests/worker-files/cluster/echoWorker.js'
35 const errorPool
= new FixedClusterPool(
37 './tests/worker-files/cluster/errorWorker.js',
39 errorHandler
: e
=> console
.error(e
)
42 const asyncErrorPool
= new FixedClusterPool(
44 './tests/worker-files/cluster/asyncErrorWorker.js',
46 errorHandler
: e
=> console
.error(e
)
49 const asyncPool
= new FixedClusterPool(
51 './tests/worker-files/cluster/asyncWorker.js'
54 after('Destroy all pools', async () => {
55 // We need to clean up the resources after our test
56 await echoPool
.destroy()
57 await asyncPool
.destroy()
58 await errorPool
.destroy()
59 await asyncErrorPool
.destroy()
60 await emptyPool
.destroy()
61 await queuePool
.destroy()
64 it('Verify that the function is executed in a worker cluster', async () => {
65 let result
= await pool
.execute({
66 function: WorkerFunctions
.fibonacci
68 expect(result
).toBe(121393)
69 result
= await pool
.execute({
70 function: WorkerFunctions
.factorial
72 expect(result
).toBe(9.33262154439441e157
)
75 it('Verify that is possible to invoke the execute() method without input', async () => {
76 const result
= await pool
.execute()
77 expect(result
).toBe(false)
80 it("Verify that 'busy' event is emitted", async () => {
82 pool
.emitter
.on(PoolEvents
.busy
, () => ++poolBusy
)
83 for (let i
= 0; i
< numberOfWorkers
* 2; i
++) {
86 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
87 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
88 expect(poolBusy
).toBe(numberOfWorkers
+ 1)
91 it('Verify that tasks queuing is working', async () => {
92 const promises
= new Set()
93 const maxMultiplier
= 2
94 for (let i
= 0; i
< numberOfWorkers
* maxMultiplier
; i
++) {
95 promises
.add(queuePool
.execute())
97 expect(promises
.size
).toBe(numberOfWorkers
* maxMultiplier
)
98 for (const workerNode
of queuePool
.workerNodes
) {
99 expect(workerNode
.tasksUsage
.running
).toBeLessThanOrEqual(
100 queuePool
.opts
.tasksQueueOptions
.concurrency
102 expect(workerNode
.tasksUsage
.ran
).toBe(0)
103 expect(workerNode
.tasksQueue
.size
).toBeGreaterThan(0)
105 expect(queuePool
.info
.runningTasks
).toBe(numberOfWorkers
)
106 expect(queuePool
.info
.queuedTasks
).toBe(
107 numberOfWorkers
* maxMultiplier
- numberOfWorkers
109 expect(queuePool
.info
.maxQueuedTasks
).toBe(
110 numberOfWorkers
* maxMultiplier
- numberOfWorkers
112 await Promise
.all(promises
)
113 for (const workerNode
of queuePool
.workerNodes
) {
114 expect(workerNode
.tasksUsage
.running
).toBe(0)
115 expect(workerNode
.tasksUsage
.ran
).toBeGreaterThan(0)
116 expect(workerNode
.tasksUsage
.ran
).toBeLessThanOrEqual(maxMultiplier
)
117 expect(workerNode
.tasksQueue
.size
).toBe(0)
121 it('Verify that is possible to have a worker that return undefined', async () => {
122 const result
= await emptyPool
.execute()
123 expect(result
).toBeUndefined()
126 it('Verify that data are sent to the worker correctly', async () => {
127 const data
= { f
: 10 }
128 const result
= await echoPool
.execute(data
)
129 expect(result
).toStrictEqual(data
)
132 it('Verify that error handling is working properly:sync', async () => {
133 const data
= { f
: 10 }
135 errorPool
.emitter
.on(PoolEvents
.taskError
, e
=> {
140 await errorPool
.execute(data
)
144 expect(inError
).toBeDefined()
145 expect(typeof inError
=== 'string').toBe(true)
146 expect(inError
).toBe('Error Message from ClusterWorker')
147 expect(taskError
).toStrictEqual({
148 error
: 'Error Message from ClusterWorker',
152 errorPool
.workerNodes
.some(
153 workerNode
=> workerNode
.tasksUsage
.error
=== 1
158 it('Verify that error handling is working properly:async', async () => {
159 const data
= { f
: 10 }
161 // errorPool.emitter.on(PoolEvents.taskError, e => {
166 await asyncErrorPool
.execute(data
)
170 expect(inError
).toBeDefined()
171 expect(typeof inError
=== 'string').toBe(true)
172 expect(inError
).toBe('Error Message from ClusterWorker:async')
173 // expect(taskError).toStrictEqual({
174 // error: 'Error Message from ClusterWorker:async',
178 asyncErrorPool
.workerNodes
.some(
179 workerNode
=> workerNode
.tasksUsage
.error
=== 1
184 it('Verify that async function is working properly', async () => {
185 const data
= { f
: 10 }
186 const startTime
= performance
.now()
187 const result
= await asyncPool
.execute(data
)
188 const usedTime
= performance
.now() - startTime
189 expect(result
).toStrictEqual(data
)
190 expect(usedTime
).toBeGreaterThanOrEqual(2000)
193 it('Shutdown test', async () => {
194 const exitPromise
= TestUtils
.waitExits(pool
, numberOfWorkers
)
196 const numberOfExitEvents
= await exitPromise
197 expect(numberOfExitEvents
).toBe(numberOfWorkers
)
200 it('Verify that cluster pool options are checked', async () => {
201 const workerFilePath
= './tests/worker-files/cluster/testWorker.js'
202 let pool1
= new FixedClusterPool(numberOfWorkers
, workerFilePath
)
203 expect(pool1
.opts
.env
).toBeUndefined()
204 expect(pool1
.opts
.settings
).toBeUndefined()
205 await pool1
.destroy()
206 pool1
= new FixedClusterPool(numberOfWorkers
, workerFilePath
, {
207 env
: { TEST
: 'test' },
208 settings
: { args
: ['--use', 'http'], silent
: true }
210 expect(pool1
.opts
.env
).toStrictEqual({ TEST
: 'test' })
211 expect(pool1
.opts
.settings
).toStrictEqual({
212 args
: ['--use', 'http'],
215 expect({ ...pool1
.opts
.settings
, exec
: workerFilePath
}).toStrictEqual({
216 args
: ['--use', 'http'],
220 await pool1
.destroy()
223 it('Should work even without opts in input', async () => {
224 const pool1
= new FixedClusterPool(
226 './tests/worker-files/cluster/testWorker.js'
228 const res
= await pool1
.execute()
229 expect(res
).toBe(false)
230 // We need to clean up the resources after our test
231 await pool1
.destroy()
234 it('Verify that a pool with zero worker fails', async () => {
237 new FixedClusterPool(0, './tests/worker-files/cluster/testWorker.js')
238 ).toThrowError('Cannot instantiate a fixed pool with no worker')