fix: fix race condition at counting executing tasks on worker node
[poolifier.git] / tests / pools / cluster / fixed.test.js
1 const { expect } = require('expect')
2 const { FixedClusterPool, PoolEvents } = require('../../../lib')
3 const { TaskFunctions } = require('../../test-types')
4 const { waitPoolEvents, waitWorkerEvents } = require('../../test-utils')
5
6 describe('Fixed cluster pool test suite', () => {
7 const numberOfWorkers = 6
8 const tasksConcurrency = 2
9 const pool = new FixedClusterPool(
10 numberOfWorkers,
11 './tests/worker-files/cluster/testWorker.js',
12 {
13 errorHandler: (e) => console.error(e)
14 }
15 )
16 const queuePool = new FixedClusterPool(
17 numberOfWorkers,
18 './tests/worker-files/cluster/testWorker.js',
19 {
20 enableTasksQueue: true,
21 tasksQueueOptions: {
22 concurrency: tasksConcurrency
23 },
24 errorHandler: (e) => console.error(e)
25 }
26 )
27 const emptyPool = new FixedClusterPool(
28 numberOfWorkers,
29 './tests/worker-files/cluster/emptyWorker.js',
30 { exitHandler: () => console.info('empty pool worker exited') }
31 )
32 const echoPool = new FixedClusterPool(
33 numberOfWorkers,
34 './tests/worker-files/cluster/echoWorker.js'
35 )
36 const errorPool = new FixedClusterPool(
37 numberOfWorkers,
38 './tests/worker-files/cluster/errorWorker.js',
39 {
40 errorHandler: (e) => console.error(e)
41 }
42 )
43 const asyncErrorPool = new FixedClusterPool(
44 numberOfWorkers,
45 './tests/worker-files/cluster/asyncErrorWorker.js',
46 {
47 errorHandler: (e) => console.error(e)
48 }
49 )
50 const asyncPool = new FixedClusterPool(
51 numberOfWorkers,
52 './tests/worker-files/cluster/asyncWorker.js'
53 )
54
55 after('Destroy all pools', async () => {
56 // We need to clean up the resources after our test
57 await echoPool.destroy()
58 await asyncPool.destroy()
59 await errorPool.destroy()
60 await asyncErrorPool.destroy()
61 await emptyPool.destroy()
62 await queuePool.destroy()
63 })
64
65 it('Verify that the function is executed in a worker cluster', async () => {
66 let result = await pool.execute({
67 function: TaskFunctions.fibonacci
68 })
69 expect(result).toBe(75025)
70 result = await pool.execute({
71 function: TaskFunctions.factorial
72 })
73 expect(result).toBe(9.33262154439441e157)
74 })
75
76 it('Verify that is possible to invoke the execute() method without input', async () => {
77 const result = await pool.execute()
78 expect(result).toStrictEqual({ ok: 1 })
79 })
80
81 it("Verify that 'ready' event is emitted", async () => {
82 const pool1 = new FixedClusterPool(
83 numberOfWorkers,
84 './tests/worker-files/cluster/testWorker.js',
85 {
86 errorHandler: (e) => console.error(e)
87 }
88 )
89 let poolReady = 0
90 pool1.emitter.on(PoolEvents.ready, () => ++poolReady)
91 await waitPoolEvents(pool1, PoolEvents.ready, 1)
92 expect(poolReady).toBe(1)
93 })
94
95 it("Verify that 'busy' event is emitted", () => {
96 let poolBusy = 0
97 pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
98 for (let i = 0; i < numberOfWorkers * 2; i++) {
99 pool.execute()
100 }
101 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
102 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
103 expect(poolBusy).toBe(numberOfWorkers + 1)
104 })
105
106 it('Verify that tasks queuing is working', async () => {
107 const promises = new Set()
108 const maxMultiplier = 3 // Must be greater than tasksConcurrency
109 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
110 promises.add(queuePool.execute())
111 }
112 expect(promises.size).toBe(numberOfWorkers * maxMultiplier)
113 for (const workerNode of queuePool.workerNodes) {
114 expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
115 expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
116 queuePool.opts.tasksQueueOptions.concurrency
117 )
118 expect(workerNode.usage.tasks.executed).toBe(0)
119 expect(workerNode.usage.tasks.queued).toBe(
120 maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
121 )
122 expect(workerNode.usage.tasks.maxQueued).toBe(
123 maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
124 )
125 }
126 expect(queuePool.info.executingTasks).toBe(
127 numberOfWorkers * queuePool.opts.tasksQueueOptions.concurrency
128 )
129 expect(queuePool.info.queuedTasks).toBe(
130 numberOfWorkers *
131 (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
132 )
133 expect(queuePool.info.maxQueuedTasks).toBe(
134 numberOfWorkers *
135 (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
136 )
137 await Promise.all(promises)
138 for (const workerNode of queuePool.workerNodes) {
139 expect(workerNode.usage.tasks.executing).toBe(0)
140 expect(workerNode.usage.tasks.executed).toBe(maxMultiplier)
141 expect(workerNode.usage.tasks.queued).toBe(0)
142 expect(workerNode.usage.tasks.maxQueued).toBe(
143 maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
144 )
145 }
146 })
147
148 it('Verify that is possible to have a worker that return undefined', async () => {
149 const result = await emptyPool.execute()
150 expect(result).toBeUndefined()
151 })
152
153 it('Verify that data are sent to the worker correctly', async () => {
154 const data = { f: 10 }
155 const result = await echoPool.execute(data)
156 expect(result).toStrictEqual(data)
157 })
158
159 it('Verify that error handling is working properly:sync', async () => {
160 const data = { f: 10 }
161 let taskError
162 errorPool.emitter.on(PoolEvents.taskError, (e) => {
163 taskError = e
164 })
165 let inError
166 try {
167 await errorPool.execute(data)
168 } catch (e) {
169 inError = e
170 }
171 expect(inError).toBeDefined()
172 expect(typeof inError === 'string').toBe(true)
173 expect(inError).toBe('Error Message from ClusterWorker')
174 expect(taskError).toStrictEqual({
175 name: 'default',
176 message: 'Error Message from ClusterWorker',
177 data
178 })
179 expect(
180 errorPool.workerNodes.some(
181 (workerNode) => workerNode.usage.tasks.failed === 1
182 )
183 ).toBe(true)
184 })
185
186 it('Verify that error handling is working properly:async', async () => {
187 const data = { f: 10 }
188 let taskError
189 asyncErrorPool.emitter.on(PoolEvents.taskError, (e) => {
190 taskError = e
191 })
192 let inError
193 try {
194 await asyncErrorPool.execute(data)
195 } catch (e) {
196 inError = e
197 }
198 expect(inError).toBeDefined()
199 expect(typeof inError === 'string').toBe(true)
200 expect(inError).toBe('Error Message from ClusterWorker:async')
201 expect(taskError).toStrictEqual({
202 name: 'default',
203 message: 'Error Message from ClusterWorker:async',
204 data
205 })
206 expect(
207 asyncErrorPool.workerNodes.some(
208 (workerNode) => workerNode.usage.tasks.failed === 1
209 )
210 ).toBe(true)
211 })
212
213 it('Verify that async function is working properly', async () => {
214 const data = { f: 10 }
215 const startTime = performance.now()
216 const result = await asyncPool.execute(data)
217 const usedTime = performance.now() - startTime
218 expect(result).toStrictEqual(data)
219 expect(usedTime).toBeGreaterThanOrEqual(2000)
220 })
221
222 it('Shutdown test', async () => {
223 const exitPromise = waitWorkerEvents(pool, 'exit', numberOfWorkers)
224 let poolDestroy = 0
225 pool.emitter.on(PoolEvents.destroy, () => ++poolDestroy)
226 await pool.destroy()
227 const numberOfExitEvents = await exitPromise
228 expect(numberOfExitEvents).toBe(numberOfWorkers)
229 expect(poolDestroy).toBe(1)
230 })
231
232 it('Verify that cluster pool options are checked', async () => {
233 const workerFilePath = './tests/worker-files/cluster/testWorker.js'
234 let pool1 = new FixedClusterPool(numberOfWorkers, workerFilePath)
235 expect(pool1.opts.env).toBeUndefined()
236 expect(pool1.opts.settings).toBeUndefined()
237 await pool1.destroy()
238 pool1 = new FixedClusterPool(numberOfWorkers, workerFilePath, {
239 env: { TEST: 'test' },
240 settings: { args: ['--use', 'http'], silent: true }
241 })
242 expect(pool1.opts.env).toStrictEqual({ TEST: 'test' })
243 expect(pool1.opts.settings).toStrictEqual({
244 args: ['--use', 'http'],
245 silent: true
246 })
247 expect({ ...pool1.opts.settings, exec: workerFilePath }).toStrictEqual({
248 args: ['--use', 'http'],
249 silent: true,
250 exec: workerFilePath
251 })
252 await pool1.destroy()
253 })
254
255 it('Should work even without opts in input', async () => {
256 const pool1 = new FixedClusterPool(
257 numberOfWorkers,
258 './tests/worker-files/cluster/testWorker.js'
259 )
260 const res = await pool1.execute()
261 expect(res).toStrictEqual({ ok: 1 })
262 // We need to clean up the resources after our test
263 await pool1.destroy()
264 })
265
266 it('Verify that a pool with zero worker fails', async () => {
267 expect(
268 () =>
269 new FixedClusterPool(0, './tests/worker-files/cluster/testWorker.js')
270 ).toThrowError('Cannot instantiate a fixed pool with zero worker')
271 })
272 })