test: less strict expectation
[poolifier.git] / tests / pools / abstract-pool.test.mjs
CommitLineData
a074ffee 1import { EventEmitterAsyncResource } from 'node:events'
2eb14889 2import { dirname, join } from 'node:path'
a074ffee 3import { readFileSync } from 'node:fs'
2eb14889 4import { fileURLToPath } from 'node:url'
f18fd12b 5import { createHook, executionAsyncId } from 'node:async_hooks'
a074ffee
JB
6import { expect } from 'expect'
7import { restore, stub } from 'sinon'
8import {
70a4f5ea 9 DynamicClusterPool,
9e619829 10 DynamicThreadPool,
aee46736 11 FixedClusterPool,
e843b904 12 FixedThreadPool,
aee46736 13 PoolEvents,
184855e6 14 PoolTypes,
3d6dd312 15 WorkerChoiceStrategies,
184855e6 16 WorkerTypes
a074ffee
JB
17} from '../../lib/index.js'
18import { CircularArray } from '../../lib/circular-array.js'
19import { Deque } from '../../lib/deque.js'
20import { DEFAULT_TASK_NAME } from '../../lib/utils.js'
21import { waitPoolEvents } from '../test-utils.js'
22import { WorkerNode } from '../../lib/pools/worker-node.js'
e1ffb94f
JB
23
24describe('Abstract pool test suite', () => {
2eb14889
JB
25 const version = JSON.parse(
26 readFileSync(
a5e5599c 27 join(dirname(fileURLToPath(import.meta.url)), '../..', 'package.json'),
2eb14889
JB
28 'utf8'
29 )
30 ).version
fc027381 31 const numberOfWorkers = 2
a8884ffd 32 class StubPoolWithIsMain extends FixedThreadPool {
e1ffb94f
JB
33 isMain () {
34 return false
35 }
3ec964d6 36 }
3ec964d6 37
dc021bcc 38 afterEach(() => {
a074ffee 39 restore()
dc021bcc
JB
40 })
41
bcf85f7f
JB
42 it('Verify that pool can be created and destroyed', async () => {
43 const pool = new FixedThreadPool(
44 numberOfWorkers,
45 './tests/worker-files/thread/testWorker.mjs'
46 )
47 expect(pool).toBeInstanceOf(FixedThreadPool)
48 await pool.destroy()
49 })
50
51 it('Verify that pool cannot be created from a non main thread/process', () => {
8d3782fa
JB
52 expect(
53 () =>
a8884ffd 54 new StubPoolWithIsMain(
7c0ba920 55 numberOfWorkers,
b2fd3f4a 56 './tests/worker-files/thread/testWorker.mjs',
8d3782fa 57 {
041dc05b 58 errorHandler: e => console.error(e)
8d3782fa
JB
59 }
60 )
948faff7 61 ).toThrow(
e695d66f
JB
62 new Error(
63 'Cannot start a pool from a worker with the same type as the pool'
64 )
04f45163 65 )
3ec964d6 66 })
c510fea7 67
bc61cfe6
JB
68 it('Verify that pool statuses properties are set', async () => {
69 const pool = new FixedThreadPool(
70 numberOfWorkers,
b2fd3f4a 71 './tests/worker-files/thread/testWorker.mjs'
bc61cfe6 72 )
bc61cfe6 73 expect(pool.started).toBe(true)
711623b8
JB
74 expect(pool.starting).toBe(false)
75 expect(pool.destroying).toBe(false)
bc61cfe6 76 await pool.destroy()
bc61cfe6
JB
77 })
78
c510fea7 79 it('Verify that filePath is checked', () => {
948faff7 80 expect(() => new FixedThreadPool(numberOfWorkers)).toThrow(
c3719753
JB
81 new TypeError('The worker file path must be specified')
82 )
83 expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrow(
84 new TypeError('The worker file path must be a string')
3d6dd312
JB
85 )
86 expect(
87 () => new FixedThreadPool(numberOfWorkers, './dummyWorker.ts')
948faff7 88 ).toThrow(new Error("Cannot find the worker file './dummyWorker.ts'"))
8d3782fa
JB
89 })
90
91 it('Verify that numberOfWorkers is checked', () => {
8003c026
JB
92 expect(
93 () =>
94 new FixedThreadPool(
95 undefined,
b2fd3f4a 96 './tests/worker-files/thread/testWorker.mjs'
8003c026 97 )
948faff7 98 ).toThrow(
e695d66f
JB
99 new Error(
100 'Cannot instantiate a pool without specifying the number of workers'
101 )
8d3782fa
JB
102 )
103 })
104
105 it('Verify that a negative number of workers is checked', () => {
106 expect(
107 () =>
108 new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
948faff7 109 ).toThrow(
473c717a
JB
110 new RangeError(
111 'Cannot instantiate a pool with a negative number of workers'
112 )
8d3782fa
JB
113 )
114 })
115
116 it('Verify that a non integer number of workers is checked', () => {
117 expect(
118 () =>
b2fd3f4a 119 new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.mjs')
948faff7 120 ).toThrow(
473c717a 121 new TypeError(
0d80593b 122 'Cannot instantiate a pool with a non safe integer number of workers'
8d3782fa
JB
123 )
124 )
c510fea7 125 })
7c0ba920 126
26ce26ca
JB
127 it('Verify that pool arguments number and pool type are checked', () => {
128 expect(
129 () =>
130 new FixedThreadPool(
131 numberOfWorkers,
132 './tests/worker-files/thread/testWorker.mjs',
133 undefined,
134 numberOfWorkers * 2
135 )
136 ).toThrow(
137 new Error(
138 'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
139 )
140 )
141 })
142
216541b6 143 it('Verify that dynamic pool sizing is checked', () => {
a5ed75b7
JB
144 expect(
145 () =>
146 new DynamicClusterPool(
147 1,
148 undefined,
149 './tests/worker-files/cluster/testWorker.js'
150 )
948faff7 151 ).toThrow(
a5ed75b7
JB
152 new TypeError(
153 'Cannot instantiate a dynamic pool without specifying the maximum pool size'
154 )
155 )
2761efb4
JB
156 expect(
157 () =>
158 new DynamicThreadPool(
159 0.5,
160 1,
b2fd3f4a 161 './tests/worker-files/thread/testWorker.mjs'
2761efb4 162 )
948faff7 163 ).toThrow(
2761efb4
JB
164 new TypeError(
165 'Cannot instantiate a pool with a non safe integer number of workers'
166 )
167 )
168 expect(
169 () =>
170 new DynamicClusterPool(
171 0,
172 0.5,
a5ed75b7 173 './tests/worker-files/cluster/testWorker.js'
2761efb4 174 )
948faff7 175 ).toThrow(
2761efb4
JB
176 new TypeError(
177 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
178 )
179 )
2431bdb4
JB
180 expect(
181 () =>
b2fd3f4a
JB
182 new DynamicThreadPool(
183 2,
184 1,
185 './tests/worker-files/thread/testWorker.mjs'
186 )
948faff7 187 ).toThrow(
2431bdb4
JB
188 new RangeError(
189 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
190 )
191 )
192 expect(
193 () =>
b2fd3f4a
JB
194 new DynamicThreadPool(
195 0,
196 0,
197 './tests/worker-files/thread/testWorker.mjs'
198 )
948faff7 199 ).toThrow(
2431bdb4 200 new RangeError(
213cbac6 201 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
2431bdb4
JB
202 )
203 )
21f710aa
JB
204 expect(
205 () =>
213cbac6
JB
206 new DynamicClusterPool(
207 1,
208 1,
209 './tests/worker-files/cluster/testWorker.js'
210 )
948faff7 211 ).toThrow(
21f710aa 212 new RangeError(
213cbac6 213 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
21f710aa
JB
214 )
215 )
2431bdb4
JB
216 })
217
fd7ebd49 218 it('Verify that pool options are checked', async () => {
7c0ba920
JB
219 let pool = new FixedThreadPool(
220 numberOfWorkers,
b2fd3f4a 221 './tests/worker-files/thread/testWorker.mjs'
7c0ba920 222 )
b5604034 223 expect(pool.emitter).toBeInstanceOf(EventEmitterAsyncResource)
47352846
JB
224 expect(pool.opts).toStrictEqual({
225 startWorkers: true,
226 enableEvents: true,
227 restartWorkerOnError: true,
228 enableTasksQueue: false,
26ce26ca 229 workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN
8990357d
JB
230 })
231 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
232 retries:
233 pool.info.maxSize +
234 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
932fc8be 235 runTime: { median: false },
5df69fab 236 waitTime: { median: false },
00e1bdeb
JB
237 elu: { median: false },
238 weights: expect.objectContaining({
239 0: expect.any(Number),
240 [pool.info.maxSize - 1]: expect.any(Number)
241 })
da309861 242 })
999ef664
JB
243 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
244 .workerChoiceStrategies) {
fde30586
JB
245 expect(workerChoiceStrategy.opts).toStrictEqual(
246 expect.objectContaining({
247 retries:
248 pool.info.maxSize +
249 Object.keys(workerChoiceStrategy.opts.weights).length,
250 runTime: { median: false },
251 waitTime: { median: false },
252 elu: { median: false }
253 // weights: expect.objectContaining({
254 // 0: expect.any(Number),
255 // [pool.info.maxSize - 1]: expect.any(Number)
256 // })
00e1bdeb 257 })
fde30586 258 )
999ef664 259 }
fd7ebd49 260 await pool.destroy()
73bfd59d 261 const testHandler = () => console.info('test handler executed')
7c0ba920
JB
262 pool = new FixedThreadPool(
263 numberOfWorkers,
b2fd3f4a 264 './tests/worker-files/thread/testWorker.mjs',
7c0ba920 265 {
e4543b14 266 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
49be33fe 267 workerChoiceStrategyOptions: {
932fc8be 268 runTime: { median: true },
fc027381 269 weights: { 0: 300, 1: 200 }
49be33fe 270 },
35cf1c03 271 enableEvents: false,
1f68cede 272 restartWorkerOnError: false,
ff733df7 273 enableTasksQueue: true,
d4aeae5a 274 tasksQueueOptions: { concurrency: 2 },
35cf1c03
JB
275 messageHandler: testHandler,
276 errorHandler: testHandler,
277 onlineHandler: testHandler,
278 exitHandler: testHandler
7c0ba920
JB
279 }
280 )
7c0ba920 281 expect(pool.emitter).toBeUndefined()
47352846
JB
282 expect(pool.opts).toStrictEqual({
283 startWorkers: true,
284 enableEvents: false,
285 restartWorkerOnError: false,
286 enableTasksQueue: true,
287 tasksQueueOptions: {
288 concurrency: 2,
2324f8c9 289 size: Math.pow(numberOfWorkers, 2),
dbd73092 290 taskStealing: true,
32b141fd 291 tasksStealingOnBackPressure: true,
568d0075 292 tasksFinishedTimeout: 2000
47352846
JB
293 },
294 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
295 workerChoiceStrategyOptions: {
47352846 296 runTime: { median: true },
47352846
JB
297 weights: { 0: 300, 1: 200 }
298 },
299 onlineHandler: testHandler,
300 messageHandler: testHandler,
301 errorHandler: testHandler,
302 exitHandler: testHandler
8990357d
JB
303 })
304 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
449cd154
JB
305 retries:
306 pool.info.maxSize +
307 Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length,
8990357d
JB
308 runTime: { median: true },
309 waitTime: { median: false },
310 elu: { median: false },
fc027381 311 weights: { 0: 300, 1: 200 }
da309861 312 })
999ef664
JB
313 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
314 .workerChoiceStrategies) {
315 expect(workerChoiceStrategy.opts).toStrictEqual({
449cd154
JB
316 retries:
317 pool.info.maxSize +
318 Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length,
999ef664
JB
319 runTime: { median: true },
320 waitTime: { median: false },
321 elu: { median: false },
322 weights: { 0: 300, 1: 200 }
323 })
324 }
fd7ebd49 325 await pool.destroy()
7c0ba920
JB
326 })
327
fe291b64 328 it('Verify that pool options are validated', () => {
d4aeae5a
JB
329 expect(
330 () =>
331 new FixedThreadPool(
332 numberOfWorkers,
b2fd3f4a 333 './tests/worker-files/thread/testWorker.mjs',
d4aeae5a 334 {
f0d7f803 335 workerChoiceStrategy: 'invalidStrategy'
d4aeae5a
JB
336 }
337 )
948faff7 338 ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'"))
49be33fe
JB
339 expect(
340 () =>
341 new FixedThreadPool(
342 numberOfWorkers,
b2fd3f4a 343 './tests/worker-files/thread/testWorker.mjs',
49be33fe
JB
344 {
345 workerChoiceStrategyOptions: { weights: {} }
346 }
347 )
948faff7 348 ).toThrow(
8735b4e5
JB
349 new Error(
350 'Invalid worker choice strategy options: must have a weight for each worker node'
351 )
49be33fe 352 )
f0d7f803
JB
353 expect(
354 () =>
355 new FixedThreadPool(
356 numberOfWorkers,
b2fd3f4a 357 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
358 {
359 workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
360 }
361 )
948faff7 362 ).toThrow(
8735b4e5
JB
363 new Error(
364 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
365 )
f0d7f803 366 )
5c4c2dee
JB
367 expect(
368 () =>
369 new FixedThreadPool(
370 numberOfWorkers,
b2fd3f4a 371 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
372 {
373 enableTasksQueue: true,
374 tasksQueueOptions: 'invalidTasksQueueOptions'
375 }
376 )
948faff7 377 ).toThrow(
5c4c2dee
JB
378 new TypeError('Invalid tasks queue options: must be a plain object')
379 )
f0d7f803
JB
380 expect(
381 () =>
382 new FixedThreadPool(
383 numberOfWorkers,
b2fd3f4a 384 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
385 {
386 enableTasksQueue: true,
387 tasksQueueOptions: { concurrency: 0 }
388 }
389 )
948faff7 390 ).toThrow(
e695d66f 391 new RangeError(
20c6f652 392 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
393 )
394 )
f0d7f803
JB
395 expect(
396 () =>
397 new FixedThreadPool(
398 numberOfWorkers,
b2fd3f4a 399 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
400 {
401 enableTasksQueue: true,
5c4c2dee 402 tasksQueueOptions: { concurrency: -1 }
f0d7f803
JB
403 }
404 )
948faff7 405 ).toThrow(
5c4c2dee
JB
406 new RangeError(
407 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
408 )
8735b4e5 409 )
f0d7f803
JB
410 expect(
411 () =>
412 new FixedThreadPool(
413 numberOfWorkers,
b2fd3f4a 414 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
415 {
416 enableTasksQueue: true,
417 tasksQueueOptions: { concurrency: 0.2 }
418 }
419 )
948faff7 420 ).toThrow(
20c6f652 421 new TypeError('Invalid worker node tasks concurrency: must be an integer')
8735b4e5 422 )
5c4c2dee
JB
423 expect(
424 () =>
425 new FixedThreadPool(
426 numberOfWorkers,
b2fd3f4a 427 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
428 {
429 enableTasksQueue: true,
430 tasksQueueOptions: { size: 0 }
431 }
432 )
948faff7 433 ).toThrow(
5c4c2dee
JB
434 new RangeError(
435 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
436 )
437 )
438 expect(
439 () =>
440 new FixedThreadPool(
441 numberOfWorkers,
b2fd3f4a 442 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
443 {
444 enableTasksQueue: true,
445 tasksQueueOptions: { size: -1 }
446 }
447 )
948faff7 448 ).toThrow(
5c4c2dee
JB
449 new RangeError(
450 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
451 )
452 )
453 expect(
454 () =>
455 new FixedThreadPool(
456 numberOfWorkers,
b2fd3f4a 457 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
458 {
459 enableTasksQueue: true,
460 tasksQueueOptions: { size: 0.2 }
461 }
462 )
948faff7 463 ).toThrow(
5c4c2dee
JB
464 new TypeError('Invalid worker node tasks queue size: must be an integer')
465 )
d4aeae5a
JB
466 })
467
2431bdb4 468 it('Verify that pool worker choice strategy options can be set', async () => {
a20f0ba5
JB
469 const pool = new FixedThreadPool(
470 numberOfWorkers,
b2fd3f4a 471 './tests/worker-files/thread/testWorker.mjs',
a20f0ba5
JB
472 { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
473 )
26ce26ca 474 expect(pool.opts.workerChoiceStrategyOptions).toBeUndefined()
8990357d 475 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
476 retries:
477 pool.info.maxSize +
478 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
932fc8be 479 runTime: { median: false },
5df69fab 480 waitTime: { median: false },
00e1bdeb
JB
481 elu: { median: false },
482 weights: expect.objectContaining({
483 0: expect.any(Number),
484 [pool.info.maxSize - 1]: expect.any(Number)
485 })
a20f0ba5
JB
486 })
487 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
488 .workerChoiceStrategies) {
86bf340d 489 expect(workerChoiceStrategy.opts).toStrictEqual({
00e1bdeb
JB
490 retries:
491 pool.info.maxSize +
492 Object.keys(workerChoiceStrategy.opts.weights).length,
932fc8be 493 runTime: { median: false },
5df69fab 494 waitTime: { median: false },
00e1bdeb
JB
495 elu: { median: false },
496 weights: expect.objectContaining({
497 0: expect.any(Number),
498 [pool.info.maxSize - 1]: expect.any(Number)
499 })
86bf340d 500 })
a20f0ba5 501 }
87de9ff5
JB
502 expect(
503 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
504 ).toStrictEqual({
932fc8be
JB
505 runTime: {
506 aggregate: true,
507 average: true,
508 median: false
509 },
510 waitTime: {
511 aggregate: false,
512 average: false,
513 median: false
514 },
5df69fab 515 elu: {
9adcefab
JB
516 aggregate: true,
517 average: true,
5df69fab
JB
518 median: false
519 }
86bf340d 520 })
9adcefab
JB
521 pool.setWorkerChoiceStrategyOptions({
522 runTime: { median: true },
523 elu: { median: true }
524 })
a20f0ba5 525 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
9adcefab 526 runTime: { median: true },
8990357d
JB
527 elu: { median: true }
528 })
529 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
530 retries:
531 pool.info.maxSize +
532 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
8990357d
JB
533 runTime: { median: true },
534 waitTime: { median: false },
00e1bdeb
JB
535 elu: { median: true },
536 weights: expect.objectContaining({
537 0: expect.any(Number),
538 [pool.info.maxSize - 1]: expect.any(Number)
539 })
a20f0ba5
JB
540 })
541 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
542 .workerChoiceStrategies) {
932fc8be 543 expect(workerChoiceStrategy.opts).toStrictEqual({
00e1bdeb
JB
544 retries:
545 pool.info.maxSize +
546 Object.keys(workerChoiceStrategy.opts.weights).length,
9adcefab 547 runTime: { median: true },
8990357d 548 waitTime: { median: false },
00e1bdeb
JB
549 elu: { median: true },
550 weights: expect.objectContaining({
551 0: expect.any(Number),
552 [pool.info.maxSize - 1]: expect.any(Number)
553 })
932fc8be 554 })
a20f0ba5 555 }
87de9ff5
JB
556 expect(
557 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
558 ).toStrictEqual({
932fc8be
JB
559 runTime: {
560 aggregate: true,
561 average: false,
562 median: true
563 },
564 waitTime: {
565 aggregate: false,
566 average: false,
567 median: false
568 },
5df69fab 569 elu: {
9adcefab 570 aggregate: true,
5df69fab 571 average: false,
9adcefab 572 median: true
5df69fab 573 }
86bf340d 574 })
9adcefab
JB
575 pool.setWorkerChoiceStrategyOptions({
576 runTime: { median: false },
577 elu: { median: false }
578 })
a20f0ba5 579 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d 580 runTime: { median: false },
8990357d
JB
581 elu: { median: false }
582 })
583 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
584 retries:
585 pool.info.maxSize +
586 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
9adcefab 587 runTime: { median: false },
8990357d 588 waitTime: { median: false },
00e1bdeb
JB
589 elu: { median: false },
590 weights: expect.objectContaining({
591 0: expect.any(Number),
592 [pool.info.maxSize - 1]: expect.any(Number)
593 })
a20f0ba5
JB
594 })
595 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
596 .workerChoiceStrategies) {
932fc8be 597 expect(workerChoiceStrategy.opts).toStrictEqual({
00e1bdeb
JB
598 retries:
599 pool.info.maxSize +
600 Object.keys(workerChoiceStrategy.opts.weights).length,
9adcefab 601 runTime: { median: false },
8990357d 602 waitTime: { median: false },
00e1bdeb
JB
603 elu: { median: false },
604 weights: expect.objectContaining({
605 0: expect.any(Number),
606 [pool.info.maxSize - 1]: expect.any(Number)
607 })
932fc8be 608 })
a20f0ba5 609 }
87de9ff5
JB
610 expect(
611 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
612 ).toStrictEqual({
932fc8be
JB
613 runTime: {
614 aggregate: true,
615 average: true,
616 median: false
617 },
618 waitTime: {
619 aggregate: false,
620 average: false,
621 median: false
622 },
5df69fab 623 elu: {
9adcefab
JB
624 aggregate: true,
625 average: true,
5df69fab
JB
626 median: false
627 }
86bf340d 628 })
1f95d544
JB
629 expect(() =>
630 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
948faff7 631 ).toThrow(
8735b4e5
JB
632 new TypeError(
633 'Invalid worker choice strategy options: must be a plain object'
634 )
1f95d544 635 )
948faff7 636 expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow(
8735b4e5
JB
637 new Error(
638 'Invalid worker choice strategy options: must have a weight for each worker node'
639 )
1f95d544
JB
640 )
641 expect(() =>
642 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
948faff7 643 ).toThrow(
8735b4e5
JB
644 new Error(
645 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
646 )
1f95d544 647 )
a20f0ba5
JB
648 await pool.destroy()
649 })
650
2431bdb4 651 it('Verify that pool tasks queue can be enabled/disabled', async () => {
a20f0ba5
JB
652 const pool = new FixedThreadPool(
653 numberOfWorkers,
b2fd3f4a 654 './tests/worker-files/thread/testWorker.mjs'
a20f0ba5
JB
655 )
656 expect(pool.opts.enableTasksQueue).toBe(false)
657 expect(pool.opts.tasksQueueOptions).toBeUndefined()
658 pool.enableTasksQueue(true)
659 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
660 expect(pool.opts.tasksQueueOptions).toStrictEqual({
661 concurrency: 1,
2324f8c9 662 size: Math.pow(numberOfWorkers, 2),
dbd73092 663 taskStealing: true,
32b141fd 664 tasksStealingOnBackPressure: true,
568d0075 665 tasksFinishedTimeout: 2000
20c6f652 666 })
a20f0ba5
JB
667 pool.enableTasksQueue(true, { concurrency: 2 })
668 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
669 expect(pool.opts.tasksQueueOptions).toStrictEqual({
670 concurrency: 2,
2324f8c9 671 size: Math.pow(numberOfWorkers, 2),
dbd73092 672 taskStealing: true,
32b141fd 673 tasksStealingOnBackPressure: true,
568d0075 674 tasksFinishedTimeout: 2000
20c6f652 675 })
a20f0ba5
JB
676 pool.enableTasksQueue(false)
677 expect(pool.opts.enableTasksQueue).toBe(false)
678 expect(pool.opts.tasksQueueOptions).toBeUndefined()
679 await pool.destroy()
680 })
681
2431bdb4 682 it('Verify that pool tasks queue options can be set', async () => {
a20f0ba5
JB
683 const pool = new FixedThreadPool(
684 numberOfWorkers,
b2fd3f4a 685 './tests/worker-files/thread/testWorker.mjs',
a20f0ba5
JB
686 { enableTasksQueue: true }
687 )
20c6f652
JB
688 expect(pool.opts.tasksQueueOptions).toStrictEqual({
689 concurrency: 1,
2324f8c9 690 size: Math.pow(numberOfWorkers, 2),
dbd73092 691 taskStealing: true,
32b141fd 692 tasksStealingOnBackPressure: true,
568d0075 693 tasksFinishedTimeout: 2000
20c6f652 694 })
d6ca1416 695 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
696 expect(workerNode.tasksQueueBackPressureSize).toBe(
697 pool.opts.tasksQueueOptions.size
698 )
d6ca1416
JB
699 }
700 pool.setTasksQueueOptions({
701 concurrency: 2,
2324f8c9 702 size: 2,
d6ca1416 703 taskStealing: false,
32b141fd 704 tasksStealingOnBackPressure: false,
568d0075 705 tasksFinishedTimeout: 3000
d6ca1416 706 })
20c6f652
JB
707 expect(pool.opts.tasksQueueOptions).toStrictEqual({
708 concurrency: 2,
2324f8c9 709 size: 2,
d6ca1416 710 taskStealing: false,
32b141fd 711 tasksStealingOnBackPressure: false,
568d0075 712 tasksFinishedTimeout: 3000
d6ca1416
JB
713 })
714 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
715 expect(workerNode.tasksQueueBackPressureSize).toBe(
716 pool.opts.tasksQueueOptions.size
717 )
d6ca1416
JB
718 }
719 pool.setTasksQueueOptions({
720 concurrency: 1,
721 taskStealing: true,
722 tasksStealingOnBackPressure: true
723 })
724 expect(pool.opts.tasksQueueOptions).toStrictEqual({
725 concurrency: 1,
2324f8c9 726 size: Math.pow(numberOfWorkers, 2),
dbd73092 727 taskStealing: true,
32b141fd 728 tasksStealingOnBackPressure: true,
568d0075 729 tasksFinishedTimeout: 2000
20c6f652 730 })
d6ca1416 731 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
732 expect(workerNode.tasksQueueBackPressureSize).toBe(
733 pool.opts.tasksQueueOptions.size
734 )
d6ca1416 735 }
948faff7 736 expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions')).toThrow(
8735b4e5
JB
737 new TypeError('Invalid tasks queue options: must be a plain object')
738 )
948faff7 739 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrow(
e695d66f 740 new RangeError(
20c6f652 741 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
742 )
743 )
948faff7 744 expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrow(
e695d66f 745 new RangeError(
20c6f652 746 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
8735b4e5 747 )
a20f0ba5 748 )
948faff7 749 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrow(
20c6f652
JB
750 new TypeError('Invalid worker node tasks concurrency: must be an integer')
751 )
948faff7 752 expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrow(
20c6f652 753 new RangeError(
68dbcdc0 754 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
20c6f652
JB
755 )
756 )
948faff7 757 expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrow(
20c6f652 758 new RangeError(
68dbcdc0 759 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
20c6f652
JB
760 )
761 )
948faff7 762 expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrow(
68dbcdc0 763 new TypeError('Invalid worker node tasks queue size: must be an integer')
f0d7f803 764 )
a20f0ba5
JB
765 await pool.destroy()
766 })
767
6b27d407
JB
768 it('Verify that pool info is set', async () => {
769 let pool = new FixedThreadPool(
770 numberOfWorkers,
b2fd3f4a 771 './tests/worker-files/thread/testWorker.mjs'
6b27d407 772 )
2dca6cab
JB
773 expect(pool.info).toStrictEqual({
774 version,
775 type: PoolTypes.fixed,
776 worker: WorkerTypes.thread,
47352846 777 started: true,
2dca6cab
JB
778 ready: true,
779 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
780 minSize: numberOfWorkers,
781 maxSize: numberOfWorkers,
782 workerNodes: numberOfWorkers,
783 idleWorkerNodes: numberOfWorkers,
784 busyWorkerNodes: 0,
785 executedTasks: 0,
786 executingTasks: 0,
2dca6cab
JB
787 failedTasks: 0
788 })
6b27d407
JB
789 await pool.destroy()
790 pool = new DynamicClusterPool(
2431bdb4 791 Math.floor(numberOfWorkers / 2),
6b27d407 792 numberOfWorkers,
ecdfbdc0 793 './tests/worker-files/cluster/testWorker.js'
6b27d407 794 )
2dca6cab
JB
795 expect(pool.info).toStrictEqual({
796 version,
797 type: PoolTypes.dynamic,
798 worker: WorkerTypes.cluster,
47352846 799 started: true,
2dca6cab
JB
800 ready: true,
801 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
802 minSize: Math.floor(numberOfWorkers / 2),
803 maxSize: numberOfWorkers,
804 workerNodes: Math.floor(numberOfWorkers / 2),
805 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
806 busyWorkerNodes: 0,
807 executedTasks: 0,
808 executingTasks: 0,
2dca6cab
JB
809 failedTasks: 0
810 })
6b27d407
JB
811 await pool.destroy()
812 })
813
2431bdb4 814 it('Verify that pool worker tasks usage are initialized', async () => {
bf9549ae
JB
815 const pool = new FixedClusterPool(
816 numberOfWorkers,
817 './tests/worker-files/cluster/testWorker.js'
818 )
f06e48d8 819 for (const workerNode of pool.workerNodes) {
47352846 820 expect(workerNode).toBeInstanceOf(WorkerNode)
465b2940 821 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
822 tasks: {
823 executed: 0,
824 executing: 0,
825 queued: 0,
df593701 826 maxQueued: 0,
463226a4 827 sequentiallyStolen: 0,
68cbdc84 828 stolen: 0,
a4e07f72
JB
829 failed: 0
830 },
831 runTime: {
4ba4c7f9 832 history: new CircularArray()
a4e07f72
JB
833 },
834 waitTime: {
4ba4c7f9 835 history: new CircularArray()
a4e07f72 836 },
5df69fab
JB
837 elu: {
838 idle: {
4ba4c7f9 839 history: new CircularArray()
5df69fab
JB
840 },
841 active: {
4ba4c7f9 842 history: new CircularArray()
f7510105 843 }
5df69fab 844 }
86bf340d 845 })
f06e48d8
JB
846 }
847 await pool.destroy()
848 })
849
2431bdb4
JB
850 it('Verify that pool worker tasks queue are initialized', async () => {
851 let pool = new FixedClusterPool(
f06e48d8
JB
852 numberOfWorkers,
853 './tests/worker-files/cluster/testWorker.js'
854 )
855 for (const workerNode of pool.workerNodes) {
47352846 856 expect(workerNode).toBeInstanceOf(WorkerNode)
574b351d 857 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
4d8bf9e4 858 expect(workerNode.tasksQueue.size).toBe(0)
9c16fb4b 859 expect(workerNode.tasksQueue.maxSize).toBe(0)
bf9549ae 860 }
fd7ebd49 861 await pool.destroy()
2431bdb4
JB
862 pool = new DynamicThreadPool(
863 Math.floor(numberOfWorkers / 2),
864 numberOfWorkers,
b2fd3f4a 865 './tests/worker-files/thread/testWorker.mjs'
2431bdb4
JB
866 )
867 for (const workerNode of pool.workerNodes) {
47352846 868 expect(workerNode).toBeInstanceOf(WorkerNode)
574b351d 869 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
2431bdb4
JB
870 expect(workerNode.tasksQueue.size).toBe(0)
871 expect(workerNode.tasksQueue.maxSize).toBe(0)
872 }
213cbac6 873 await pool.destroy()
2431bdb4
JB
874 })
875
876 it('Verify that pool worker info are initialized', async () => {
877 let pool = new FixedClusterPool(
878 numberOfWorkers,
879 './tests/worker-files/cluster/testWorker.js'
880 )
2dca6cab 881 for (const workerNode of pool.workerNodes) {
47352846 882 expect(workerNode).toBeInstanceOf(WorkerNode)
2dca6cab
JB
883 expect(workerNode.info).toStrictEqual({
884 id: expect.any(Number),
885 type: WorkerTypes.cluster,
886 dynamic: false,
887 ready: true
888 })
889 }
2431bdb4
JB
890 await pool.destroy()
891 pool = new DynamicThreadPool(
892 Math.floor(numberOfWorkers / 2),
893 numberOfWorkers,
b2fd3f4a 894 './tests/worker-files/thread/testWorker.mjs'
2431bdb4 895 )
2dca6cab 896 for (const workerNode of pool.workerNodes) {
47352846 897 expect(workerNode).toBeInstanceOf(WorkerNode)
2dca6cab
JB
898 expect(workerNode.info).toStrictEqual({
899 id: expect.any(Number),
900 type: WorkerTypes.thread,
901 dynamic: false,
7884d183 902 ready: true
2dca6cab
JB
903 })
904 }
213cbac6 905 await pool.destroy()
bf9549ae
JB
906 })
907
711623b8
JB
908 it('Verify that pool statuses are checked at start or destroy', async () => {
909 const pool = new FixedThreadPool(
910 numberOfWorkers,
911 './tests/worker-files/thread/testWorker.mjs'
912 )
913 expect(pool.info.started).toBe(true)
914 expect(pool.info.ready).toBe(true)
915 expect(() => pool.start()).toThrow(
916 new Error('Cannot start an already started pool')
917 )
918 await pool.destroy()
919 expect(pool.info.started).toBe(false)
920 expect(pool.info.ready).toBe(false)
921 await expect(pool.destroy()).rejects.toThrow(
922 new Error('Cannot destroy an already destroyed pool')
923 )
924 })
925
47352846
JB
926 it('Verify that pool can be started after initialization', async () => {
927 const pool = new FixedClusterPool(
928 numberOfWorkers,
929 './tests/worker-files/cluster/testWorker.js',
930 {
931 startWorkers: false
932 }
933 )
934 expect(pool.info.started).toBe(false)
935 expect(pool.info.ready).toBe(false)
55082af9 936 expect(pool.readyEventEmitted).toBe(false)
47352846 937 expect(pool.workerNodes).toStrictEqual([])
948faff7 938 await expect(pool.execute()).rejects.toThrow(
47352846
JB
939 new Error('Cannot execute a task on not started pool')
940 )
941 pool.start()
942 expect(pool.info.started).toBe(true)
943 expect(pool.info.ready).toBe(true)
55082af9
JB
944 await waitPoolEvents(pool, PoolEvents.ready, 1)
945 expect(pool.readyEventEmitted).toBe(true)
47352846
JB
946 expect(pool.workerNodes.length).toBe(numberOfWorkers)
947 for (const workerNode of pool.workerNodes) {
948 expect(workerNode).toBeInstanceOf(WorkerNode)
949 }
950 await pool.destroy()
951 })
952
9d2d0da1
JB
953 it('Verify that pool execute() arguments are checked', async () => {
954 const pool = new FixedClusterPool(
955 numberOfWorkers,
956 './tests/worker-files/cluster/testWorker.js'
957 )
948faff7 958 await expect(pool.execute(undefined, 0)).rejects.toThrow(
9d2d0da1
JB
959 new TypeError('name argument must be a string')
960 )
948faff7 961 await expect(pool.execute(undefined, '')).rejects.toThrow(
9d2d0da1
JB
962 new TypeError('name argument must not be an empty string')
963 )
948faff7 964 await expect(pool.execute(undefined, undefined, {})).rejects.toThrow(
9d2d0da1
JB
965 new TypeError('transferList argument must be an array')
966 )
967 await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
968 "Task function 'unknown' not found"
969 )
970 await pool.destroy()
948faff7 971 await expect(pool.execute()).rejects.toThrow(
47352846 972 new Error('Cannot execute a task on not started pool')
9d2d0da1
JB
973 )
974 })
975
2431bdb4 976 it('Verify that pool worker tasks usage are computed', async () => {
bf9549ae
JB
977 const pool = new FixedClusterPool(
978 numberOfWorkers,
979 './tests/worker-files/cluster/testWorker.js'
980 )
09c2d0d3 981 const promises = new Set()
fc027381
JB
982 const maxMultiplier = 2
983 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 984 promises.add(pool.execute())
bf9549ae 985 }
f06e48d8 986 for (const workerNode of pool.workerNodes) {
465b2940 987 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
988 tasks: {
989 executed: 0,
990 executing: maxMultiplier,
991 queued: 0,
df593701 992 maxQueued: 0,
463226a4 993 sequentiallyStolen: 0,
68cbdc84 994 stolen: 0,
a4e07f72
JB
995 failed: 0
996 },
997 runTime: {
a4e07f72
JB
998 history: expect.any(CircularArray)
999 },
1000 waitTime: {
a4e07f72
JB
1001 history: expect.any(CircularArray)
1002 },
5df69fab
JB
1003 elu: {
1004 idle: {
5df69fab
JB
1005 history: expect.any(CircularArray)
1006 },
1007 active: {
5df69fab 1008 history: expect.any(CircularArray)
f7510105 1009 }
5df69fab 1010 }
86bf340d 1011 })
bf9549ae
JB
1012 }
1013 await Promise.all(promises)
f06e48d8 1014 for (const workerNode of pool.workerNodes) {
465b2940 1015 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
1016 tasks: {
1017 executed: maxMultiplier,
1018 executing: 0,
1019 queued: 0,
df593701 1020 maxQueued: 0,
463226a4 1021 sequentiallyStolen: 0,
68cbdc84 1022 stolen: 0,
a4e07f72
JB
1023 failed: 0
1024 },
1025 runTime: {
a4e07f72
JB
1026 history: expect.any(CircularArray)
1027 },
1028 waitTime: {
a4e07f72
JB
1029 history: expect.any(CircularArray)
1030 },
5df69fab
JB
1031 elu: {
1032 idle: {
5df69fab
JB
1033 history: expect.any(CircularArray)
1034 },
1035 active: {
5df69fab 1036 history: expect.any(CircularArray)
f7510105 1037 }
5df69fab 1038 }
86bf340d 1039 })
bf9549ae 1040 }
fd7ebd49 1041 await pool.destroy()
bf9549ae
JB
1042 })
1043
2431bdb4 1044 it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
7fd82a1c 1045 const pool = new DynamicThreadPool(
2431bdb4 1046 Math.floor(numberOfWorkers / 2),
8f4878b7 1047 numberOfWorkers,
b2fd3f4a 1048 './tests/worker-files/thread/testWorker.mjs'
9e619829 1049 )
09c2d0d3 1050 const promises = new Set()
ee9f5295
JB
1051 const maxMultiplier = 2
1052 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 1053 promises.add(pool.execute())
9e619829
JB
1054 }
1055 await Promise.all(promises)
f06e48d8 1056 for (const workerNode of pool.workerNodes) {
465b2940 1057 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
1058 tasks: {
1059 executed: expect.any(Number),
1060 executing: 0,
1061 queued: 0,
df593701 1062 maxQueued: 0,
463226a4 1063 sequentiallyStolen: 0,
68cbdc84 1064 stolen: 0,
a4e07f72
JB
1065 failed: 0
1066 },
1067 runTime: {
a4e07f72
JB
1068 history: expect.any(CircularArray)
1069 },
1070 waitTime: {
a4e07f72
JB
1071 history: expect.any(CircularArray)
1072 },
5df69fab
JB
1073 elu: {
1074 idle: {
5df69fab
JB
1075 history: expect.any(CircularArray)
1076 },
1077 active: {
5df69fab 1078 history: expect.any(CircularArray)
f7510105 1079 }
5df69fab 1080 }
86bf340d 1081 })
465b2940 1082 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
94407def
JB
1083 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1084 numberOfWorkers * maxMultiplier
1085 )
b97d82d8
JB
1086 expect(workerNode.usage.runTime.history.length).toBe(0)
1087 expect(workerNode.usage.waitTime.history.length).toBe(0)
1088 expect(workerNode.usage.elu.idle.history.length).toBe(0)
1089 expect(workerNode.usage.elu.active.history.length).toBe(0)
9e619829
JB
1090 }
1091 pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
f06e48d8 1092 for (const workerNode of pool.workerNodes) {
465b2940 1093 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
1094 tasks: {
1095 executed: 0,
1096 executing: 0,
1097 queued: 0,
df593701 1098 maxQueued: 0,
463226a4 1099 sequentiallyStolen: 0,
68cbdc84 1100 stolen: 0,
a4e07f72
JB
1101 failed: 0
1102 },
1103 runTime: {
a4e07f72
JB
1104 history: expect.any(CircularArray)
1105 },
1106 waitTime: {
a4e07f72
JB
1107 history: expect.any(CircularArray)
1108 },
5df69fab
JB
1109 elu: {
1110 idle: {
5df69fab
JB
1111 history: expect.any(CircularArray)
1112 },
1113 active: {
5df69fab 1114 history: expect.any(CircularArray)
f7510105 1115 }
5df69fab 1116 }
86bf340d 1117 })
465b2940
JB
1118 expect(workerNode.usage.runTime.history.length).toBe(0)
1119 expect(workerNode.usage.waitTime.history.length).toBe(0)
b97d82d8
JB
1120 expect(workerNode.usage.elu.idle.history.length).toBe(0)
1121 expect(workerNode.usage.elu.active.history.length).toBe(0)
ee11a4a2 1122 }
fd7ebd49 1123 await pool.destroy()
ee11a4a2
JB
1124 })
1125
a1763c54
JB
1126 it("Verify that pool event emitter 'ready' event can register a callback", async () => {
1127 const pool = new DynamicClusterPool(
2431bdb4 1128 Math.floor(numberOfWorkers / 2),
164d950a 1129 numberOfWorkers,
a1763c54 1130 './tests/worker-files/cluster/testWorker.js'
164d950a 1131 )
c726f66c 1132 expect(pool.emitter.eventNames()).toStrictEqual([])
d46660cd 1133 let poolInfo
a1763c54 1134 let poolReady = 0
041dc05b 1135 pool.emitter.on(PoolEvents.ready, info => {
a1763c54 1136 ++poolReady
d46660cd
JB
1137 poolInfo = info
1138 })
a1763c54 1139 await waitPoolEvents(pool, PoolEvents.ready, 1)
c726f66c 1140 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.ready])
a1763c54 1141 expect(poolReady).toBe(1)
d46660cd 1142 expect(poolInfo).toStrictEqual({
23ccf9d7 1143 version,
d46660cd 1144 type: PoolTypes.dynamic,
a1763c54 1145 worker: WorkerTypes.cluster,
47352846 1146 started: true,
a1763c54 1147 ready: true,
2431bdb4
JB
1148 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1149 minSize: expect.any(Number),
1150 maxSize: expect.any(Number),
1151 workerNodes: expect.any(Number),
1152 idleWorkerNodes: expect.any(Number),
1153 busyWorkerNodes: expect.any(Number),
1154 executedTasks: expect.any(Number),
1155 executingTasks: expect.any(Number),
2431bdb4
JB
1156 failedTasks: expect.any(Number)
1157 })
1158 await pool.destroy()
1159 })
1160
a1763c54
JB
1161 it("Verify that pool event emitter 'busy' event can register a callback", async () => {
1162 const pool = new FixedThreadPool(
2431bdb4 1163 numberOfWorkers,
b2fd3f4a 1164 './tests/worker-files/thread/testWorker.mjs'
2431bdb4 1165 )
c726f66c 1166 expect(pool.emitter.eventNames()).toStrictEqual([])
a1763c54
JB
1167 const promises = new Set()
1168 let poolBusy = 0
2431bdb4 1169 let poolInfo
041dc05b 1170 pool.emitter.on(PoolEvents.busy, info => {
a1763c54 1171 ++poolBusy
2431bdb4
JB
1172 poolInfo = info
1173 })
c726f66c 1174 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
a1763c54
JB
1175 for (let i = 0; i < numberOfWorkers * 2; i++) {
1176 promises.add(pool.execute())
1177 }
1178 await Promise.all(promises)
1179 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
1180 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
1181 expect(poolBusy).toBe(numberOfWorkers + 1)
2431bdb4
JB
1182 expect(poolInfo).toStrictEqual({
1183 version,
a1763c54
JB
1184 type: PoolTypes.fixed,
1185 worker: WorkerTypes.thread,
47352846
JB
1186 started: true,
1187 ready: true,
2431bdb4 1188 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
1189 minSize: expect.any(Number),
1190 maxSize: expect.any(Number),
1191 workerNodes: expect.any(Number),
1192 idleWorkerNodes: expect.any(Number),
1193 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1194 executedTasks: expect.any(Number),
1195 executingTasks: expect.any(Number),
a4e07f72 1196 failedTasks: expect.any(Number)
d46660cd 1197 })
164d950a
JB
1198 await pool.destroy()
1199 })
1200
a1763c54
JB
1201 it("Verify that pool event emitter 'full' event can register a callback", async () => {
1202 const pool = new DynamicThreadPool(
1203 Math.floor(numberOfWorkers / 2),
7c0ba920 1204 numberOfWorkers,
b2fd3f4a 1205 './tests/worker-files/thread/testWorker.mjs'
7c0ba920 1206 )
c726f66c 1207 expect(pool.emitter.eventNames()).toStrictEqual([])
09c2d0d3 1208 const promises = new Set()
a1763c54 1209 let poolFull = 0
d46660cd 1210 let poolInfo
041dc05b 1211 pool.emitter.on(PoolEvents.full, info => {
a1763c54 1212 ++poolFull
d46660cd
JB
1213 poolInfo = info
1214 })
c726f66c 1215 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.full])
7c0ba920 1216 for (let i = 0; i < numberOfWorkers * 2; i++) {
f5d14e90 1217 promises.add(pool.execute())
7c0ba920 1218 }
cf597bc5 1219 await Promise.all(promises)
33e6bb4c 1220 expect(poolFull).toBe(1)
d46660cd 1221 expect(poolInfo).toStrictEqual({
23ccf9d7 1222 version,
a1763c54 1223 type: PoolTypes.dynamic,
d46660cd 1224 worker: WorkerTypes.thread,
47352846
JB
1225 started: true,
1226 ready: true,
2431bdb4 1227 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
8735b4e5
JB
1228 minSize: expect.any(Number),
1229 maxSize: expect.any(Number),
1230 workerNodes: expect.any(Number),
1231 idleWorkerNodes: expect.any(Number),
1232 busyWorkerNodes: expect.any(Number),
1233 executedTasks: expect.any(Number),
1234 executingTasks: expect.any(Number),
1235 failedTasks: expect.any(Number)
1236 })
1237 await pool.destroy()
1238 })
1239
3e8611a8 1240 it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
b1aae695 1241 const pool = new FixedThreadPool(
8735b4e5 1242 numberOfWorkers,
b2fd3f4a 1243 './tests/worker-files/thread/testWorker.mjs',
8735b4e5
JB
1244 {
1245 enableTasksQueue: true
1246 }
1247 )
a074ffee 1248 stub(pool, 'hasBackPressure').returns(true)
c726f66c 1249 expect(pool.emitter.eventNames()).toStrictEqual([])
8735b4e5
JB
1250 const promises = new Set()
1251 let poolBackPressure = 0
1252 let poolInfo
041dc05b 1253 pool.emitter.on(PoolEvents.backPressure, info => {
8735b4e5
JB
1254 ++poolBackPressure
1255 poolInfo = info
1256 })
c726f66c 1257 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.backPressure])
033f1776 1258 for (let i = 0; i < numberOfWorkers + 1; i++) {
8735b4e5
JB
1259 promises.add(pool.execute())
1260 }
1261 await Promise.all(promises)
033f1776 1262 expect(poolBackPressure).toBe(1)
8735b4e5
JB
1263 expect(poolInfo).toStrictEqual({
1264 version,
3e8611a8 1265 type: PoolTypes.fixed,
8735b4e5 1266 worker: WorkerTypes.thread,
47352846
JB
1267 started: true,
1268 ready: true,
8735b4e5 1269 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
1270 minSize: expect.any(Number),
1271 maxSize: expect.any(Number),
1272 workerNodes: expect.any(Number),
1273 idleWorkerNodes: expect.any(Number),
1274 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1275 executedTasks: expect.any(Number),
1276 executingTasks: expect.any(Number),
3e8611a8
JB
1277 maxQueuedTasks: expect.any(Number),
1278 queuedTasks: expect.any(Number),
1279 backPressure: true,
68cbdc84 1280 stolenTasks: expect.any(Number),
a4e07f72 1281 failedTasks: expect.any(Number)
d46660cd 1282 })
e458c82e 1283 expect(pool.hasBackPressure.callCount).toBe(5)
fd7ebd49 1284 await pool.destroy()
7c0ba920 1285 })
70a4f5ea 1286
85b2561d
JB
1287 it('Verify that destroy() waits for queued tasks to finish', async () => {
1288 const tasksFinishedTimeout = 2500
1289 const pool = new FixedThreadPool(
1290 numberOfWorkers,
1291 './tests/worker-files/thread/asyncWorker.mjs',
1292 {
1293 enableTasksQueue: true,
1294 tasksQueueOptions: { tasksFinishedTimeout }
1295 }
1296 )
1297 const maxMultiplier = 4
1298 let tasksFinished = 0
1299 for (const workerNode of pool.workerNodes) {
1300 workerNode.on('taskFinished', () => {
1301 ++tasksFinished
1302 })
1303 }
1304 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1305 pool.execute()
1306 }
1307 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1308 const startTime = performance.now()
1309 await pool.destroy()
1310 const elapsedTime = performance.now() - startTime
1311 expect(tasksFinished).toBe(numberOfWorkers * maxMultiplier)
1312 expect(elapsedTime).toBeGreaterThanOrEqual(2000)
6d41131f 1313 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 100)
85b2561d
JB
1314 })
1315
1316 it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
1317 const tasksFinishedTimeout = 1000
1318 const pool = new FixedThreadPool(
1319 numberOfWorkers,
1320 './tests/worker-files/thread/asyncWorker.mjs',
1321 {
1322 enableTasksQueue: true,
1323 tasksQueueOptions: { tasksFinishedTimeout }
1324 }
1325 )
1326 const maxMultiplier = 4
1327 let tasksFinished = 0
1328 for (const workerNode of pool.workerNodes) {
1329 workerNode.on('taskFinished', () => {
1330 ++tasksFinished
1331 })
1332 }
1333 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1334 pool.execute()
1335 }
1336 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1337 const startTime = performance.now()
1338 await pool.destroy()
1339 const elapsedTime = performance.now() - startTime
1340 expect(tasksFinished).toBe(0)
c004ecec 1341 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 600)
85b2561d
JB
1342 })
1343
f18fd12b
JB
1344 it('Verify that pool asynchronous resource track tasks execution', async () => {
1345 let taskAsyncId
1346 let initCalls = 0
1347 let beforeCalls = 0
1348 let afterCalls = 0
1349 let resolveCalls = 0
1350 const hook = createHook({
1351 init (asyncId, type) {
1352 if (type === 'poolifier:task') {
1353 initCalls++
1354 taskAsyncId = asyncId
1355 }
1356 },
1357 before (asyncId) {
1358 if (asyncId === taskAsyncId) beforeCalls++
1359 },
1360 after (asyncId) {
1361 if (asyncId === taskAsyncId) afterCalls++
1362 },
1363 promiseResolve () {
1364 if (executionAsyncId() === taskAsyncId) resolveCalls++
1365 }
1366 })
f18fd12b
JB
1367 const pool = new FixedThreadPool(
1368 numberOfWorkers,
1369 './tests/worker-files/thread/testWorker.mjs'
1370 )
8954c0a3 1371 hook.enable()
f18fd12b
JB
1372 await pool.execute()
1373 hook.disable()
1374 expect(initCalls).toBe(1)
1375 expect(beforeCalls).toBe(1)
1376 expect(afterCalls).toBe(1)
1377 expect(resolveCalls).toBe(1)
8954c0a3 1378 await pool.destroy()
f18fd12b
JB
1379 })
1380
9eae3c69
JB
1381 it('Verify that hasTaskFunction() is working', async () => {
1382 const dynamicThreadPool = new DynamicThreadPool(
1383 Math.floor(numberOfWorkers / 2),
1384 numberOfWorkers,
b2fd3f4a 1385 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
9eae3c69
JB
1386 )
1387 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1388 expect(dynamicThreadPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1389 expect(dynamicThreadPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1390 true
1391 )
1392 expect(dynamicThreadPool.hasTaskFunction('factorial')).toBe(true)
1393 expect(dynamicThreadPool.hasTaskFunction('fibonacci')).toBe(true)
1394 expect(dynamicThreadPool.hasTaskFunction('unknown')).toBe(false)
1395 await dynamicThreadPool.destroy()
1396 const fixedClusterPool = new FixedClusterPool(
1397 numberOfWorkers,
1398 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1399 )
1400 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1401 expect(fixedClusterPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1402 expect(fixedClusterPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1403 true
1404 )
1405 expect(fixedClusterPool.hasTaskFunction('factorial')).toBe(true)
1406 expect(fixedClusterPool.hasTaskFunction('fibonacci')).toBe(true)
1407 expect(fixedClusterPool.hasTaskFunction('unknown')).toBe(false)
1408 await fixedClusterPool.destroy()
1409 })
1410
1411 it('Verify that addTaskFunction() is working', async () => {
1412 const dynamicThreadPool = new DynamicThreadPool(
1413 Math.floor(numberOfWorkers / 2),
1414 numberOfWorkers,
b2fd3f4a 1415 './tests/worker-files/thread/testWorker.mjs'
9eae3c69
JB
1416 )
1417 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
3feeab69
JB
1418 await expect(
1419 dynamicThreadPool.addTaskFunction(0, () => {})
948faff7 1420 ).rejects.toThrow(new TypeError('name argument must be a string'))
3feeab69
JB
1421 await expect(
1422 dynamicThreadPool.addTaskFunction('', () => {})
948faff7 1423 ).rejects.toThrow(
3feeab69
JB
1424 new TypeError('name argument must not be an empty string')
1425 )
948faff7
JB
1426 await expect(dynamicThreadPool.addTaskFunction('test', 0)).rejects.toThrow(
1427 new TypeError('fn argument must be a function')
1428 )
1429 await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow(
1430 new TypeError('fn argument must be a function')
1431 )
9eae3c69
JB
1432 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1433 DEFAULT_TASK_NAME,
1434 'test'
1435 ])
1436 const echoTaskFunction = data => {
1437 return data
1438 }
1439 await expect(
1440 dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1441 ).resolves.toBe(true)
1442 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1443 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1444 echoTaskFunction
1445 )
1446 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1447 DEFAULT_TASK_NAME,
1448 'test',
1449 'echo'
1450 ])
1451 const taskFunctionData = { test: 'test' }
1452 const echoResult = await dynamicThreadPool.execute(taskFunctionData, 'echo')
1453 expect(echoResult).toStrictEqual(taskFunctionData)
adee6053
JB
1454 for (const workerNode of dynamicThreadPool.workerNodes) {
1455 expect(workerNode.getTaskFunctionWorkerUsage('echo')).toStrictEqual({
1456 tasks: {
1457 executed: expect.any(Number),
1458 executing: 0,
1459 queued: 0,
463226a4 1460 sequentiallyStolen: 0,
5ad42e34 1461 stolen: 0,
adee6053
JB
1462 failed: 0
1463 },
1464 runTime: {
1465 history: new CircularArray()
1466 },
1467 waitTime: {
1468 history: new CircularArray()
1469 },
1470 elu: {
1471 idle: {
1472 history: new CircularArray()
1473 },
1474 active: {
1475 history: new CircularArray()
1476 }
1477 }
1478 })
1479 }
9eae3c69
JB
1480 await dynamicThreadPool.destroy()
1481 })
1482
1483 it('Verify that removeTaskFunction() is working', async () => {
1484 const dynamicThreadPool = new DynamicThreadPool(
1485 Math.floor(numberOfWorkers / 2),
1486 numberOfWorkers,
b2fd3f4a 1487 './tests/worker-files/thread/testWorker.mjs'
9eae3c69
JB
1488 )
1489 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1490 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1491 DEFAULT_TASK_NAME,
1492 'test'
1493 ])
948faff7 1494 await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow(
16248b23 1495 new Error('Cannot remove a task function not handled on the pool side')
9eae3c69
JB
1496 )
1497 const echoTaskFunction = data => {
1498 return data
1499 }
1500 await dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1501 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1502 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1503 echoTaskFunction
1504 )
1505 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1506 DEFAULT_TASK_NAME,
1507 'test',
1508 'echo'
1509 ])
1510 await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe(
1511 true
1512 )
1513 expect(dynamicThreadPool.taskFunctions.size).toBe(0)
1514 expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined()
1515 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1516 DEFAULT_TASK_NAME,
1517 'test'
1518 ])
1519 await dynamicThreadPool.destroy()
1520 })
1521
30500265 1522 it('Verify that listTaskFunctionNames() is working', async () => {
90d7d101
JB
1523 const dynamicThreadPool = new DynamicThreadPool(
1524 Math.floor(numberOfWorkers / 2),
1525 numberOfWorkers,
b2fd3f4a 1526 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
90d7d101
JB
1527 )
1528 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
66979634 1529 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
6cd5248f 1530 DEFAULT_TASK_NAME,
90d7d101
JB
1531 'jsonIntegerSerialization',
1532 'factorial',
1533 'fibonacci'
1534 ])
9eae3c69 1535 await dynamicThreadPool.destroy()
90d7d101
JB
1536 const fixedClusterPool = new FixedClusterPool(
1537 numberOfWorkers,
1538 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1539 )
1540 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
66979634 1541 expect(fixedClusterPool.listTaskFunctionNames()).toStrictEqual([
6cd5248f 1542 DEFAULT_TASK_NAME,
90d7d101
JB
1543 'jsonIntegerSerialization',
1544 'factorial',
1545 'fibonacci'
1546 ])
0fe39c97 1547 await fixedClusterPool.destroy()
90d7d101
JB
1548 })
1549
9eae3c69 1550 it('Verify that setDefaultTaskFunction() is working', async () => {
30500265
JB
1551 const dynamicThreadPool = new DynamicThreadPool(
1552 Math.floor(numberOfWorkers / 2),
1553 numberOfWorkers,
b2fd3f4a 1554 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
30500265
JB
1555 )
1556 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
711623b8 1557 const workerId = dynamicThreadPool.workerNodes[0].info.id
948faff7 1558 await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow(
b0b55f57 1559 new Error(
711623b8 1560 `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'`
b0b55f57
JB
1561 )
1562 )
1563 await expect(
1564 dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME)
948faff7 1565 ).rejects.toThrow(
b0b55f57 1566 new Error(
711623b8 1567 `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function reserved name as the default task function'`
b0b55f57
JB
1568 )
1569 )
1570 await expect(
1571 dynamicThreadPool.setDefaultTaskFunction('unknown')
948faff7 1572 ).rejects.toThrow(
b0b55f57 1573 new Error(
711623b8 1574 `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function to a non-existing task function'`
b0b55f57
JB
1575 )
1576 )
9eae3c69
JB
1577 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1578 DEFAULT_TASK_NAME,
1579 'jsonIntegerSerialization',
1580 'factorial',
1581 'fibonacci'
1582 ])
1583 await expect(
1584 dynamicThreadPool.setDefaultTaskFunction('factorial')
1585 ).resolves.toBe(true)
1586 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1587 DEFAULT_TASK_NAME,
1588 'factorial',
1589 'jsonIntegerSerialization',
1590 'fibonacci'
1591 ])
1592 await expect(
1593 dynamicThreadPool.setDefaultTaskFunction('fibonacci')
1594 ).resolves.toBe(true)
1595 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1596 DEFAULT_TASK_NAME,
1597 'fibonacci',
1598 'jsonIntegerSerialization',
1599 'factorial'
1600 ])
cda9ba34 1601 await dynamicThreadPool.destroy()
30500265
JB
1602 })
1603
90d7d101 1604 it('Verify that multiple task functions worker is working', async () => {
70a4f5ea 1605 const pool = new DynamicClusterPool(
2431bdb4 1606 Math.floor(numberOfWorkers / 2),
70a4f5ea 1607 numberOfWorkers,
90d7d101 1608 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
70a4f5ea
JB
1609 )
1610 const data = { n: 10 }
82888165 1611 const result0 = await pool.execute(data)
30b963d4 1612 expect(result0).toStrictEqual({ ok: 1 })
70a4f5ea 1613 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
30b963d4 1614 expect(result1).toStrictEqual({ ok: 1 })
70a4f5ea
JB
1615 const result2 = await pool.execute(data, 'factorial')
1616 expect(result2).toBe(3628800)
1617 const result3 = await pool.execute(data, 'fibonacci')
024daf59 1618 expect(result3).toBe(55)
5bb5be17
JB
1619 expect(pool.info.executingTasks).toBe(0)
1620 expect(pool.info.executedTasks).toBe(4)
b414b84c 1621 for (const workerNode of pool.workerNodes) {
66979634 1622 expect(workerNode.info.taskFunctionNames).toStrictEqual([
6cd5248f 1623 DEFAULT_TASK_NAME,
b414b84c
JB
1624 'jsonIntegerSerialization',
1625 'factorial',
1626 'fibonacci'
1627 ])
1628 expect(workerNode.taskFunctionsUsage.size).toBe(3)
66979634 1629 for (const name of pool.listTaskFunctionNames()) {
5bb5be17
JB
1630 expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1631 tasks: {
1632 executed: expect.any(Number),
4ba4c7f9 1633 executing: 0,
5bb5be17 1634 failed: 0,
68cbdc84 1635 queued: 0,
463226a4 1636 sequentiallyStolen: 0,
68cbdc84 1637 stolen: 0
5bb5be17
JB
1638 },
1639 runTime: {
1640 history: expect.any(CircularArray)
1641 },
1642 waitTime: {
1643 history: expect.any(CircularArray)
1644 },
1645 elu: {
1646 idle: {
1647 history: expect.any(CircularArray)
1648 },
1649 active: {
1650 history: expect.any(CircularArray)
1651 }
1652 }
1653 })
1654 expect(
4ba4c7f9
JB
1655 workerNode.getTaskFunctionWorkerUsage(name).tasks.executed
1656 ).toBeGreaterThan(0)
5bb5be17 1657 }
dfd7ec01
JB
1658 expect(
1659 workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1660 ).toStrictEqual(
66979634
JB
1661 workerNode.getTaskFunctionWorkerUsage(
1662 workerNode.info.taskFunctionNames[1]
1663 )
dfd7ec01 1664 )
5bb5be17 1665 }
0fe39c97 1666 await pool.destroy()
70a4f5ea 1667 })
52a23942
JB
1668
1669 it('Verify sendKillMessageToWorker()', async () => {
1670 const pool = new DynamicClusterPool(
1671 Math.floor(numberOfWorkers / 2),
1672 numberOfWorkers,
1673 './tests/worker-files/cluster/testWorker.js'
1674 )
1675 const workerNodeKey = 0
1676 await expect(
adee6053 1677 pool.sendKillMessageToWorker(workerNodeKey)
52a23942 1678 ).resolves.toBeUndefined()
85b2561d
JB
1679 await expect(
1680 pool.sendKillMessageToWorker(numberOfWorkers)
1681 ).rejects.toStrictEqual(
1682 new Error(`Invalid worker node key '${numberOfWorkers}'`)
1683 )
52a23942
JB
1684 await pool.destroy()
1685 })
adee6053
JB
1686
1687 it('Verify sendTaskFunctionOperationToWorker()', async () => {
1688 const pool = new DynamicClusterPool(
1689 Math.floor(numberOfWorkers / 2),
1690 numberOfWorkers,
1691 './tests/worker-files/cluster/testWorker.js'
1692 )
1693 const workerNodeKey = 0
1694 await expect(
1695 pool.sendTaskFunctionOperationToWorker(workerNodeKey, {
1696 taskFunctionOperation: 'add',
1697 taskFunctionName: 'empty',
1698 taskFunction: (() => {}).toString()
1699 })
1700 ).resolves.toBe(true)
1701 expect(
1702 pool.workerNodes[workerNodeKey].info.taskFunctionNames
1703 ).toStrictEqual([DEFAULT_TASK_NAME, 'test', 'empty'])
1704 await pool.destroy()
1705 })
1706
1707 it('Verify sendTaskFunctionOperationToWorkers()', async () => {
1708 const pool = new DynamicClusterPool(
1709 Math.floor(numberOfWorkers / 2),
1710 numberOfWorkers,
1711 './tests/worker-files/cluster/testWorker.js'
1712 )
adee6053
JB
1713 await expect(
1714 pool.sendTaskFunctionOperationToWorkers({
1715 taskFunctionOperation: 'add',
1716 taskFunctionName: 'empty',
1717 taskFunction: (() => {}).toString()
1718 })
1719 ).resolves.toBe(true)
1720 for (const workerNode of pool.workerNodes) {
1721 expect(workerNode.info.taskFunctionNames).toStrictEqual([
1722 DEFAULT_TASK_NAME,
1723 'test',
1724 'empty'
1725 ])
1726 }
1727 await pool.destroy()
1728 })
3ec964d6 1729})