fix: fix pool `execute()` arguments checking
[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 )
49be33fe
JB
260 expect(
261 () =>
262 new FixedThreadPool(
263 numberOfWorkers,
264 './tests/worker-files/thread/testWorker.js',
265 {
266 workerChoiceStrategyOptions: { weights: {} }
267 }
268 )
269 ).toThrowError(
8735b4e5
JB
270 new Error(
271 'Invalid worker choice strategy options: must have a weight for each worker node'
272 )
49be33fe 273 )
f0d7f803
JB
274 expect(
275 () =>
276 new FixedThreadPool(
277 numberOfWorkers,
278 './tests/worker-files/thread/testWorker.js',
279 {
280 workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
281 }
282 )
283 ).toThrowError(
8735b4e5
JB
284 new Error(
285 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
286 )
f0d7f803
JB
287 )
288 expect(
289 () =>
290 new FixedThreadPool(
291 numberOfWorkers,
292 './tests/worker-files/thread/testWorker.js',
293 {
294 enableTasksQueue: true,
295 tasksQueueOptions: { concurrency: 0 }
296 }
297 )
8735b4e5 298 ).toThrowError(
e695d66f 299 new RangeError(
20c6f652 300 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
301 )
302 )
f0d7f803
JB
303 expect(
304 () =>
305 new FixedThreadPool(
306 numberOfWorkers,
307 './tests/worker-files/thread/testWorker.js',
308 {
309 enableTasksQueue: true,
310 tasksQueueOptions: 'invalidTasksQueueOptions'
311 }
312 )
8735b4e5
JB
313 ).toThrowError(
314 new TypeError('Invalid tasks queue options: must be a plain object')
315 )
f0d7f803
JB
316 expect(
317 () =>
318 new FixedThreadPool(
319 numberOfWorkers,
320 './tests/worker-files/thread/testWorker.js',
321 {
322 enableTasksQueue: true,
323 tasksQueueOptions: { concurrency: 0.2 }
324 }
325 )
8735b4e5 326 ).toThrowError(
20c6f652 327 new TypeError('Invalid worker node tasks concurrency: must be an integer')
8735b4e5 328 )
d4aeae5a
JB
329 })
330
2431bdb4 331 it('Verify that pool worker choice strategy options can be set', async () => {
a20f0ba5
JB
332 const pool = new FixedThreadPool(
333 numberOfWorkers,
334 './tests/worker-files/thread/testWorker.js',
335 { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
336 )
337 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d
JB
338 choiceRetries: 6,
339 runTime: { median: false },
340 waitTime: { median: false },
341 elu: { median: false }
342 })
343 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
344 choiceRetries: 6,
932fc8be 345 runTime: { median: false },
5df69fab
JB
346 waitTime: { median: false },
347 elu: { median: false }
a20f0ba5
JB
348 })
349 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
350 .workerChoiceStrategies) {
86bf340d 351 expect(workerChoiceStrategy.opts).toStrictEqual({
8990357d 352 choiceRetries: 6,
932fc8be 353 runTime: { median: false },
5df69fab
JB
354 waitTime: { median: false },
355 elu: { median: false }
86bf340d 356 })
a20f0ba5 357 }
87de9ff5
JB
358 expect(
359 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
360 ).toStrictEqual({
932fc8be
JB
361 runTime: {
362 aggregate: true,
363 average: true,
364 median: false
365 },
366 waitTime: {
367 aggregate: false,
368 average: false,
369 median: false
370 },
5df69fab 371 elu: {
9adcefab
JB
372 aggregate: true,
373 average: true,
5df69fab
JB
374 median: false
375 }
86bf340d 376 })
9adcefab
JB
377 pool.setWorkerChoiceStrategyOptions({
378 runTime: { median: true },
379 elu: { median: true }
380 })
a20f0ba5 381 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d 382 choiceRetries: 6,
9adcefab 383 runTime: { median: true },
8990357d
JB
384 waitTime: { median: false },
385 elu: { median: true }
386 })
387 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
388 choiceRetries: 6,
389 runTime: { median: true },
390 waitTime: { median: false },
9adcefab 391 elu: { median: true }
a20f0ba5
JB
392 })
393 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
394 .workerChoiceStrategies) {
932fc8be 395 expect(workerChoiceStrategy.opts).toStrictEqual({
8990357d 396 choiceRetries: 6,
9adcefab 397 runTime: { median: true },
8990357d 398 waitTime: { median: false },
9adcefab 399 elu: { median: true }
932fc8be 400 })
a20f0ba5 401 }
87de9ff5
JB
402 expect(
403 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
404 ).toStrictEqual({
932fc8be
JB
405 runTime: {
406 aggregate: true,
407 average: false,
408 median: true
409 },
410 waitTime: {
411 aggregate: false,
412 average: false,
413 median: false
414 },
5df69fab 415 elu: {
9adcefab 416 aggregate: true,
5df69fab 417 average: false,
9adcefab 418 median: true
5df69fab 419 }
86bf340d 420 })
9adcefab
JB
421 pool.setWorkerChoiceStrategyOptions({
422 runTime: { median: false },
423 elu: { median: false }
424 })
a20f0ba5 425 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d
JB
426 choiceRetries: 6,
427 runTime: { median: false },
428 waitTime: { median: false },
429 elu: { median: false }
430 })
431 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
432 choiceRetries: 6,
9adcefab 433 runTime: { median: false },
8990357d 434 waitTime: { median: false },
9adcefab 435 elu: { median: false }
a20f0ba5
JB
436 })
437 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
438 .workerChoiceStrategies) {
932fc8be 439 expect(workerChoiceStrategy.opts).toStrictEqual({
8990357d 440 choiceRetries: 6,
9adcefab 441 runTime: { median: false },
8990357d 442 waitTime: { median: false },
9adcefab 443 elu: { median: false }
932fc8be 444 })
a20f0ba5 445 }
87de9ff5
JB
446 expect(
447 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
448 ).toStrictEqual({
932fc8be
JB
449 runTime: {
450 aggregate: true,
451 average: true,
452 median: false
453 },
454 waitTime: {
455 aggregate: false,
456 average: false,
457 median: false
458 },
5df69fab 459 elu: {
9adcefab
JB
460 aggregate: true,
461 average: true,
5df69fab
JB
462 median: false
463 }
86bf340d 464 })
1f95d544
JB
465 expect(() =>
466 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
467 ).toThrowError(
8735b4e5
JB
468 new TypeError(
469 'Invalid worker choice strategy options: must be a plain object'
470 )
1f95d544
JB
471 )
472 expect(() =>
473 pool.setWorkerChoiceStrategyOptions({ weights: {} })
474 ).toThrowError(
8735b4e5
JB
475 new Error(
476 'Invalid worker choice strategy options: must have a weight for each worker node'
477 )
1f95d544
JB
478 )
479 expect(() =>
480 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
481 ).toThrowError(
8735b4e5
JB
482 new Error(
483 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
484 )
1f95d544 485 )
a20f0ba5
JB
486 await pool.destroy()
487 })
488
2431bdb4 489 it('Verify that pool tasks queue can be enabled/disabled', async () => {
a20f0ba5
JB
490 const pool = new FixedThreadPool(
491 numberOfWorkers,
492 './tests/worker-files/thread/testWorker.js'
493 )
494 expect(pool.opts.enableTasksQueue).toBe(false)
495 expect(pool.opts.tasksQueueOptions).toBeUndefined()
496 pool.enableTasksQueue(true)
497 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
498 expect(pool.opts.tasksQueueOptions).toStrictEqual({
499 concurrency: 1,
ff3f866a 500 size: 4
20c6f652 501 })
a20f0ba5
JB
502 pool.enableTasksQueue(true, { concurrency: 2 })
503 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
504 expect(pool.opts.tasksQueueOptions).toStrictEqual({
505 concurrency: 2,
ff3f866a 506 size: 4
20c6f652 507 })
a20f0ba5
JB
508 pool.enableTasksQueue(false)
509 expect(pool.opts.enableTasksQueue).toBe(false)
510 expect(pool.opts.tasksQueueOptions).toBeUndefined()
511 await pool.destroy()
512 })
513
2431bdb4 514 it('Verify that pool tasks queue options can be set', async () => {
a20f0ba5
JB
515 const pool = new FixedThreadPool(
516 numberOfWorkers,
517 './tests/worker-files/thread/testWorker.js',
518 { enableTasksQueue: true }
519 )
20c6f652
JB
520 expect(pool.opts.tasksQueueOptions).toStrictEqual({
521 concurrency: 1,
ff3f866a 522 size: 4
20c6f652 523 })
a20f0ba5 524 pool.setTasksQueueOptions({ concurrency: 2 })
20c6f652
JB
525 expect(pool.opts.tasksQueueOptions).toStrictEqual({
526 concurrency: 2,
ff3f866a 527 size: 4
20c6f652 528 })
f0d7f803
JB
529 expect(() =>
530 pool.setTasksQueueOptions('invalidTasksQueueOptions')
8735b4e5
JB
531 ).toThrowError(
532 new TypeError('Invalid tasks queue options: must be a plain object')
533 )
a20f0ba5 534 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
e695d66f 535 new RangeError(
20c6f652 536 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
537 )
538 )
539 expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError(
e695d66f 540 new RangeError(
20c6f652 541 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
8735b4e5 542 )
a20f0ba5 543 )
f0d7f803 544 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
20c6f652
JB
545 new TypeError('Invalid worker node tasks concurrency: must be an integer')
546 )
68dbcdc0
JB
547 expect(() => pool.setTasksQueueOptions({ queueMaxSize: 2 })).toThrowError(
548 new Error(
549 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
550 )
551 )
ff3f866a 552 expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
20c6f652 553 new RangeError(
68dbcdc0 554 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
20c6f652
JB
555 )
556 )
ff3f866a 557 expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError(
20c6f652 558 new RangeError(
68dbcdc0 559 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
20c6f652
JB
560 )
561 )
ff3f866a 562 expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError(
68dbcdc0 563 new TypeError('Invalid worker node tasks queue size: must be an integer')
f0d7f803 564 )
a20f0ba5
JB
565 await pool.destroy()
566 })
567
6b27d407
JB
568 it('Verify that pool info is set', async () => {
569 let pool = new FixedThreadPool(
570 numberOfWorkers,
571 './tests/worker-files/thread/testWorker.js'
572 )
2dca6cab
JB
573 expect(pool.info).toStrictEqual({
574 version,
575 type: PoolTypes.fixed,
576 worker: WorkerTypes.thread,
577 ready: true,
578 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
579 minSize: numberOfWorkers,
580 maxSize: numberOfWorkers,
581 workerNodes: numberOfWorkers,
582 idleWorkerNodes: numberOfWorkers,
583 busyWorkerNodes: 0,
584 executedTasks: 0,
585 executingTasks: 0,
2dca6cab
JB
586 failedTasks: 0
587 })
6b27d407
JB
588 await pool.destroy()
589 pool = new DynamicClusterPool(
2431bdb4 590 Math.floor(numberOfWorkers / 2),
6b27d407 591 numberOfWorkers,
ecdfbdc0 592 './tests/worker-files/cluster/testWorker.js'
6b27d407 593 )
2dca6cab
JB
594 expect(pool.info).toStrictEqual({
595 version,
596 type: PoolTypes.dynamic,
597 worker: WorkerTypes.cluster,
598 ready: true,
599 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
600 minSize: Math.floor(numberOfWorkers / 2),
601 maxSize: numberOfWorkers,
602 workerNodes: Math.floor(numberOfWorkers / 2),
603 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
604 busyWorkerNodes: 0,
605 executedTasks: 0,
606 executingTasks: 0,
2dca6cab
JB
607 failedTasks: 0
608 })
6b27d407
JB
609 await pool.destroy()
610 })
611
2431bdb4 612 it('Verify that pool worker tasks usage are initialized', async () => {
bf9549ae
JB
613 const pool = new FixedClusterPool(
614 numberOfWorkers,
615 './tests/worker-files/cluster/testWorker.js'
616 )
f06e48d8 617 for (const workerNode of pool.workerNodes) {
465b2940 618 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
619 tasks: {
620 executed: 0,
621 executing: 0,
622 queued: 0,
df593701 623 maxQueued: 0,
68cbdc84 624 stolen: 0,
a4e07f72
JB
625 failed: 0
626 },
627 runTime: {
a4e07f72
JB
628 history: expect.any(CircularArray)
629 },
630 waitTime: {
a4e07f72
JB
631 history: expect.any(CircularArray)
632 },
5df69fab
JB
633 elu: {
634 idle: {
5df69fab
JB
635 history: expect.any(CircularArray)
636 },
637 active: {
5df69fab 638 history: expect.any(CircularArray)
f7510105 639 }
5df69fab 640 }
86bf340d 641 })
f06e48d8
JB
642 }
643 await pool.destroy()
644 })
645
2431bdb4
JB
646 it('Verify that pool worker tasks queue are initialized', async () => {
647 let pool = new FixedClusterPool(
f06e48d8
JB
648 numberOfWorkers,
649 './tests/worker-files/cluster/testWorker.js'
650 )
651 for (const workerNode of pool.workerNodes) {
652 expect(workerNode.tasksQueue).toBeDefined()
574b351d 653 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
4d8bf9e4 654 expect(workerNode.tasksQueue.size).toBe(0)
9c16fb4b 655 expect(workerNode.tasksQueue.maxSize).toBe(0)
bf9549ae 656 }
fd7ebd49 657 await pool.destroy()
2431bdb4
JB
658 pool = new DynamicThreadPool(
659 Math.floor(numberOfWorkers / 2),
660 numberOfWorkers,
661 './tests/worker-files/thread/testWorker.js'
662 )
663 for (const workerNode of pool.workerNodes) {
664 expect(workerNode.tasksQueue).toBeDefined()
574b351d 665 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
2431bdb4
JB
666 expect(workerNode.tasksQueue.size).toBe(0)
667 expect(workerNode.tasksQueue.maxSize).toBe(0)
668 }
669 })
670
671 it('Verify that pool worker info are initialized', async () => {
672 let pool = new FixedClusterPool(
673 numberOfWorkers,
674 './tests/worker-files/cluster/testWorker.js'
675 )
2dca6cab
JB
676 for (const workerNode of pool.workerNodes) {
677 expect(workerNode.info).toStrictEqual({
678 id: expect.any(Number),
679 type: WorkerTypes.cluster,
680 dynamic: false,
681 ready: true
682 })
683 }
2431bdb4
JB
684 await pool.destroy()
685 pool = new DynamicThreadPool(
686 Math.floor(numberOfWorkers / 2),
687 numberOfWorkers,
688 './tests/worker-files/thread/testWorker.js'
689 )
2dca6cab
JB
690 for (const workerNode of pool.workerNodes) {
691 expect(workerNode.info).toStrictEqual({
692 id: expect.any(Number),
693 type: WorkerTypes.thread,
694 dynamic: false,
7884d183 695 ready: true
2dca6cab
JB
696 })
697 }
bf9549ae
JB
698 })
699
9d2d0da1
JB
700 it('Verify that pool execute() arguments are checked', async () => {
701 const pool = new FixedClusterPool(
702 numberOfWorkers,
703 './tests/worker-files/cluster/testWorker.js'
704 )
705 await expect(pool.execute(undefined, 0)).rejects.toThrowError(
706 new TypeError('name argument must be a string')
707 )
708 await expect(pool.execute(undefined, '')).rejects.toThrowError(
709 new TypeError('name argument must not be an empty string')
710 )
711 await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
712 new TypeError('transferList argument must be an array')
713 )
714 await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
715 "Task function 'unknown' not found"
716 )
717 await pool.destroy()
718 await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
719 new Error('Cannot execute a task on destroyed pool')
720 )
721 })
722
2431bdb4 723 it('Verify that pool worker tasks usage are computed', async () => {
bf9549ae
JB
724 const pool = new FixedClusterPool(
725 numberOfWorkers,
726 './tests/worker-files/cluster/testWorker.js'
727 )
09c2d0d3 728 const promises = new Set()
fc027381
JB
729 const maxMultiplier = 2
730 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 731 promises.add(pool.execute())
bf9549ae 732 }
f06e48d8 733 for (const workerNode of pool.workerNodes) {
465b2940 734 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
735 tasks: {
736 executed: 0,
737 executing: maxMultiplier,
738 queued: 0,
df593701 739 maxQueued: 0,
68cbdc84 740 stolen: 0,
a4e07f72
JB
741 failed: 0
742 },
743 runTime: {
a4e07f72
JB
744 history: expect.any(CircularArray)
745 },
746 waitTime: {
a4e07f72
JB
747 history: expect.any(CircularArray)
748 },
5df69fab
JB
749 elu: {
750 idle: {
5df69fab
JB
751 history: expect.any(CircularArray)
752 },
753 active: {
5df69fab 754 history: expect.any(CircularArray)
f7510105 755 }
5df69fab 756 }
86bf340d 757 })
bf9549ae
JB
758 }
759 await Promise.all(promises)
f06e48d8 760 for (const workerNode of pool.workerNodes) {
465b2940 761 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
762 tasks: {
763 executed: maxMultiplier,
764 executing: 0,
765 queued: 0,
df593701 766 maxQueued: 0,
68cbdc84 767 stolen: 0,
a4e07f72
JB
768 failed: 0
769 },
770 runTime: {
a4e07f72
JB
771 history: expect.any(CircularArray)
772 },
773 waitTime: {
a4e07f72
JB
774 history: expect.any(CircularArray)
775 },
5df69fab
JB
776 elu: {
777 idle: {
5df69fab
JB
778 history: expect.any(CircularArray)
779 },
780 active: {
5df69fab 781 history: expect.any(CircularArray)
f7510105 782 }
5df69fab 783 }
86bf340d 784 })
bf9549ae 785 }
fd7ebd49 786 await pool.destroy()
bf9549ae
JB
787 })
788
2431bdb4 789 it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
7fd82a1c 790 const pool = new DynamicThreadPool(
2431bdb4 791 Math.floor(numberOfWorkers / 2),
8f4878b7 792 numberOfWorkers,
9e619829
JB
793 './tests/worker-files/thread/testWorker.js'
794 )
09c2d0d3 795 const promises = new Set()
ee9f5295
JB
796 const maxMultiplier = 2
797 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 798 promises.add(pool.execute())
9e619829
JB
799 }
800 await Promise.all(promises)
f06e48d8 801 for (const workerNode of pool.workerNodes) {
465b2940 802 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
803 tasks: {
804 executed: expect.any(Number),
805 executing: 0,
806 queued: 0,
df593701 807 maxQueued: 0,
68cbdc84 808 stolen: 0,
a4e07f72
JB
809 failed: 0
810 },
811 runTime: {
a4e07f72
JB
812 history: expect.any(CircularArray)
813 },
814 waitTime: {
a4e07f72
JB
815 history: expect.any(CircularArray)
816 },
5df69fab
JB
817 elu: {
818 idle: {
5df69fab
JB
819 history: expect.any(CircularArray)
820 },
821 active: {
5df69fab 822 history: expect.any(CircularArray)
f7510105 823 }
5df69fab 824 }
86bf340d 825 })
465b2940 826 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
94407def
JB
827 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
828 numberOfWorkers * maxMultiplier
829 )
b97d82d8
JB
830 expect(workerNode.usage.runTime.history.length).toBe(0)
831 expect(workerNode.usage.waitTime.history.length).toBe(0)
832 expect(workerNode.usage.elu.idle.history.length).toBe(0)
833 expect(workerNode.usage.elu.active.history.length).toBe(0)
9e619829
JB
834 }
835 pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
f06e48d8 836 for (const workerNode of pool.workerNodes) {
465b2940 837 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
838 tasks: {
839 executed: 0,
840 executing: 0,
841 queued: 0,
df593701 842 maxQueued: 0,
68cbdc84 843 stolen: 0,
a4e07f72
JB
844 failed: 0
845 },
846 runTime: {
a4e07f72
JB
847 history: expect.any(CircularArray)
848 },
849 waitTime: {
a4e07f72
JB
850 history: expect.any(CircularArray)
851 },
5df69fab
JB
852 elu: {
853 idle: {
5df69fab
JB
854 history: expect.any(CircularArray)
855 },
856 active: {
5df69fab 857 history: expect.any(CircularArray)
f7510105 858 }
5df69fab 859 }
86bf340d 860 })
465b2940
JB
861 expect(workerNode.usage.runTime.history.length).toBe(0)
862 expect(workerNode.usage.waitTime.history.length).toBe(0)
b97d82d8
JB
863 expect(workerNode.usage.elu.idle.history.length).toBe(0)
864 expect(workerNode.usage.elu.active.history.length).toBe(0)
ee11a4a2 865 }
fd7ebd49 866 await pool.destroy()
ee11a4a2
JB
867 })
868
a1763c54
JB
869 it("Verify that pool event emitter 'ready' event can register a callback", async () => {
870 const pool = new DynamicClusterPool(
2431bdb4 871 Math.floor(numberOfWorkers / 2),
164d950a 872 numberOfWorkers,
a1763c54 873 './tests/worker-files/cluster/testWorker.js'
164d950a 874 )
d46660cd 875 let poolInfo
a1763c54
JB
876 let poolReady = 0
877 pool.emitter.on(PoolEvents.ready, (info) => {
878 ++poolReady
d46660cd
JB
879 poolInfo = info
880 })
a1763c54
JB
881 await waitPoolEvents(pool, PoolEvents.ready, 1)
882 expect(poolReady).toBe(1)
d46660cd 883 expect(poolInfo).toStrictEqual({
23ccf9d7 884 version,
d46660cd 885 type: PoolTypes.dynamic,
a1763c54
JB
886 worker: WorkerTypes.cluster,
887 ready: true,
2431bdb4
JB
888 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
889 minSize: expect.any(Number),
890 maxSize: expect.any(Number),
891 workerNodes: expect.any(Number),
892 idleWorkerNodes: expect.any(Number),
893 busyWorkerNodes: expect.any(Number),
894 executedTasks: expect.any(Number),
895 executingTasks: expect.any(Number),
2431bdb4
JB
896 failedTasks: expect.any(Number)
897 })
898 await pool.destroy()
899 })
900
a1763c54
JB
901 it("Verify that pool event emitter 'busy' event can register a callback", async () => {
902 const pool = new FixedThreadPool(
2431bdb4 903 numberOfWorkers,
a1763c54 904 './tests/worker-files/thread/testWorker.js'
2431bdb4 905 )
a1763c54
JB
906 const promises = new Set()
907 let poolBusy = 0
2431bdb4 908 let poolInfo
a1763c54
JB
909 pool.emitter.on(PoolEvents.busy, (info) => {
910 ++poolBusy
2431bdb4
JB
911 poolInfo = info
912 })
a1763c54
JB
913 for (let i = 0; i < numberOfWorkers * 2; i++) {
914 promises.add(pool.execute())
915 }
916 await Promise.all(promises)
917 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
918 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
919 expect(poolBusy).toBe(numberOfWorkers + 1)
2431bdb4
JB
920 expect(poolInfo).toStrictEqual({
921 version,
a1763c54
JB
922 type: PoolTypes.fixed,
923 worker: WorkerTypes.thread,
924 ready: expect.any(Boolean),
2431bdb4 925 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
926 minSize: expect.any(Number),
927 maxSize: expect.any(Number),
928 workerNodes: expect.any(Number),
929 idleWorkerNodes: expect.any(Number),
930 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
931 executedTasks: expect.any(Number),
932 executingTasks: expect.any(Number),
a4e07f72 933 failedTasks: expect.any(Number)
d46660cd 934 })
164d950a
JB
935 await pool.destroy()
936 })
937
a1763c54
JB
938 it("Verify that pool event emitter 'full' event can register a callback", async () => {
939 const pool = new DynamicThreadPool(
940 Math.floor(numberOfWorkers / 2),
7c0ba920
JB
941 numberOfWorkers,
942 './tests/worker-files/thread/testWorker.js'
943 )
09c2d0d3 944 const promises = new Set()
a1763c54 945 let poolFull = 0
d46660cd 946 let poolInfo
a1763c54
JB
947 pool.emitter.on(PoolEvents.full, (info) => {
948 ++poolFull
d46660cd
JB
949 poolInfo = info
950 })
7c0ba920 951 for (let i = 0; i < numberOfWorkers * 2; i++) {
f5d14e90 952 promises.add(pool.execute())
7c0ba920 953 }
cf597bc5 954 await Promise.all(promises)
33e6bb4c 955 expect(poolFull).toBe(1)
d46660cd 956 expect(poolInfo).toStrictEqual({
23ccf9d7 957 version,
a1763c54 958 type: PoolTypes.dynamic,
d46660cd 959 worker: WorkerTypes.thread,
2431bdb4
JB
960 ready: expect.any(Boolean),
961 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
8735b4e5
JB
962 minSize: expect.any(Number),
963 maxSize: expect.any(Number),
964 workerNodes: expect.any(Number),
965 idleWorkerNodes: expect.any(Number),
966 busyWorkerNodes: expect.any(Number),
967 executedTasks: expect.any(Number),
968 executingTasks: expect.any(Number),
969 failedTasks: expect.any(Number)
970 })
971 await pool.destroy()
972 })
973
3e8611a8 974 it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
b1aae695 975 const pool = new FixedThreadPool(
8735b4e5
JB
976 numberOfWorkers,
977 './tests/worker-files/thread/testWorker.js',
978 {
979 enableTasksQueue: true
980 }
981 )
3e8611a8 982 sinon.stub(pool, 'hasBackPressure').returns(true)
8735b4e5
JB
983 const promises = new Set()
984 let poolBackPressure = 0
985 let poolInfo
986 pool.emitter.on(PoolEvents.backPressure, (info) => {
987 ++poolBackPressure
988 poolInfo = info
989 })
033f1776 990 for (let i = 0; i < numberOfWorkers + 1; i++) {
8735b4e5
JB
991 promises.add(pool.execute())
992 }
993 await Promise.all(promises)
033f1776 994 expect(poolBackPressure).toBe(1)
8735b4e5
JB
995 expect(poolInfo).toStrictEqual({
996 version,
3e8611a8 997 type: PoolTypes.fixed,
8735b4e5
JB
998 worker: WorkerTypes.thread,
999 ready: expect.any(Boolean),
1000 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
1001 minSize: expect.any(Number),
1002 maxSize: expect.any(Number),
1003 workerNodes: expect.any(Number),
1004 idleWorkerNodes: expect.any(Number),
1005 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1006 executedTasks: expect.any(Number),
1007 executingTasks: expect.any(Number),
3e8611a8
JB
1008 maxQueuedTasks: expect.any(Number),
1009 queuedTasks: expect.any(Number),
1010 backPressure: true,
68cbdc84 1011 stolenTasks: expect.any(Number),
a4e07f72 1012 failedTasks: expect.any(Number)
d46660cd 1013 })
3e8611a8 1014 expect(pool.hasBackPressure.called).toBe(true)
fd7ebd49 1015 await pool.destroy()
7c0ba920 1016 })
70a4f5ea 1017
90d7d101
JB
1018 it('Verify that listTaskFunctions() is working', async () => {
1019 const dynamicThreadPool = new DynamicThreadPool(
1020 Math.floor(numberOfWorkers / 2),
1021 numberOfWorkers,
1022 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
1023 )
1024 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1025 expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
1026 'default',
1027 'jsonIntegerSerialization',
1028 'factorial',
1029 'fibonacci'
1030 ])
1031 const fixedClusterPool = new FixedClusterPool(
1032 numberOfWorkers,
1033 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1034 )
1035 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1036 expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
1037 'default',
1038 'jsonIntegerSerialization',
1039 'factorial',
1040 'fibonacci'
1041 ])
1042 })
1043
1044 it('Verify that multiple task functions worker is working', async () => {
70a4f5ea 1045 const pool = new DynamicClusterPool(
2431bdb4 1046 Math.floor(numberOfWorkers / 2),
70a4f5ea 1047 numberOfWorkers,
90d7d101 1048 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
70a4f5ea
JB
1049 )
1050 const data = { n: 10 }
82888165 1051 const result0 = await pool.execute(data)
30b963d4 1052 expect(result0).toStrictEqual({ ok: 1 })
70a4f5ea 1053 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
30b963d4 1054 expect(result1).toStrictEqual({ ok: 1 })
70a4f5ea
JB
1055 const result2 = await pool.execute(data, 'factorial')
1056 expect(result2).toBe(3628800)
1057 const result3 = await pool.execute(data, 'fibonacci')
024daf59 1058 expect(result3).toBe(55)
5bb5be17
JB
1059 expect(pool.info.executingTasks).toBe(0)
1060 expect(pool.info.executedTasks).toBe(4)
b414b84c
JB
1061 for (const workerNode of pool.workerNodes) {
1062 expect(workerNode.info.taskFunctions).toStrictEqual([
1063 'default',
1064 'jsonIntegerSerialization',
1065 'factorial',
1066 'fibonacci'
1067 ])
1068 expect(workerNode.taskFunctionsUsage.size).toBe(3)
1069 for (const name of pool.listTaskFunctions()) {
5bb5be17
JB
1070 expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1071 tasks: {
1072 executed: expect.any(Number),
1073 executing: expect.any(Number),
1074 failed: 0,
68cbdc84
JB
1075 queued: 0,
1076 stolen: 0
5bb5be17
JB
1077 },
1078 runTime: {
1079 history: expect.any(CircularArray)
1080 },
1081 waitTime: {
1082 history: expect.any(CircularArray)
1083 },
1084 elu: {
1085 idle: {
1086 history: expect.any(CircularArray)
1087 },
1088 active: {
1089 history: expect.any(CircularArray)
1090 }
1091 }
1092 })
1093 expect(
1094 workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1095 ).toBeGreaterThanOrEqual(0)
1096 }
1097 }
70a4f5ea 1098 })
3ec964d6 1099})