test: improve coverage
[poolifier.git] / tests / pools / abstract / abstract-pool.test.js
CommitLineData
a61a0724 1const { expect } = require('expect')
b1aae695 2const sinon = require('sinon')
e843b904 3const {
70a4f5ea 4 DynamicClusterPool,
9e619829 5 DynamicThreadPool,
aee46736 6 FixedClusterPool,
e843b904 7 FixedThreadPool,
aee46736 8 PoolEvents,
184855e6 9 PoolTypes,
3d6dd312 10 WorkerChoiceStrategies,
184855e6 11 WorkerTypes
cdace0e5 12} = require('../../../lib')
78099a15 13const { CircularArray } = require('../../../lib/circular-array')
574b351d 14const { Deque } = require('../../../lib/deque')
23ccf9d7 15const { version } = require('../../../package.json')
2431bdb4 16const { waitPoolEvents } = require('../../test-utils')
e1ffb94f
JB
17
18describe('Abstract pool test suite', () => {
fc027381 19 const numberOfWorkers = 2
a8884ffd 20 class StubPoolWithIsMain extends FixedThreadPool {
e1ffb94f
JB
21 isMain () {
22 return false
23 }
3ec964d6 24 }
3ec964d6 25
dc021bcc
JB
26 afterEach(() => {
27 sinon.restore()
28 })
29
3ec964d6 30 it('Simulate pool creation from a non main thread/process', () => {
8d3782fa
JB
31 expect(
32 () =>
a8884ffd 33 new StubPoolWithIsMain(
7c0ba920 34 numberOfWorkers,
8d3782fa
JB
35 './tests/worker-files/thread/testWorker.js',
36 {
8ebe6c30 37 errorHandler: (e) => console.error(e)
8d3782fa
JB
38 }
39 )
04f45163 40 ).toThrowError(
e695d66f
JB
41 new Error(
42 'Cannot start a pool from a worker with the same type as the pool'
43 )
04f45163 44 )
3ec964d6 45 })
c510fea7
APA
46
47 it('Verify that filePath is checked', () => {
292ad316
JB
48 const expectedError = new Error(
49 'Please specify a file with a worker implementation'
50 )
7c0ba920 51 expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError(
292ad316 52 expectedError
8d3782fa 53 )
7c0ba920 54 expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError(
292ad316 55 expectedError
8d3782fa 56 )
3d6dd312
JB
57 expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrowError(
58 expectedError
59 )
60 expect(() => new FixedThreadPool(numberOfWorkers, true)).toThrowError(
61 expectedError
62 )
63 expect(
64 () => new FixedThreadPool(numberOfWorkers, './dummyWorker.ts')
65 ).toThrowError(new Error("Cannot find the worker file './dummyWorker.ts'"))
8d3782fa
JB
66 })
67
68 it('Verify that numberOfWorkers is checked', () => {
69 expect(() => new FixedThreadPool()).toThrowError(
e695d66f
JB
70 new Error(
71 'Cannot instantiate a pool without specifying the number of workers'
72 )
8d3782fa
JB
73 )
74 })
75
76 it('Verify that a negative number of workers is checked', () => {
77 expect(
78 () =>
79 new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
80 ).toThrowError(
473c717a
JB
81 new RangeError(
82 'Cannot instantiate a pool with a negative number of workers'
83 )
8d3782fa
JB
84 )
85 })
86
87 it('Verify that a non integer number of workers is checked', () => {
88 expect(
89 () =>
90 new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.js')
91 ).toThrowError(
473c717a 92 new TypeError(
0d80593b 93 'Cannot instantiate a pool with a non safe integer number of workers'
8d3782fa
JB
94 )
95 )
c510fea7 96 })
7c0ba920 97
216541b6 98 it('Verify that dynamic pool sizing is checked', () => {
a5ed75b7
JB
99 expect(
100 () =>
101 new DynamicClusterPool(
102 1,
103 undefined,
104 './tests/worker-files/cluster/testWorker.js'
105 )
106 ).toThrowError(
107 new TypeError(
108 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
109 )
110 )
2761efb4
JB
111 expect(
112 () =>
113 new DynamicThreadPool(
114 0.5,
115 1,
116 './tests/worker-files/thread/testWorker.js'
117 )
118 ).toThrowError(
119 new TypeError(
120 'Cannot instantiate a pool with a non safe integer number of workers'
121 )
122 )
123 expect(
124 () =>
125 new DynamicClusterPool(
126 0,
127 0.5,
a5ed75b7 128 './tests/worker-files/cluster/testWorker.js'
2761efb4
JB
129 )
130 ).toThrowError(
131 new TypeError(
132 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
133 )
134 )
2431bdb4
JB
135 expect(
136 () =>
137 new DynamicThreadPool(2, 1, './tests/worker-files/thread/testWorker.js')
138 ).toThrowError(
139 new RangeError(
140 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
141 )
142 )
143 expect(
144 () =>
2761efb4
JB
145 new DynamicClusterPool(
146 1,
147 1,
a5ed75b7 148 './tests/worker-files/cluster/testWorker.js'
2761efb4 149 )
2431bdb4
JB
150 ).toThrowError(
151 new RangeError(
152 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
153 )
154 )
21f710aa
JB
155 expect(
156 () =>
157 new DynamicThreadPool(0, 0, './tests/worker-files/thread/testWorker.js')
158 ).toThrowError(
159 new RangeError(
d640b48b 160 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
21f710aa
JB
161 )
162 )
2431bdb4
JB
163 })
164
fd7ebd49 165 it('Verify that pool options are checked', async () => {
7c0ba920
JB
166 let pool = new FixedThreadPool(
167 numberOfWorkers,
168 './tests/worker-files/thread/testWorker.js'
169 )
7c0ba920 170 expect(pool.emitter).toBeDefined()
1f68cede
JB
171 expect(pool.opts.enableEvents).toBe(true)
172 expect(pool.opts.restartWorkerOnError).toBe(true)
ff733df7 173 expect(pool.opts.enableTasksQueue).toBe(false)
d4aeae5a 174 expect(pool.opts.tasksQueueOptions).toBeUndefined()
e843b904
JB
175 expect(pool.opts.workerChoiceStrategy).toBe(
176 WorkerChoiceStrategies.ROUND_ROBIN
177 )
da309861 178 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d
JB
179 choiceRetries: 6,
180 runTime: { median: false },
181 waitTime: { median: false },
182 elu: { median: false }
183 })
184 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
185 choiceRetries: 6,
932fc8be 186 runTime: { median: false },
5df69fab
JB
187 waitTime: { median: false },
188 elu: { median: false }
da309861 189 })
35cf1c03
JB
190 expect(pool.opts.messageHandler).toBeUndefined()
191 expect(pool.opts.errorHandler).toBeUndefined()
192 expect(pool.opts.onlineHandler).toBeUndefined()
193 expect(pool.opts.exitHandler).toBeUndefined()
fd7ebd49 194 await pool.destroy()
73bfd59d 195 const testHandler = () => console.info('test handler executed')
7c0ba920
JB
196 pool = new FixedThreadPool(
197 numberOfWorkers,
198 './tests/worker-files/thread/testWorker.js',
199 {
e4543b14 200 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
49be33fe 201 workerChoiceStrategyOptions: {
932fc8be 202 runTime: { median: true },
fc027381 203 weights: { 0: 300, 1: 200 }
49be33fe 204 },
35cf1c03 205 enableEvents: false,
1f68cede 206 restartWorkerOnError: false,
ff733df7 207 enableTasksQueue: true,
d4aeae5a 208 tasksQueueOptions: { concurrency: 2 },
35cf1c03
JB
209 messageHandler: testHandler,
210 errorHandler: testHandler,
211 onlineHandler: testHandler,
212 exitHandler: testHandler
7c0ba920
JB
213 }
214 )
7c0ba920 215 expect(pool.emitter).toBeUndefined()
1f68cede
JB
216 expect(pool.opts.enableEvents).toBe(false)
217 expect(pool.opts.restartWorkerOnError).toBe(false)
ff733df7 218 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
219 expect(pool.opts.tasksQueueOptions).toStrictEqual({
220 concurrency: 2,
ff3f866a 221 size: 4
20c6f652 222 })
e843b904 223 expect(pool.opts.workerChoiceStrategy).toBe(
e4543b14 224 WorkerChoiceStrategies.LEAST_USED
e843b904 225 )
da309861 226 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d 227 choiceRetries: 6,
932fc8be 228 runTime: { median: true },
8990357d
JB
229 waitTime: { median: false },
230 elu: { median: false },
231 weights: { 0: 300, 1: 200 }
232 })
233 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
234 choiceRetries: 6,
235 runTime: { median: true },
236 waitTime: { median: false },
237 elu: { median: false },
fc027381 238 weights: { 0: 300, 1: 200 }
da309861 239 })
35cf1c03
JB
240 expect(pool.opts.messageHandler).toStrictEqual(testHandler)
241 expect(pool.opts.errorHandler).toStrictEqual(testHandler)
242 expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
243 expect(pool.opts.exitHandler).toStrictEqual(testHandler)
fd7ebd49 244 await pool.destroy()
7c0ba920
JB
245 })
246
a20f0ba5 247 it('Verify that pool options are validated', async () => {
d4aeae5a
JB
248 expect(
249 () =>
250 new FixedThreadPool(
251 numberOfWorkers,
252 './tests/worker-files/thread/testWorker.js',
253 {
f0d7f803 254 workerChoiceStrategy: 'invalidStrategy'
d4aeae5a
JB
255 }
256 )
8735b4e5
JB
257 ).toThrowError(
258 new Error("Invalid worker choice strategy 'invalidStrategy'")
259 )
7e653ee0
JB
260 expect(
261 () =>
262 new FixedThreadPool(
263 numberOfWorkers,
264 './tests/worker-files/thread/testWorker.js',
265 {
266 workerChoiceStrategyOptions: {
267 choiceRetries: 'invalidChoiceRetries'
268 }
269 }
270 )
271 ).toThrowError(
272 new TypeError(
273 'Invalid worker choice strategy options: choice retries must be an integer'
274 )
275 )
276 expect(
277 () =>
278 new FixedThreadPool(
279 numberOfWorkers,
280 './tests/worker-files/thread/testWorker.js',
281 {
282 workerChoiceStrategyOptions: {
283 choiceRetries: -1
284 }
285 }
286 )
287 ).toThrowError(
288 new RangeError(
289 "Invalid worker choice strategy options: choice retries '-1' must be greater or equal than zero"
290 )
291 )
49be33fe
JB
292 expect(
293 () =>
294 new FixedThreadPool(
295 numberOfWorkers,
296 './tests/worker-files/thread/testWorker.js',
297 {
298 workerChoiceStrategyOptions: { weights: {} }
299 }
300 )
301 ).toThrowError(
8735b4e5
JB
302 new Error(
303 'Invalid worker choice strategy options: must have a weight for each worker node'
304 )
49be33fe 305 )
f0d7f803
JB
306 expect(
307 () =>
308 new FixedThreadPool(
309 numberOfWorkers,
310 './tests/worker-files/thread/testWorker.js',
311 {
312 workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
313 }
314 )
315 ).toThrowError(
8735b4e5
JB
316 new Error(
317 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
318 )
f0d7f803
JB
319 )
320 expect(
321 () =>
322 new FixedThreadPool(
323 numberOfWorkers,
324 './tests/worker-files/thread/testWorker.js',
325 {
326 enableTasksQueue: true,
327 tasksQueueOptions: { concurrency: 0 }
328 }
329 )
8735b4e5 330 ).toThrowError(
e695d66f 331 new RangeError(
20c6f652 332 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
333 )
334 )
f0d7f803
JB
335 expect(
336 () =>
337 new FixedThreadPool(
338 numberOfWorkers,
339 './tests/worker-files/thread/testWorker.js',
340 {
341 enableTasksQueue: true,
342 tasksQueueOptions: 'invalidTasksQueueOptions'
343 }
344 )
8735b4e5
JB
345 ).toThrowError(
346 new TypeError('Invalid tasks queue options: must be a plain object')
347 )
f0d7f803
JB
348 expect(
349 () =>
350 new FixedThreadPool(
351 numberOfWorkers,
352 './tests/worker-files/thread/testWorker.js',
353 {
354 enableTasksQueue: true,
355 tasksQueueOptions: { concurrency: 0.2 }
356 }
357 )
8735b4e5 358 ).toThrowError(
20c6f652 359 new TypeError('Invalid worker node tasks concurrency: must be an integer')
8735b4e5 360 )
d4aeae5a
JB
361 })
362
2431bdb4 363 it('Verify that pool worker choice strategy options can be set', async () => {
a20f0ba5
JB
364 const pool = new FixedThreadPool(
365 numberOfWorkers,
366 './tests/worker-files/thread/testWorker.js',
367 { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
368 )
369 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d
JB
370 choiceRetries: 6,
371 runTime: { median: false },
372 waitTime: { median: false },
373 elu: { median: false }
374 })
375 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
376 choiceRetries: 6,
932fc8be 377 runTime: { median: false },
5df69fab
JB
378 waitTime: { median: false },
379 elu: { median: false }
a20f0ba5
JB
380 })
381 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
382 .workerChoiceStrategies) {
86bf340d 383 expect(workerChoiceStrategy.opts).toStrictEqual({
8990357d 384 choiceRetries: 6,
932fc8be 385 runTime: { median: false },
5df69fab
JB
386 waitTime: { median: false },
387 elu: { median: false }
86bf340d 388 })
a20f0ba5 389 }
87de9ff5
JB
390 expect(
391 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
392 ).toStrictEqual({
932fc8be
JB
393 runTime: {
394 aggregate: true,
395 average: true,
396 median: false
397 },
398 waitTime: {
399 aggregate: false,
400 average: false,
401 median: false
402 },
5df69fab 403 elu: {
9adcefab
JB
404 aggregate: true,
405 average: true,
5df69fab
JB
406 median: false
407 }
86bf340d 408 })
9adcefab
JB
409 pool.setWorkerChoiceStrategyOptions({
410 runTime: { median: true },
411 elu: { median: true }
412 })
a20f0ba5 413 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d 414 choiceRetries: 6,
9adcefab 415 runTime: { median: true },
8990357d
JB
416 waitTime: { median: false },
417 elu: { median: true }
418 })
419 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
420 choiceRetries: 6,
421 runTime: { median: true },
422 waitTime: { median: false },
9adcefab 423 elu: { median: true }
a20f0ba5
JB
424 })
425 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
426 .workerChoiceStrategies) {
932fc8be 427 expect(workerChoiceStrategy.opts).toStrictEqual({
8990357d 428 choiceRetries: 6,
9adcefab 429 runTime: { median: true },
8990357d 430 waitTime: { median: false },
9adcefab 431 elu: { median: true }
932fc8be 432 })
a20f0ba5 433 }
87de9ff5
JB
434 expect(
435 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
436 ).toStrictEqual({
932fc8be
JB
437 runTime: {
438 aggregate: true,
439 average: false,
440 median: true
441 },
442 waitTime: {
443 aggregate: false,
444 average: false,
445 median: false
446 },
5df69fab 447 elu: {
9adcefab 448 aggregate: true,
5df69fab 449 average: false,
9adcefab 450 median: true
5df69fab 451 }
86bf340d 452 })
9adcefab
JB
453 pool.setWorkerChoiceStrategyOptions({
454 runTime: { median: false },
455 elu: { median: false }
456 })
a20f0ba5 457 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d
JB
458 choiceRetries: 6,
459 runTime: { median: false },
460 waitTime: { median: false },
461 elu: { median: false }
462 })
463 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
464 choiceRetries: 6,
9adcefab 465 runTime: { median: false },
8990357d 466 waitTime: { median: false },
9adcefab 467 elu: { median: false }
a20f0ba5
JB
468 })
469 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
470 .workerChoiceStrategies) {
932fc8be 471 expect(workerChoiceStrategy.opts).toStrictEqual({
8990357d 472 choiceRetries: 6,
9adcefab 473 runTime: { median: false },
8990357d 474 waitTime: { median: false },
9adcefab 475 elu: { median: false }
932fc8be 476 })
a20f0ba5 477 }
87de9ff5
JB
478 expect(
479 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
480 ).toStrictEqual({
932fc8be
JB
481 runTime: {
482 aggregate: true,
483 average: true,
484 median: false
485 },
486 waitTime: {
487 aggregate: false,
488 average: false,
489 median: false
490 },
5df69fab 491 elu: {
9adcefab
JB
492 aggregate: true,
493 average: true,
5df69fab
JB
494 median: false
495 }
86bf340d 496 })
1f95d544
JB
497 expect(() =>
498 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
499 ).toThrowError(
8735b4e5
JB
500 new TypeError(
501 'Invalid worker choice strategy options: must be a plain object'
502 )
1f95d544 503 )
7e653ee0
JB
504 expect(() =>
505 pool.setWorkerChoiceStrategyOptions({
506 choiceRetries: 'invalidChoiceRetries'
507 })
508 ).toThrowError(
509 new TypeError(
510 'Invalid worker choice strategy options: choice retries must be an integer'
511 )
512 )
513 expect(() =>
514 pool.setWorkerChoiceStrategyOptions({ choiceRetries: -1 })
515 ).toThrowError(
516 new RangeError(
517 "Invalid worker choice strategy options: choice retries '-1' must be greater or equal than zero"
518 )
519 )
1f95d544
JB
520 expect(() =>
521 pool.setWorkerChoiceStrategyOptions({ weights: {} })
522 ).toThrowError(
8735b4e5
JB
523 new Error(
524 'Invalid worker choice strategy options: must have a weight for each worker node'
525 )
1f95d544
JB
526 )
527 expect(() =>
528 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
529 ).toThrowError(
8735b4e5
JB
530 new Error(
531 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
532 )
1f95d544 533 )
a20f0ba5
JB
534 await pool.destroy()
535 })
536
2431bdb4 537 it('Verify that pool tasks queue can be enabled/disabled', async () => {
a20f0ba5
JB
538 const pool = new FixedThreadPool(
539 numberOfWorkers,
540 './tests/worker-files/thread/testWorker.js'
541 )
542 expect(pool.opts.enableTasksQueue).toBe(false)
543 expect(pool.opts.tasksQueueOptions).toBeUndefined()
544 pool.enableTasksQueue(true)
545 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
546 expect(pool.opts.tasksQueueOptions).toStrictEqual({
547 concurrency: 1,
ff3f866a 548 size: 4
20c6f652 549 })
a20f0ba5
JB
550 pool.enableTasksQueue(true, { concurrency: 2 })
551 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
552 expect(pool.opts.tasksQueueOptions).toStrictEqual({
553 concurrency: 2,
ff3f866a 554 size: 4
20c6f652 555 })
a20f0ba5
JB
556 pool.enableTasksQueue(false)
557 expect(pool.opts.enableTasksQueue).toBe(false)
558 expect(pool.opts.tasksQueueOptions).toBeUndefined()
559 await pool.destroy()
560 })
561
2431bdb4 562 it('Verify that pool tasks queue options can be set', async () => {
a20f0ba5
JB
563 const pool = new FixedThreadPool(
564 numberOfWorkers,
565 './tests/worker-files/thread/testWorker.js',
566 { enableTasksQueue: true }
567 )
20c6f652
JB
568 expect(pool.opts.tasksQueueOptions).toStrictEqual({
569 concurrency: 1,
ff3f866a 570 size: 4
20c6f652 571 })
a20f0ba5 572 pool.setTasksQueueOptions({ concurrency: 2 })
20c6f652
JB
573 expect(pool.opts.tasksQueueOptions).toStrictEqual({
574 concurrency: 2,
ff3f866a 575 size: 4
20c6f652 576 })
f0d7f803
JB
577 expect(() =>
578 pool.setTasksQueueOptions('invalidTasksQueueOptions')
8735b4e5
JB
579 ).toThrowError(
580 new TypeError('Invalid tasks queue options: must be a plain object')
581 )
a20f0ba5 582 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
e695d66f 583 new RangeError(
20c6f652 584 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
585 )
586 )
587 expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError(
e695d66f 588 new RangeError(
20c6f652 589 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
8735b4e5 590 )
a20f0ba5 591 )
f0d7f803 592 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
20c6f652
JB
593 new TypeError('Invalid worker node tasks concurrency: must be an integer')
594 )
68dbcdc0
JB
595 expect(() => pool.setTasksQueueOptions({ queueMaxSize: 2 })).toThrowError(
596 new Error(
597 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
598 )
599 )
ff3f866a 600 expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
20c6f652 601 new RangeError(
68dbcdc0 602 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
20c6f652
JB
603 )
604 )
ff3f866a 605 expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError(
20c6f652 606 new RangeError(
68dbcdc0 607 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
20c6f652
JB
608 )
609 )
ff3f866a 610 expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError(
68dbcdc0 611 new TypeError('Invalid worker node tasks queue size: must be an integer')
f0d7f803 612 )
a20f0ba5
JB
613 await pool.destroy()
614 })
615
6b27d407
JB
616 it('Verify that pool info is set', async () => {
617 let pool = new FixedThreadPool(
618 numberOfWorkers,
619 './tests/worker-files/thread/testWorker.js'
620 )
2dca6cab
JB
621 expect(pool.info).toStrictEqual({
622 version,
623 type: PoolTypes.fixed,
624 worker: WorkerTypes.thread,
625 ready: true,
626 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
627 minSize: numberOfWorkers,
628 maxSize: numberOfWorkers,
629 workerNodes: numberOfWorkers,
630 idleWorkerNodes: numberOfWorkers,
631 busyWorkerNodes: 0,
632 executedTasks: 0,
633 executingTasks: 0,
2dca6cab
JB
634 failedTasks: 0
635 })
6b27d407
JB
636 await pool.destroy()
637 pool = new DynamicClusterPool(
2431bdb4 638 Math.floor(numberOfWorkers / 2),
6b27d407 639 numberOfWorkers,
ecdfbdc0 640 './tests/worker-files/cluster/testWorker.js'
6b27d407 641 )
2dca6cab
JB
642 expect(pool.info).toStrictEqual({
643 version,
644 type: PoolTypes.dynamic,
645 worker: WorkerTypes.cluster,
646 ready: true,
647 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
648 minSize: Math.floor(numberOfWorkers / 2),
649 maxSize: numberOfWorkers,
650 workerNodes: Math.floor(numberOfWorkers / 2),
651 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
652 busyWorkerNodes: 0,
653 executedTasks: 0,
654 executingTasks: 0,
2dca6cab
JB
655 failedTasks: 0
656 })
6b27d407
JB
657 await pool.destroy()
658 })
659
2431bdb4 660 it('Verify that pool worker tasks usage are initialized', async () => {
bf9549ae
JB
661 const pool = new FixedClusterPool(
662 numberOfWorkers,
663 './tests/worker-files/cluster/testWorker.js'
664 )
f06e48d8 665 for (const workerNode of pool.workerNodes) {
465b2940 666 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
667 tasks: {
668 executed: 0,
669 executing: 0,
670 queued: 0,
df593701 671 maxQueued: 0,
68cbdc84 672 stolen: 0,
a4e07f72
JB
673 failed: 0
674 },
675 runTime: {
a4e07f72
JB
676 history: expect.any(CircularArray)
677 },
678 waitTime: {
a4e07f72
JB
679 history: expect.any(CircularArray)
680 },
5df69fab
JB
681 elu: {
682 idle: {
5df69fab
JB
683 history: expect.any(CircularArray)
684 },
685 active: {
5df69fab 686 history: expect.any(CircularArray)
f7510105 687 }
5df69fab 688 }
86bf340d 689 })
f06e48d8
JB
690 }
691 await pool.destroy()
692 })
693
2431bdb4
JB
694 it('Verify that pool worker tasks queue are initialized', async () => {
695 let pool = new FixedClusterPool(
f06e48d8
JB
696 numberOfWorkers,
697 './tests/worker-files/cluster/testWorker.js'
698 )
699 for (const workerNode of pool.workerNodes) {
700 expect(workerNode.tasksQueue).toBeDefined()
574b351d 701 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
4d8bf9e4 702 expect(workerNode.tasksQueue.size).toBe(0)
9c16fb4b 703 expect(workerNode.tasksQueue.maxSize).toBe(0)
bf9549ae 704 }
fd7ebd49 705 await pool.destroy()
2431bdb4
JB
706 pool = new DynamicThreadPool(
707 Math.floor(numberOfWorkers / 2),
708 numberOfWorkers,
709 './tests/worker-files/thread/testWorker.js'
710 )
711 for (const workerNode of pool.workerNodes) {
712 expect(workerNode.tasksQueue).toBeDefined()
574b351d 713 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
2431bdb4
JB
714 expect(workerNode.tasksQueue.size).toBe(0)
715 expect(workerNode.tasksQueue.maxSize).toBe(0)
716 }
717 })
718
719 it('Verify that pool worker info are initialized', async () => {
720 let pool = new FixedClusterPool(
721 numberOfWorkers,
722 './tests/worker-files/cluster/testWorker.js'
723 )
2dca6cab
JB
724 for (const workerNode of pool.workerNodes) {
725 expect(workerNode.info).toStrictEqual({
726 id: expect.any(Number),
727 type: WorkerTypes.cluster,
728 dynamic: false,
729 ready: true
730 })
731 }
2431bdb4
JB
732 await pool.destroy()
733 pool = new DynamicThreadPool(
734 Math.floor(numberOfWorkers / 2),
735 numberOfWorkers,
736 './tests/worker-files/thread/testWorker.js'
737 )
2dca6cab
JB
738 for (const workerNode of pool.workerNodes) {
739 expect(workerNode.info).toStrictEqual({
740 id: expect.any(Number),
741 type: WorkerTypes.thread,
742 dynamic: false,
7884d183 743 ready: true
2dca6cab
JB
744 })
745 }
bf9549ae
JB
746 })
747
9d2d0da1
JB
748 it('Verify that pool execute() arguments are checked', async () => {
749 const pool = new FixedClusterPool(
750 numberOfWorkers,
751 './tests/worker-files/cluster/testWorker.js'
752 )
753 await expect(pool.execute(undefined, 0)).rejects.toThrowError(
754 new TypeError('name argument must be a string')
755 )
756 await expect(pool.execute(undefined, '')).rejects.toThrowError(
757 new TypeError('name argument must not be an empty string')
758 )
759 await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
760 new TypeError('transferList argument must be an array')
761 )
762 await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
763 "Task function 'unknown' not found"
764 )
765 await pool.destroy()
766 await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
767 new Error('Cannot execute a task on destroyed pool')
768 )
769 })
770
2431bdb4 771 it('Verify that pool worker tasks usage are computed', async () => {
bf9549ae
JB
772 const pool = new FixedClusterPool(
773 numberOfWorkers,
774 './tests/worker-files/cluster/testWorker.js'
775 )
09c2d0d3 776 const promises = new Set()
fc027381
JB
777 const maxMultiplier = 2
778 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 779 promises.add(pool.execute())
bf9549ae 780 }
f06e48d8 781 for (const workerNode of pool.workerNodes) {
465b2940 782 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
783 tasks: {
784 executed: 0,
785 executing: maxMultiplier,
786 queued: 0,
df593701 787 maxQueued: 0,
68cbdc84 788 stolen: 0,
a4e07f72
JB
789 failed: 0
790 },
791 runTime: {
a4e07f72
JB
792 history: expect.any(CircularArray)
793 },
794 waitTime: {
a4e07f72
JB
795 history: expect.any(CircularArray)
796 },
5df69fab
JB
797 elu: {
798 idle: {
5df69fab
JB
799 history: expect.any(CircularArray)
800 },
801 active: {
5df69fab 802 history: expect.any(CircularArray)
f7510105 803 }
5df69fab 804 }
86bf340d 805 })
bf9549ae
JB
806 }
807 await Promise.all(promises)
f06e48d8 808 for (const workerNode of pool.workerNodes) {
465b2940 809 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
810 tasks: {
811 executed: maxMultiplier,
812 executing: 0,
813 queued: 0,
df593701 814 maxQueued: 0,
68cbdc84 815 stolen: 0,
a4e07f72
JB
816 failed: 0
817 },
818 runTime: {
a4e07f72
JB
819 history: expect.any(CircularArray)
820 },
821 waitTime: {
a4e07f72
JB
822 history: expect.any(CircularArray)
823 },
5df69fab
JB
824 elu: {
825 idle: {
5df69fab
JB
826 history: expect.any(CircularArray)
827 },
828 active: {
5df69fab 829 history: expect.any(CircularArray)
f7510105 830 }
5df69fab 831 }
86bf340d 832 })
bf9549ae 833 }
fd7ebd49 834 await pool.destroy()
bf9549ae
JB
835 })
836
2431bdb4 837 it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
7fd82a1c 838 const pool = new DynamicThreadPool(
2431bdb4 839 Math.floor(numberOfWorkers / 2),
8f4878b7 840 numberOfWorkers,
9e619829
JB
841 './tests/worker-files/thread/testWorker.js'
842 )
09c2d0d3 843 const promises = new Set()
ee9f5295
JB
844 const maxMultiplier = 2
845 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 846 promises.add(pool.execute())
9e619829
JB
847 }
848 await Promise.all(promises)
f06e48d8 849 for (const workerNode of pool.workerNodes) {
465b2940 850 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
851 tasks: {
852 executed: expect.any(Number),
853 executing: 0,
854 queued: 0,
df593701 855 maxQueued: 0,
68cbdc84 856 stolen: 0,
a4e07f72
JB
857 failed: 0
858 },
859 runTime: {
a4e07f72
JB
860 history: expect.any(CircularArray)
861 },
862 waitTime: {
a4e07f72
JB
863 history: expect.any(CircularArray)
864 },
5df69fab
JB
865 elu: {
866 idle: {
5df69fab
JB
867 history: expect.any(CircularArray)
868 },
869 active: {
5df69fab 870 history: expect.any(CircularArray)
f7510105 871 }
5df69fab 872 }
86bf340d 873 })
465b2940 874 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
94407def
JB
875 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
876 numberOfWorkers * maxMultiplier
877 )
b97d82d8
JB
878 expect(workerNode.usage.runTime.history.length).toBe(0)
879 expect(workerNode.usage.waitTime.history.length).toBe(0)
880 expect(workerNode.usage.elu.idle.history.length).toBe(0)
881 expect(workerNode.usage.elu.active.history.length).toBe(0)
9e619829
JB
882 }
883 pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
f06e48d8 884 for (const workerNode of pool.workerNodes) {
465b2940 885 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
886 tasks: {
887 executed: 0,
888 executing: 0,
889 queued: 0,
df593701 890 maxQueued: 0,
68cbdc84 891 stolen: 0,
a4e07f72
JB
892 failed: 0
893 },
894 runTime: {
a4e07f72
JB
895 history: expect.any(CircularArray)
896 },
897 waitTime: {
a4e07f72
JB
898 history: expect.any(CircularArray)
899 },
5df69fab
JB
900 elu: {
901 idle: {
5df69fab
JB
902 history: expect.any(CircularArray)
903 },
904 active: {
5df69fab 905 history: expect.any(CircularArray)
f7510105 906 }
5df69fab 907 }
86bf340d 908 })
465b2940
JB
909 expect(workerNode.usage.runTime.history.length).toBe(0)
910 expect(workerNode.usage.waitTime.history.length).toBe(0)
b97d82d8
JB
911 expect(workerNode.usage.elu.idle.history.length).toBe(0)
912 expect(workerNode.usage.elu.active.history.length).toBe(0)
ee11a4a2 913 }
fd7ebd49 914 await pool.destroy()
ee11a4a2
JB
915 })
916
a1763c54
JB
917 it("Verify that pool event emitter 'ready' event can register a callback", async () => {
918 const pool = new DynamicClusterPool(
2431bdb4 919 Math.floor(numberOfWorkers / 2),
164d950a 920 numberOfWorkers,
a1763c54 921 './tests/worker-files/cluster/testWorker.js'
164d950a 922 )
d46660cd 923 let poolInfo
a1763c54
JB
924 let poolReady = 0
925 pool.emitter.on(PoolEvents.ready, (info) => {
926 ++poolReady
d46660cd
JB
927 poolInfo = info
928 })
a1763c54
JB
929 await waitPoolEvents(pool, PoolEvents.ready, 1)
930 expect(poolReady).toBe(1)
d46660cd 931 expect(poolInfo).toStrictEqual({
23ccf9d7 932 version,
d46660cd 933 type: PoolTypes.dynamic,
a1763c54
JB
934 worker: WorkerTypes.cluster,
935 ready: true,
2431bdb4
JB
936 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
937 minSize: expect.any(Number),
938 maxSize: expect.any(Number),
939 workerNodes: expect.any(Number),
940 idleWorkerNodes: expect.any(Number),
941 busyWorkerNodes: expect.any(Number),
942 executedTasks: expect.any(Number),
943 executingTasks: expect.any(Number),
2431bdb4
JB
944 failedTasks: expect.any(Number)
945 })
946 await pool.destroy()
947 })
948
a1763c54
JB
949 it("Verify that pool event emitter 'busy' event can register a callback", async () => {
950 const pool = new FixedThreadPool(
2431bdb4 951 numberOfWorkers,
a1763c54 952 './tests/worker-files/thread/testWorker.js'
2431bdb4 953 )
a1763c54
JB
954 const promises = new Set()
955 let poolBusy = 0
2431bdb4 956 let poolInfo
a1763c54
JB
957 pool.emitter.on(PoolEvents.busy, (info) => {
958 ++poolBusy
2431bdb4
JB
959 poolInfo = info
960 })
a1763c54
JB
961 for (let i = 0; i < numberOfWorkers * 2; i++) {
962 promises.add(pool.execute())
963 }
964 await Promise.all(promises)
965 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
966 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
967 expect(poolBusy).toBe(numberOfWorkers + 1)
2431bdb4
JB
968 expect(poolInfo).toStrictEqual({
969 version,
a1763c54
JB
970 type: PoolTypes.fixed,
971 worker: WorkerTypes.thread,
972 ready: expect.any(Boolean),
2431bdb4 973 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
974 minSize: expect.any(Number),
975 maxSize: expect.any(Number),
976 workerNodes: expect.any(Number),
977 idleWorkerNodes: expect.any(Number),
978 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
979 executedTasks: expect.any(Number),
980 executingTasks: expect.any(Number),
a4e07f72 981 failedTasks: expect.any(Number)
d46660cd 982 })
164d950a
JB
983 await pool.destroy()
984 })
985
a1763c54
JB
986 it("Verify that pool event emitter 'full' event can register a callback", async () => {
987 const pool = new DynamicThreadPool(
988 Math.floor(numberOfWorkers / 2),
7c0ba920
JB
989 numberOfWorkers,
990 './tests/worker-files/thread/testWorker.js'
991 )
09c2d0d3 992 const promises = new Set()
a1763c54 993 let poolFull = 0
d46660cd 994 let poolInfo
a1763c54
JB
995 pool.emitter.on(PoolEvents.full, (info) => {
996 ++poolFull
d46660cd
JB
997 poolInfo = info
998 })
7c0ba920 999 for (let i = 0; i < numberOfWorkers * 2; i++) {
f5d14e90 1000 promises.add(pool.execute())
7c0ba920 1001 }
cf597bc5 1002 await Promise.all(promises)
33e6bb4c 1003 expect(poolFull).toBe(1)
d46660cd 1004 expect(poolInfo).toStrictEqual({
23ccf9d7 1005 version,
a1763c54 1006 type: PoolTypes.dynamic,
d46660cd 1007 worker: WorkerTypes.thread,
2431bdb4
JB
1008 ready: expect.any(Boolean),
1009 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
8735b4e5
JB
1010 minSize: expect.any(Number),
1011 maxSize: expect.any(Number),
1012 workerNodes: expect.any(Number),
1013 idleWorkerNodes: expect.any(Number),
1014 busyWorkerNodes: expect.any(Number),
1015 executedTasks: expect.any(Number),
1016 executingTasks: expect.any(Number),
1017 failedTasks: expect.any(Number)
1018 })
1019 await pool.destroy()
1020 })
1021
3e8611a8 1022 it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
b1aae695 1023 const pool = new FixedThreadPool(
8735b4e5
JB
1024 numberOfWorkers,
1025 './tests/worker-files/thread/testWorker.js',
1026 {
1027 enableTasksQueue: true
1028 }
1029 )
3e8611a8 1030 sinon.stub(pool, 'hasBackPressure').returns(true)
8735b4e5
JB
1031 const promises = new Set()
1032 let poolBackPressure = 0
1033 let poolInfo
1034 pool.emitter.on(PoolEvents.backPressure, (info) => {
1035 ++poolBackPressure
1036 poolInfo = info
1037 })
033f1776 1038 for (let i = 0; i < numberOfWorkers + 1; i++) {
8735b4e5
JB
1039 promises.add(pool.execute())
1040 }
1041 await Promise.all(promises)
033f1776 1042 expect(poolBackPressure).toBe(1)
8735b4e5
JB
1043 expect(poolInfo).toStrictEqual({
1044 version,
3e8611a8 1045 type: PoolTypes.fixed,
8735b4e5
JB
1046 worker: WorkerTypes.thread,
1047 ready: expect.any(Boolean),
1048 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
1049 minSize: expect.any(Number),
1050 maxSize: expect.any(Number),
1051 workerNodes: expect.any(Number),
1052 idleWorkerNodes: expect.any(Number),
1053 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1054 executedTasks: expect.any(Number),
1055 executingTasks: expect.any(Number),
3e8611a8
JB
1056 maxQueuedTasks: expect.any(Number),
1057 queuedTasks: expect.any(Number),
1058 backPressure: true,
68cbdc84 1059 stolenTasks: expect.any(Number),
a4e07f72 1060 failedTasks: expect.any(Number)
d46660cd 1061 })
3e8611a8 1062 expect(pool.hasBackPressure.called).toBe(true)
fd7ebd49 1063 await pool.destroy()
7c0ba920 1064 })
70a4f5ea 1065
90d7d101
JB
1066 it('Verify that listTaskFunctions() is working', async () => {
1067 const dynamicThreadPool = new DynamicThreadPool(
1068 Math.floor(numberOfWorkers / 2),
1069 numberOfWorkers,
1070 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
1071 )
1072 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1073 expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
1074 'default',
1075 'jsonIntegerSerialization',
1076 'factorial',
1077 'fibonacci'
1078 ])
1079 const fixedClusterPool = new FixedClusterPool(
1080 numberOfWorkers,
1081 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1082 )
1083 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1084 expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
1085 'default',
1086 'jsonIntegerSerialization',
1087 'factorial',
1088 'fibonacci'
1089 ])
1090 })
1091
1092 it('Verify that multiple task functions worker is working', async () => {
70a4f5ea 1093 const pool = new DynamicClusterPool(
2431bdb4 1094 Math.floor(numberOfWorkers / 2),
70a4f5ea 1095 numberOfWorkers,
90d7d101 1096 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
70a4f5ea
JB
1097 )
1098 const data = { n: 10 }
82888165 1099 const result0 = await pool.execute(data)
30b963d4 1100 expect(result0).toStrictEqual({ ok: 1 })
70a4f5ea 1101 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
30b963d4 1102 expect(result1).toStrictEqual({ ok: 1 })
70a4f5ea
JB
1103 const result2 = await pool.execute(data, 'factorial')
1104 expect(result2).toBe(3628800)
1105 const result3 = await pool.execute(data, 'fibonacci')
024daf59 1106 expect(result3).toBe(55)
5bb5be17
JB
1107 expect(pool.info.executingTasks).toBe(0)
1108 expect(pool.info.executedTasks).toBe(4)
b414b84c
JB
1109 for (const workerNode of pool.workerNodes) {
1110 expect(workerNode.info.taskFunctions).toStrictEqual([
1111 'default',
1112 'jsonIntegerSerialization',
1113 'factorial',
1114 'fibonacci'
1115 ])
1116 expect(workerNode.taskFunctionsUsage.size).toBe(3)
1117 for (const name of pool.listTaskFunctions()) {
5bb5be17
JB
1118 expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1119 tasks: {
1120 executed: expect.any(Number),
1121 executing: expect.any(Number),
1122 failed: 0,
68cbdc84
JB
1123 queued: 0,
1124 stolen: 0
5bb5be17
JB
1125 },
1126 runTime: {
1127 history: expect.any(CircularArray)
1128 },
1129 waitTime: {
1130 history: expect.any(CircularArray)
1131 },
1132 elu: {
1133 idle: {
1134 history: expect.any(CircularArray)
1135 },
1136 active: {
1137 history: expect.any(CircularArray)
1138 }
1139 }
1140 })
1141 expect(
1142 workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1143 ).toBeGreaterThanOrEqual(0)
1144 }
1145 }
70a4f5ea 1146 })
3ec964d6 1147})