build: switch from prettier to rome as code formatter
[poolifier.git] / tests / pools / thread / fixed.test.js
1 const { expect } = require('expect')
2 const { FixedThreadPool, PoolEvents } = require('../../../lib')
3 const { TaskFunctions } = require('../../test-types')
4 const { waitPoolEvents, waitWorkerEvents } = require('../../test-utils')
5
6 describe('Fixed thread pool test suite', () => {
7 const numberOfThreads = 6
8 const tasksConcurrency = 2
9 const pool = new FixedThreadPool(
10 numberOfThreads,
11 './tests/worker-files/thread/testWorker.js',
12 {
13 errorHandler: (e) => console.error(e)
14 }
15 )
16 const queuePool = new FixedThreadPool(
17 numberOfThreads,
18 './tests/worker-files/thread/testWorker.js',
19 {
20 enableTasksQueue: true,
21 tasksQueueOptions: {
22 concurrency: tasksConcurrency
23 },
24 errorHandler: (e) => console.error(e)
25 }
26 )
27 const emptyPool = new FixedThreadPool(
28 numberOfThreads,
29 './tests/worker-files/thread/emptyWorker.js',
30 { exitHandler: () => console.info('empty pool worker exited') }
31 )
32 const echoPool = new FixedThreadPool(
33 numberOfThreads,
34 './tests/worker-files/thread/echoWorker.js'
35 )
36 const errorPool = new FixedThreadPool(
37 numberOfThreads,
38 './tests/worker-files/thread/errorWorker.js',
39 {
40 errorHandler: (e) => console.error(e)
41 }
42 )
43 const asyncErrorPool = new FixedThreadPool(
44 numberOfThreads,
45 './tests/worker-files/thread/asyncErrorWorker.js',
46 {
47 errorHandler: (e) => console.error(e)
48 }
49 )
50 const asyncPool = new FixedThreadPool(
51 numberOfThreads,
52 './tests/worker-files/thread/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 thread', 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 FixedThreadPool(
83 numberOfThreads,
84 './tests/worker-files/thread/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, '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 < numberOfThreads * 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 numberOfThreads + 1 times for a loop submitting up to numberOfThreads * 2 tasks to the fixed pool.
103 expect(poolBusy).toBe(numberOfThreads + 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 < numberOfThreads * maxMultiplier; i++) {
110 promises.add(queuePool.execute())
111 }
112 expect(promises.size).toBe(numberOfThreads * maxMultiplier)
113 for (const workerNode of queuePool.workerNodes) {
114 expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
115 queuePool.opts.tasksQueueOptions.concurrency
116 )
117 expect(workerNode.usage.tasks.executed).toBe(0)
118 expect(workerNode.usage.tasks.queued).toBe(
119 maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
120 )
121 expect(workerNode.usage.tasks.maxQueued).toBe(
122 maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
123 )
124 }
125 expect(queuePool.info.executingTasks).toBe(
126 numberOfThreads * queuePool.opts.tasksQueueOptions.concurrency
127 )
128 expect(queuePool.info.queuedTasks).toBe(
129 numberOfThreads *
130 (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
131 )
132 expect(queuePool.info.maxQueuedTasks).toBe(
133 numberOfThreads *
134 (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
135 )
136 await Promise.all(promises)
137 for (const workerNode of queuePool.workerNodes) {
138 expect(workerNode.usage.tasks.executing).toBe(0)
139 expect(workerNode.usage.tasks.executed).toBe(maxMultiplier)
140 expect(workerNode.usage.tasks.queued).toBe(0)
141 expect(workerNode.usage.tasks.maxQueued).toBe(
142 maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
143 )
144 }
145 })
146
147 it('Verify that is possible to have a worker that return undefined', async () => {
148 const result = await emptyPool.execute()
149 expect(result).toBeUndefined()
150 })
151
152 it('Verify that data are sent to the worker correctly', async () => {
153 const data = { f: 10 }
154 const result = await echoPool.execute(data)
155 expect(result).toStrictEqual(data)
156 })
157
158 it('Verify that transferable objects are sent to the worker correctly', async () => {
159 let error
160 let result
161 try {
162 result = await pool.execute(undefined, undefined, [
163 new ArrayBuffer(16),
164 new MessageChannel().port1
165 ])
166 } catch (e) {
167 error = e
168 }
169 expect(result).toStrictEqual({ ok: 1 })
170 expect(error).toBeUndefined()
171 try {
172 result = await pool.execute(undefined, undefined, [
173 new SharedArrayBuffer(16)
174 ])
175 } catch (e) {
176 error = e
177 }
178 expect(result).toStrictEqual({ ok: 1 })
179 expect(error).toStrictEqual(
180 new TypeError('Found invalid object in transferList')
181 )
182 })
183
184 it('Verify that error handling is working properly:sync', async () => {
185 const data = { f: 10 }
186 let taskError
187 errorPool.emitter.on(PoolEvents.taskError, (e) => {
188 taskError = e
189 })
190 let inError
191 try {
192 await errorPool.execute(data)
193 } catch (e) {
194 inError = e
195 }
196 expect(inError).toBeDefined()
197 expect(inError).toBeInstanceOf(Error)
198 expect(inError.message).toBeDefined()
199 expect(typeof inError.message === 'string').toBe(true)
200 expect(inError.message).toBe('Error Message from ThreadWorker')
201 expect(taskError).toStrictEqual({
202 name: 'default',
203 message: new Error('Error Message from ThreadWorker'),
204 data
205 })
206 expect(
207 errorPool.workerNodes.some(
208 (workerNode) => workerNode.usage.tasks.failed === 1
209 )
210 ).toBe(true)
211 })
212
213 it('Verify that error handling is working properly:async', async () => {
214 const data = { f: 10 }
215 let taskError
216 asyncErrorPool.emitter.on(PoolEvents.taskError, (e) => {
217 taskError = e
218 })
219 let inError
220 try {
221 await asyncErrorPool.execute(data)
222 } catch (e) {
223 inError = e
224 }
225 expect(inError).toBeDefined()
226 expect(inError).toBeInstanceOf(Error)
227 expect(inError.message).toBeDefined()
228 expect(typeof inError.message === 'string').toBe(true)
229 expect(inError.message).toBe('Error Message from ThreadWorker:async')
230 expect(taskError).toStrictEqual({
231 name: 'default',
232 message: new Error('Error Message from ThreadWorker:async'),
233 data
234 })
235 expect(
236 asyncErrorPool.workerNodes.some(
237 (workerNode) => workerNode.usage.tasks.failed === 1
238 )
239 ).toBe(true)
240 })
241
242 it('Verify that async function is working properly', async () => {
243 const data = { f: 10 }
244 const startTime = performance.now()
245 const result = await asyncPool.execute(data)
246 const usedTime = performance.now() - startTime
247 expect(result).toStrictEqual(data)
248 expect(usedTime).toBeGreaterThanOrEqual(2000)
249 })
250
251 it('Shutdown test', async () => {
252 const exitPromise = waitWorkerEvents(pool, 'exit', numberOfThreads)
253 await pool.destroy()
254 const numberOfExitEvents = await exitPromise
255 expect(numberOfExitEvents).toBe(numberOfThreads)
256 })
257
258 it('Verify that thread pool options are checked', async () => {
259 const workerFilePath = './tests/worker-files/thread/testWorker.js'
260 let pool1 = new FixedThreadPool(numberOfThreads, workerFilePath)
261 expect(pool1.opts.workerOptions).toBeUndefined()
262 await pool1.destroy()
263 pool1 = new FixedThreadPool(numberOfThreads, workerFilePath, {
264 workerOptions: {
265 env: { TEST: 'test' },
266 name: 'test'
267 }
268 })
269 expect(pool1.opts.workerOptions).toStrictEqual({
270 env: { TEST: 'test' },
271 name: 'test'
272 })
273 await pool1.destroy()
274 })
275
276 it('Should work even without opts in input', async () => {
277 const pool1 = new FixedThreadPool(
278 numberOfThreads,
279 './tests/worker-files/thread/testWorker.js'
280 )
281 const res = await pool1.execute()
282 expect(res).toStrictEqual({ ok: 1 })
283 // We need to clean up the resources after our test
284 await pool1.destroy()
285 })
286
287 it('Verify that a pool with zero worker fails', async () => {
288 expect(
289 () => new FixedThreadPool(0, './tests/worker-files/thread/testWorker.js')
290 ).toThrowError('Cannot instantiate a fixed pool with zero worker')
291 })
292 })