Merge branch 'master' of github.com:poolifier/poolifier
[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) {
86cbb766
JB
245 expect(workerChoiceStrategy.opts).toStrictEqual({
246 retries:
247 pool.info.maxSize +
248 Object.keys(workerChoiceStrategy.opts.weights).length,
249 runTime: { median: false },
250 waitTime: { median: false },
251 elu: { median: false },
252 weights: expect.objectContaining({
253 0: expect.any(Number),
254 [pool.info.maxSize - 1]: expect.any(Number)
00e1bdeb 255 })
86cbb766 256 })
999ef664 257 }
fd7ebd49 258 await pool.destroy()
73bfd59d 259 const testHandler = () => console.info('test handler executed')
7c0ba920
JB
260 pool = new FixedThreadPool(
261 numberOfWorkers,
b2fd3f4a 262 './tests/worker-files/thread/testWorker.mjs',
7c0ba920 263 {
e4543b14 264 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
49be33fe 265 workerChoiceStrategyOptions: {
932fc8be 266 runTime: { median: true },
fc027381 267 weights: { 0: 300, 1: 200 }
49be33fe 268 },
35cf1c03 269 enableEvents: false,
1f68cede 270 restartWorkerOnError: false,
ff733df7 271 enableTasksQueue: true,
d4aeae5a 272 tasksQueueOptions: { concurrency: 2 },
35cf1c03
JB
273 messageHandler: testHandler,
274 errorHandler: testHandler,
275 onlineHandler: testHandler,
276 exitHandler: testHandler
7c0ba920
JB
277 }
278 )
7c0ba920 279 expect(pool.emitter).toBeUndefined()
47352846
JB
280 expect(pool.opts).toStrictEqual({
281 startWorkers: true,
282 enableEvents: false,
283 restartWorkerOnError: false,
284 enableTasksQueue: true,
285 tasksQueueOptions: {
286 concurrency: 2,
2324f8c9 287 size: Math.pow(numberOfWorkers, 2),
dbd73092 288 taskStealing: true,
32b141fd 289 tasksStealingOnBackPressure: true,
568d0075 290 tasksFinishedTimeout: 2000
47352846
JB
291 },
292 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
293 workerChoiceStrategyOptions: {
47352846 294 runTime: { median: true },
47352846
JB
295 weights: { 0: 300, 1: 200 }
296 },
297 onlineHandler: testHandler,
298 messageHandler: testHandler,
299 errorHandler: testHandler,
300 exitHandler: testHandler
8990357d
JB
301 })
302 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
449cd154
JB
303 retries:
304 pool.info.maxSize +
305 Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length,
8990357d
JB
306 runTime: { median: true },
307 waitTime: { median: false },
308 elu: { median: false },
fc027381 309 weights: { 0: 300, 1: 200 }
da309861 310 })
999ef664
JB
311 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
312 .workerChoiceStrategies) {
313 expect(workerChoiceStrategy.opts).toStrictEqual({
449cd154
JB
314 retries:
315 pool.info.maxSize +
316 Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length,
999ef664
JB
317 runTime: { median: true },
318 waitTime: { median: false },
319 elu: { median: false },
320 weights: { 0: 300, 1: 200 }
321 })
322 }
fd7ebd49 323 await pool.destroy()
7c0ba920
JB
324 })
325
fe291b64 326 it('Verify that pool options are validated', () => {
d4aeae5a
JB
327 expect(
328 () =>
329 new FixedThreadPool(
330 numberOfWorkers,
b2fd3f4a 331 './tests/worker-files/thread/testWorker.mjs',
d4aeae5a 332 {
f0d7f803 333 workerChoiceStrategy: 'invalidStrategy'
d4aeae5a
JB
334 }
335 )
948faff7 336 ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'"))
49be33fe
JB
337 expect(
338 () =>
339 new FixedThreadPool(
340 numberOfWorkers,
b2fd3f4a 341 './tests/worker-files/thread/testWorker.mjs',
49be33fe
JB
342 {
343 workerChoiceStrategyOptions: { weights: {} }
344 }
345 )
948faff7 346 ).toThrow(
8735b4e5
JB
347 new Error(
348 'Invalid worker choice strategy options: must have a weight for each worker node'
349 )
49be33fe 350 )
f0d7f803
JB
351 expect(
352 () =>
353 new FixedThreadPool(
354 numberOfWorkers,
b2fd3f4a 355 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
356 {
357 workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
358 }
359 )
948faff7 360 ).toThrow(
8735b4e5
JB
361 new Error(
362 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
363 )
f0d7f803 364 )
5c4c2dee
JB
365 expect(
366 () =>
367 new FixedThreadPool(
368 numberOfWorkers,
b2fd3f4a 369 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
370 {
371 enableTasksQueue: true,
372 tasksQueueOptions: 'invalidTasksQueueOptions'
373 }
374 )
948faff7 375 ).toThrow(
5c4c2dee
JB
376 new TypeError('Invalid tasks queue options: must be a plain object')
377 )
f0d7f803
JB
378 expect(
379 () =>
380 new FixedThreadPool(
381 numberOfWorkers,
b2fd3f4a 382 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
383 {
384 enableTasksQueue: true,
385 tasksQueueOptions: { concurrency: 0 }
386 }
387 )
948faff7 388 ).toThrow(
e695d66f 389 new RangeError(
20c6f652 390 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
391 )
392 )
f0d7f803
JB
393 expect(
394 () =>
395 new FixedThreadPool(
396 numberOfWorkers,
b2fd3f4a 397 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
398 {
399 enableTasksQueue: true,
5c4c2dee 400 tasksQueueOptions: { concurrency: -1 }
f0d7f803
JB
401 }
402 )
948faff7 403 ).toThrow(
5c4c2dee
JB
404 new RangeError(
405 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
406 )
8735b4e5 407 )
f0d7f803
JB
408 expect(
409 () =>
410 new FixedThreadPool(
411 numberOfWorkers,
b2fd3f4a 412 './tests/worker-files/thread/testWorker.mjs',
f0d7f803
JB
413 {
414 enableTasksQueue: true,
415 tasksQueueOptions: { concurrency: 0.2 }
416 }
417 )
948faff7 418 ).toThrow(
20c6f652 419 new TypeError('Invalid worker node tasks concurrency: must be an integer')
8735b4e5 420 )
5c4c2dee
JB
421 expect(
422 () =>
423 new FixedThreadPool(
424 numberOfWorkers,
b2fd3f4a 425 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
426 {
427 enableTasksQueue: true,
428 tasksQueueOptions: { size: 0 }
429 }
430 )
948faff7 431 ).toThrow(
5c4c2dee
JB
432 new RangeError(
433 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
434 )
435 )
436 expect(
437 () =>
438 new FixedThreadPool(
439 numberOfWorkers,
b2fd3f4a 440 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
441 {
442 enableTasksQueue: true,
443 tasksQueueOptions: { size: -1 }
444 }
445 )
948faff7 446 ).toThrow(
5c4c2dee
JB
447 new RangeError(
448 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
449 )
450 )
451 expect(
452 () =>
453 new FixedThreadPool(
454 numberOfWorkers,
b2fd3f4a 455 './tests/worker-files/thread/testWorker.mjs',
5c4c2dee
JB
456 {
457 enableTasksQueue: true,
458 tasksQueueOptions: { size: 0.2 }
459 }
460 )
948faff7 461 ).toThrow(
5c4c2dee
JB
462 new TypeError('Invalid worker node tasks queue size: must be an integer')
463 )
d4aeae5a
JB
464 })
465
2431bdb4 466 it('Verify that pool worker choice strategy options can be set', async () => {
a20f0ba5
JB
467 const pool = new FixedThreadPool(
468 numberOfWorkers,
b2fd3f4a 469 './tests/worker-files/thread/testWorker.mjs',
a20f0ba5
JB
470 { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
471 )
26ce26ca 472 expect(pool.opts.workerChoiceStrategyOptions).toBeUndefined()
8990357d 473 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
474 retries:
475 pool.info.maxSize +
476 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
932fc8be 477 runTime: { median: false },
5df69fab 478 waitTime: { median: false },
00e1bdeb
JB
479 elu: { median: false },
480 weights: expect.objectContaining({
481 0: expect.any(Number),
482 [pool.info.maxSize - 1]: expect.any(Number)
483 })
a20f0ba5
JB
484 })
485 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
486 .workerChoiceStrategies) {
86cbb766
JB
487 expect(workerChoiceStrategy.opts).toStrictEqual({
488 retries:
489 pool.info.maxSize +
490 Object.keys(workerChoiceStrategy.opts.weights).length,
491 runTime: { median: false },
492 waitTime: { median: false },
493 elu: { median: false },
494 weights: expect.objectContaining({
495 0: expect.any(Number),
496 [pool.info.maxSize - 1]: expect.any(Number)
00e1bdeb 497 })
86cbb766 498 })
a20f0ba5 499 }
87de9ff5
JB
500 expect(
501 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
502 ).toStrictEqual({
932fc8be
JB
503 runTime: {
504 aggregate: true,
505 average: true,
506 median: false
507 },
508 waitTime: {
509 aggregate: false,
510 average: false,
511 median: false
512 },
5df69fab 513 elu: {
9adcefab
JB
514 aggregate: true,
515 average: true,
5df69fab
JB
516 median: false
517 }
86bf340d 518 })
9adcefab
JB
519 pool.setWorkerChoiceStrategyOptions({
520 runTime: { median: true },
521 elu: { median: true }
522 })
a20f0ba5 523 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
9adcefab 524 runTime: { median: true },
8990357d
JB
525 elu: { median: true }
526 })
527 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
528 retries:
529 pool.info.maxSize +
530 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
8990357d
JB
531 runTime: { median: true },
532 waitTime: { median: false },
00e1bdeb
JB
533 elu: { median: true },
534 weights: expect.objectContaining({
535 0: expect.any(Number),
536 [pool.info.maxSize - 1]: expect.any(Number)
537 })
a20f0ba5
JB
538 })
539 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
540 .workerChoiceStrategies) {
86cbb766
JB
541 expect(workerChoiceStrategy.opts).toStrictEqual({
542 retries:
543 pool.info.maxSize +
544 Object.keys(workerChoiceStrategy.opts.weights).length,
545 runTime: { median: true },
546 waitTime: { median: false },
547 elu: { median: true },
548 weights: expect.objectContaining({
549 0: expect.any(Number),
550 [pool.info.maxSize - 1]: expect.any(Number)
00e1bdeb 551 })
86cbb766 552 })
a20f0ba5 553 }
87de9ff5
JB
554 expect(
555 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
556 ).toStrictEqual({
932fc8be
JB
557 runTime: {
558 aggregate: true,
559 average: false,
560 median: true
561 },
562 waitTime: {
563 aggregate: false,
564 average: false,
565 median: false
566 },
5df69fab 567 elu: {
9adcefab 568 aggregate: true,
5df69fab 569 average: false,
9adcefab 570 median: true
5df69fab 571 }
86bf340d 572 })
9adcefab
JB
573 pool.setWorkerChoiceStrategyOptions({
574 runTime: { median: false },
575 elu: { median: false }
576 })
a20f0ba5 577 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d 578 runTime: { median: false },
8990357d
JB
579 elu: { median: false }
580 })
581 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
582 retries:
583 pool.info.maxSize +
584 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
9adcefab 585 runTime: { median: false },
8990357d 586 waitTime: { median: false },
00e1bdeb
JB
587 elu: { median: false },
588 weights: expect.objectContaining({
589 0: expect.any(Number),
590 [pool.info.maxSize - 1]: expect.any(Number)
591 })
a20f0ba5
JB
592 })
593 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
594 .workerChoiceStrategies) {
86cbb766
JB
595 expect(workerChoiceStrategy.opts).toStrictEqual({
596 retries:
597 pool.info.maxSize +
598 Object.keys(workerChoiceStrategy.opts.weights).length,
599 runTime: { median: false },
600 waitTime: { median: false },
601 elu: { median: false },
602 weights: expect.objectContaining({
603 0: expect.any(Number),
604 [pool.info.maxSize - 1]: expect.any(Number)
00e1bdeb 605 })
86cbb766 606 })
a20f0ba5 607 }
87de9ff5
JB
608 expect(
609 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
610 ).toStrictEqual({
932fc8be
JB
611 runTime: {
612 aggregate: true,
613 average: true,
614 median: false
615 },
616 waitTime: {
617 aggregate: false,
618 average: false,
619 median: false
620 },
5df69fab 621 elu: {
9adcefab
JB
622 aggregate: true,
623 average: true,
5df69fab
JB
624 median: false
625 }
86bf340d 626 })
1f95d544
JB
627 expect(() =>
628 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
948faff7 629 ).toThrow(
8735b4e5
JB
630 new TypeError(
631 'Invalid worker choice strategy options: must be a plain object'
632 )
1f95d544 633 )
948faff7 634 expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow(
8735b4e5
JB
635 new Error(
636 'Invalid worker choice strategy options: must have a weight for each worker node'
637 )
1f95d544
JB
638 )
639 expect(() =>
640 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
948faff7 641 ).toThrow(
8735b4e5
JB
642 new Error(
643 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
644 )
1f95d544 645 )
a20f0ba5
JB
646 await pool.destroy()
647 })
648
2431bdb4 649 it('Verify that pool tasks queue can be enabled/disabled', async () => {
a20f0ba5
JB
650 const pool = new FixedThreadPool(
651 numberOfWorkers,
b2fd3f4a 652 './tests/worker-files/thread/testWorker.mjs'
a20f0ba5
JB
653 )
654 expect(pool.opts.enableTasksQueue).toBe(false)
655 expect(pool.opts.tasksQueueOptions).toBeUndefined()
656 pool.enableTasksQueue(true)
657 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
658 expect(pool.opts.tasksQueueOptions).toStrictEqual({
659 concurrency: 1,
2324f8c9 660 size: Math.pow(numberOfWorkers, 2),
dbd73092 661 taskStealing: true,
32b141fd 662 tasksStealingOnBackPressure: true,
568d0075 663 tasksFinishedTimeout: 2000
20c6f652 664 })
a20f0ba5
JB
665 pool.enableTasksQueue(true, { concurrency: 2 })
666 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
667 expect(pool.opts.tasksQueueOptions).toStrictEqual({
668 concurrency: 2,
2324f8c9 669 size: Math.pow(numberOfWorkers, 2),
dbd73092 670 taskStealing: true,
32b141fd 671 tasksStealingOnBackPressure: true,
568d0075 672 tasksFinishedTimeout: 2000
20c6f652 673 })
a20f0ba5
JB
674 pool.enableTasksQueue(false)
675 expect(pool.opts.enableTasksQueue).toBe(false)
676 expect(pool.opts.tasksQueueOptions).toBeUndefined()
677 await pool.destroy()
678 })
679
2431bdb4 680 it('Verify that pool tasks queue options can be set', async () => {
a20f0ba5
JB
681 const pool = new FixedThreadPool(
682 numberOfWorkers,
b2fd3f4a 683 './tests/worker-files/thread/testWorker.mjs',
a20f0ba5
JB
684 { enableTasksQueue: true }
685 )
20c6f652
JB
686 expect(pool.opts.tasksQueueOptions).toStrictEqual({
687 concurrency: 1,
2324f8c9 688 size: Math.pow(numberOfWorkers, 2),
dbd73092 689 taskStealing: true,
32b141fd 690 tasksStealingOnBackPressure: true,
568d0075 691 tasksFinishedTimeout: 2000
20c6f652 692 })
d6ca1416 693 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
694 expect(workerNode.tasksQueueBackPressureSize).toBe(
695 pool.opts.tasksQueueOptions.size
696 )
d6ca1416
JB
697 }
698 pool.setTasksQueueOptions({
699 concurrency: 2,
2324f8c9 700 size: 2,
d6ca1416 701 taskStealing: false,
32b141fd 702 tasksStealingOnBackPressure: false,
568d0075 703 tasksFinishedTimeout: 3000
d6ca1416 704 })
20c6f652
JB
705 expect(pool.opts.tasksQueueOptions).toStrictEqual({
706 concurrency: 2,
2324f8c9 707 size: 2,
d6ca1416 708 taskStealing: false,
32b141fd 709 tasksStealingOnBackPressure: false,
568d0075 710 tasksFinishedTimeout: 3000
d6ca1416
JB
711 })
712 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
713 expect(workerNode.tasksQueueBackPressureSize).toBe(
714 pool.opts.tasksQueueOptions.size
715 )
d6ca1416
JB
716 }
717 pool.setTasksQueueOptions({
718 concurrency: 1,
719 taskStealing: true,
720 tasksStealingOnBackPressure: true
721 })
722 expect(pool.opts.tasksQueueOptions).toStrictEqual({
723 concurrency: 1,
2324f8c9 724 size: Math.pow(numberOfWorkers, 2),
dbd73092 725 taskStealing: true,
32b141fd 726 tasksStealingOnBackPressure: true,
568d0075 727 tasksFinishedTimeout: 2000
20c6f652 728 })
d6ca1416 729 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
730 expect(workerNode.tasksQueueBackPressureSize).toBe(
731 pool.opts.tasksQueueOptions.size
732 )
d6ca1416 733 }
948faff7 734 expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions')).toThrow(
8735b4e5
JB
735 new TypeError('Invalid tasks queue options: must be a plain object')
736 )
948faff7 737 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrow(
e695d66f 738 new RangeError(
20c6f652 739 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
740 )
741 )
948faff7 742 expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrow(
e695d66f 743 new RangeError(
20c6f652 744 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
8735b4e5 745 )
a20f0ba5 746 )
948faff7 747 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrow(
20c6f652
JB
748 new TypeError('Invalid worker node tasks concurrency: must be an integer')
749 )
948faff7 750 expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrow(
20c6f652 751 new RangeError(
68dbcdc0 752 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
20c6f652
JB
753 )
754 )
948faff7 755 expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrow(
20c6f652 756 new RangeError(
68dbcdc0 757 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
20c6f652
JB
758 )
759 )
948faff7 760 expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrow(
68dbcdc0 761 new TypeError('Invalid worker node tasks queue size: must be an integer')
f0d7f803 762 )
a20f0ba5
JB
763 await pool.destroy()
764 })
765
6b27d407
JB
766 it('Verify that pool info is set', async () => {
767 let pool = new FixedThreadPool(
768 numberOfWorkers,
b2fd3f4a 769 './tests/worker-files/thread/testWorker.mjs'
6b27d407 770 )
2dca6cab
JB
771 expect(pool.info).toStrictEqual({
772 version,
773 type: PoolTypes.fixed,
774 worker: WorkerTypes.thread,
47352846 775 started: true,
2dca6cab
JB
776 ready: true,
777 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
778 minSize: numberOfWorkers,
779 maxSize: numberOfWorkers,
780 workerNodes: numberOfWorkers,
781 idleWorkerNodes: numberOfWorkers,
782 busyWorkerNodes: 0,
783 executedTasks: 0,
784 executingTasks: 0,
2dca6cab
JB
785 failedTasks: 0
786 })
6b27d407
JB
787 await pool.destroy()
788 pool = new DynamicClusterPool(
2431bdb4 789 Math.floor(numberOfWorkers / 2),
6b27d407 790 numberOfWorkers,
ecdfbdc0 791 './tests/worker-files/cluster/testWorker.js'
6b27d407 792 )
2dca6cab
JB
793 expect(pool.info).toStrictEqual({
794 version,
795 type: PoolTypes.dynamic,
796 worker: WorkerTypes.cluster,
47352846 797 started: true,
2dca6cab
JB
798 ready: true,
799 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
800 minSize: Math.floor(numberOfWorkers / 2),
801 maxSize: numberOfWorkers,
802 workerNodes: Math.floor(numberOfWorkers / 2),
803 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
804 busyWorkerNodes: 0,
805 executedTasks: 0,
806 executingTasks: 0,
2dca6cab
JB
807 failedTasks: 0
808 })
6b27d407
JB
809 await pool.destroy()
810 })
811
2431bdb4 812 it('Verify that pool worker tasks usage are initialized', async () => {
bf9549ae
JB
813 const pool = new FixedClusterPool(
814 numberOfWorkers,
815 './tests/worker-files/cluster/testWorker.js'
816 )
f06e48d8 817 for (const workerNode of pool.workerNodes) {
47352846 818 expect(workerNode).toBeInstanceOf(WorkerNode)
465b2940 819 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
820 tasks: {
821 executed: 0,
822 executing: 0,
823 queued: 0,
df593701 824 maxQueued: 0,
463226a4 825 sequentiallyStolen: 0,
68cbdc84 826 stolen: 0,
a4e07f72
JB
827 failed: 0
828 },
829 runTime: {
4ba4c7f9 830 history: new CircularArray()
a4e07f72
JB
831 },
832 waitTime: {
4ba4c7f9 833 history: new CircularArray()
a4e07f72 834 },
5df69fab
JB
835 elu: {
836 idle: {
4ba4c7f9 837 history: new CircularArray()
5df69fab
JB
838 },
839 active: {
4ba4c7f9 840 history: new CircularArray()
f7510105 841 }
5df69fab 842 }
86bf340d 843 })
f06e48d8
JB
844 }
845 await pool.destroy()
846 })
847
2431bdb4
JB
848 it('Verify that pool worker tasks queue are initialized', async () => {
849 let pool = new FixedClusterPool(
f06e48d8
JB
850 numberOfWorkers,
851 './tests/worker-files/cluster/testWorker.js'
852 )
853 for (const workerNode of pool.workerNodes) {
47352846 854 expect(workerNode).toBeInstanceOf(WorkerNode)
574b351d 855 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
4d8bf9e4 856 expect(workerNode.tasksQueue.size).toBe(0)
9c16fb4b 857 expect(workerNode.tasksQueue.maxSize).toBe(0)
bf9549ae 858 }
fd7ebd49 859 await pool.destroy()
2431bdb4
JB
860 pool = new DynamicThreadPool(
861 Math.floor(numberOfWorkers / 2),
862 numberOfWorkers,
b2fd3f4a 863 './tests/worker-files/thread/testWorker.mjs'
2431bdb4
JB
864 )
865 for (const workerNode of pool.workerNodes) {
47352846 866 expect(workerNode).toBeInstanceOf(WorkerNode)
574b351d 867 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
2431bdb4
JB
868 expect(workerNode.tasksQueue.size).toBe(0)
869 expect(workerNode.tasksQueue.maxSize).toBe(0)
870 }
213cbac6 871 await pool.destroy()
2431bdb4
JB
872 })
873
874 it('Verify that pool worker info are initialized', async () => {
875 let pool = new FixedClusterPool(
876 numberOfWorkers,
877 './tests/worker-files/cluster/testWorker.js'
878 )
2dca6cab 879 for (const workerNode of pool.workerNodes) {
47352846 880 expect(workerNode).toBeInstanceOf(WorkerNode)
2dca6cab
JB
881 expect(workerNode.info).toStrictEqual({
882 id: expect.any(Number),
883 type: WorkerTypes.cluster,
884 dynamic: false,
5eb72b9e
JB
885 ready: true,
886 stealing: false
2dca6cab
JB
887 })
888 }
2431bdb4
JB
889 await pool.destroy()
890 pool = new DynamicThreadPool(
891 Math.floor(numberOfWorkers / 2),
892 numberOfWorkers,
b2fd3f4a 893 './tests/worker-files/thread/testWorker.mjs'
2431bdb4 894 )
2dca6cab 895 for (const workerNode of pool.workerNodes) {
47352846 896 expect(workerNode).toBeInstanceOf(WorkerNode)
2dca6cab
JB
897 expect(workerNode.info).toStrictEqual({
898 id: expect.any(Number),
899 type: WorkerTypes.thread,
900 dynamic: false,
5eb72b9e
JB
901 ready: true,
902 stealing: false
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),
5eb72b9e 1274 stealingWorkerNodes: expect.any(Number),
d46660cd 1275 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1276 executedTasks: expect.any(Number),
1277 executingTasks: expect.any(Number),
3e8611a8
JB
1278 maxQueuedTasks: expect.any(Number),
1279 queuedTasks: expect.any(Number),
1280 backPressure: true,
68cbdc84 1281 stolenTasks: expect.any(Number),
a4e07f72 1282 failedTasks: expect.any(Number)
d46660cd 1283 })
5eb72b9e 1284 expect(pool.hasBackPressure.callCount).toBeGreaterThanOrEqual(7)
fd7ebd49 1285 await pool.destroy()
7c0ba920 1286 })
70a4f5ea 1287
85b2561d
JB
1288 it('Verify that destroy() waits for queued tasks to finish', async () => {
1289 const tasksFinishedTimeout = 2500
1290 const pool = new FixedThreadPool(
1291 numberOfWorkers,
1292 './tests/worker-files/thread/asyncWorker.mjs',
1293 {
1294 enableTasksQueue: true,
1295 tasksQueueOptions: { tasksFinishedTimeout }
1296 }
1297 )
1298 const maxMultiplier = 4
1299 let tasksFinished = 0
1300 for (const workerNode of pool.workerNodes) {
1301 workerNode.on('taskFinished', () => {
1302 ++tasksFinished
1303 })
1304 }
1305 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1306 pool.execute()
1307 }
1308 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1309 const startTime = performance.now()
1310 await pool.destroy()
1311 const elapsedTime = performance.now() - startTime
90afa746 1312 expect(tasksFinished).toBeLessThanOrEqual(numberOfWorkers * maxMultiplier)
85b2561d 1313 expect(elapsedTime).toBeGreaterThanOrEqual(2000)
ecab2e2a 1314 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 400)
85b2561d
JB
1315 })
1316
1317 it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
1318 const tasksFinishedTimeout = 1000
1319 const pool = new FixedThreadPool(
1320 numberOfWorkers,
1321 './tests/worker-files/thread/asyncWorker.mjs',
1322 {
1323 enableTasksQueue: true,
1324 tasksQueueOptions: { tasksFinishedTimeout }
1325 }
1326 )
1327 const maxMultiplier = 4
1328 let tasksFinished = 0
1329 for (const workerNode of pool.workerNodes) {
1330 workerNode.on('taskFinished', () => {
1331 ++tasksFinished
1332 })
1333 }
1334 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1335 pool.execute()
1336 }
1337 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1338 const startTime = performance.now()
1339 await pool.destroy()
1340 const elapsedTime = performance.now() - startTime
1341 expect(tasksFinished).toBe(0)
2885534c 1342 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 800)
85b2561d
JB
1343 })
1344
f18fd12b
JB
1345 it('Verify that pool asynchronous resource track tasks execution', async () => {
1346 let taskAsyncId
1347 let initCalls = 0
1348 let beforeCalls = 0
1349 let afterCalls = 0
1350 let resolveCalls = 0
1351 const hook = createHook({
1352 init (asyncId, type) {
1353 if (type === 'poolifier:task') {
1354 initCalls++
1355 taskAsyncId = asyncId
1356 }
1357 },
1358 before (asyncId) {
1359 if (asyncId === taskAsyncId) beforeCalls++
1360 },
1361 after (asyncId) {
1362 if (asyncId === taskAsyncId) afterCalls++
1363 },
1364 promiseResolve () {
1365 if (executionAsyncId() === taskAsyncId) resolveCalls++
1366 }
1367 })
f18fd12b
JB
1368 const pool = new FixedThreadPool(
1369 numberOfWorkers,
1370 './tests/worker-files/thread/testWorker.mjs'
1371 )
8954c0a3 1372 hook.enable()
f18fd12b
JB
1373 await pool.execute()
1374 hook.disable()
1375 expect(initCalls).toBe(1)
1376 expect(beforeCalls).toBe(1)
1377 expect(afterCalls).toBe(1)
1378 expect(resolveCalls).toBe(1)
8954c0a3 1379 await pool.destroy()
f18fd12b
JB
1380 })
1381
9eae3c69
JB
1382 it('Verify that hasTaskFunction() is working', async () => {
1383 const dynamicThreadPool = new DynamicThreadPool(
1384 Math.floor(numberOfWorkers / 2),
1385 numberOfWorkers,
b2fd3f4a 1386 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
9eae3c69
JB
1387 )
1388 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1389 expect(dynamicThreadPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1390 expect(dynamicThreadPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1391 true
1392 )
1393 expect(dynamicThreadPool.hasTaskFunction('factorial')).toBe(true)
1394 expect(dynamicThreadPool.hasTaskFunction('fibonacci')).toBe(true)
1395 expect(dynamicThreadPool.hasTaskFunction('unknown')).toBe(false)
1396 await dynamicThreadPool.destroy()
1397 const fixedClusterPool = new FixedClusterPool(
1398 numberOfWorkers,
1399 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1400 )
1401 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1402 expect(fixedClusterPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1403 expect(fixedClusterPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1404 true
1405 )
1406 expect(fixedClusterPool.hasTaskFunction('factorial')).toBe(true)
1407 expect(fixedClusterPool.hasTaskFunction('fibonacci')).toBe(true)
1408 expect(fixedClusterPool.hasTaskFunction('unknown')).toBe(false)
1409 await fixedClusterPool.destroy()
1410 })
1411
1412 it('Verify that addTaskFunction() is working', async () => {
1413 const dynamicThreadPool = new DynamicThreadPool(
1414 Math.floor(numberOfWorkers / 2),
1415 numberOfWorkers,
b2fd3f4a 1416 './tests/worker-files/thread/testWorker.mjs'
9eae3c69
JB
1417 )
1418 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
3feeab69
JB
1419 await expect(
1420 dynamicThreadPool.addTaskFunction(0, () => {})
948faff7 1421 ).rejects.toThrow(new TypeError('name argument must be a string'))
3feeab69
JB
1422 await expect(
1423 dynamicThreadPool.addTaskFunction('', () => {})
948faff7 1424 ).rejects.toThrow(
3feeab69
JB
1425 new TypeError('name argument must not be an empty string')
1426 )
948faff7
JB
1427 await expect(dynamicThreadPool.addTaskFunction('test', 0)).rejects.toThrow(
1428 new TypeError('fn argument must be a function')
1429 )
1430 await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow(
1431 new TypeError('fn argument must be a function')
1432 )
9eae3c69
JB
1433 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1434 DEFAULT_TASK_NAME,
1435 'test'
1436 ])
1437 const echoTaskFunction = data => {
1438 return data
1439 }
1440 await expect(
1441 dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1442 ).resolves.toBe(true)
1443 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1444 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1445 echoTaskFunction
1446 )
1447 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1448 DEFAULT_TASK_NAME,
1449 'test',
1450 'echo'
1451 ])
1452 const taskFunctionData = { test: 'test' }
1453 const echoResult = await dynamicThreadPool.execute(taskFunctionData, 'echo')
1454 expect(echoResult).toStrictEqual(taskFunctionData)
adee6053
JB
1455 for (const workerNode of dynamicThreadPool.workerNodes) {
1456 expect(workerNode.getTaskFunctionWorkerUsage('echo')).toStrictEqual({
1457 tasks: {
1458 executed: expect.any(Number),
1459 executing: 0,
1460 queued: 0,
463226a4 1461 sequentiallyStolen: 0,
5ad42e34 1462 stolen: 0,
adee6053
JB
1463 failed: 0
1464 },
1465 runTime: {
1466 history: new CircularArray()
1467 },
1468 waitTime: {
1469 history: new CircularArray()
1470 },
1471 elu: {
1472 idle: {
1473 history: new CircularArray()
1474 },
1475 active: {
1476 history: new CircularArray()
1477 }
1478 }
1479 })
1480 }
9eae3c69
JB
1481 await dynamicThreadPool.destroy()
1482 })
1483
1484 it('Verify that removeTaskFunction() is working', async () => {
1485 const dynamicThreadPool = new DynamicThreadPool(
1486 Math.floor(numberOfWorkers / 2),
1487 numberOfWorkers,
b2fd3f4a 1488 './tests/worker-files/thread/testWorker.mjs'
9eae3c69
JB
1489 )
1490 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1491 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1492 DEFAULT_TASK_NAME,
1493 'test'
1494 ])
948faff7 1495 await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow(
16248b23 1496 new Error('Cannot remove a task function not handled on the pool side')
9eae3c69
JB
1497 )
1498 const echoTaskFunction = data => {
1499 return data
1500 }
1501 await dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1502 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1503 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1504 echoTaskFunction
1505 )
1506 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1507 DEFAULT_TASK_NAME,
1508 'test',
1509 'echo'
1510 ])
1511 await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe(
1512 true
1513 )
1514 expect(dynamicThreadPool.taskFunctions.size).toBe(0)
1515 expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined()
1516 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1517 DEFAULT_TASK_NAME,
1518 'test'
1519 ])
1520 await dynamicThreadPool.destroy()
1521 })
1522
30500265 1523 it('Verify that listTaskFunctionNames() is working', async () => {
90d7d101
JB
1524 const dynamicThreadPool = new DynamicThreadPool(
1525 Math.floor(numberOfWorkers / 2),
1526 numberOfWorkers,
b2fd3f4a 1527 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
90d7d101
JB
1528 )
1529 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
66979634 1530 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
6cd5248f 1531 DEFAULT_TASK_NAME,
90d7d101
JB
1532 'jsonIntegerSerialization',
1533 'factorial',
1534 'fibonacci'
1535 ])
9eae3c69 1536 await dynamicThreadPool.destroy()
90d7d101
JB
1537 const fixedClusterPool = new FixedClusterPool(
1538 numberOfWorkers,
1539 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1540 )
1541 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
66979634 1542 expect(fixedClusterPool.listTaskFunctionNames()).toStrictEqual([
6cd5248f 1543 DEFAULT_TASK_NAME,
90d7d101
JB
1544 'jsonIntegerSerialization',
1545 'factorial',
1546 'fibonacci'
1547 ])
0fe39c97 1548 await fixedClusterPool.destroy()
90d7d101
JB
1549 })
1550
9eae3c69 1551 it('Verify that setDefaultTaskFunction() is working', async () => {
30500265
JB
1552 const dynamicThreadPool = new DynamicThreadPool(
1553 Math.floor(numberOfWorkers / 2),
1554 numberOfWorkers,
b2fd3f4a 1555 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
30500265
JB
1556 )
1557 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
711623b8 1558 const workerId = dynamicThreadPool.workerNodes[0].info.id
948faff7 1559 await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow(
b0b55f57 1560 new Error(
711623b8 1561 `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'`
b0b55f57
JB
1562 )
1563 )
1564 await expect(
1565 dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME)
948faff7 1566 ).rejects.toThrow(
b0b55f57 1567 new Error(
711623b8 1568 `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
1569 )
1570 )
1571 await expect(
1572 dynamicThreadPool.setDefaultTaskFunction('unknown')
948faff7 1573 ).rejects.toThrow(
b0b55f57 1574 new Error(
711623b8 1575 `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
1576 )
1577 )
9eae3c69
JB
1578 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1579 DEFAULT_TASK_NAME,
1580 'jsonIntegerSerialization',
1581 'factorial',
1582 'fibonacci'
1583 ])
1584 await expect(
1585 dynamicThreadPool.setDefaultTaskFunction('factorial')
1586 ).resolves.toBe(true)
1587 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1588 DEFAULT_TASK_NAME,
1589 'factorial',
1590 'jsonIntegerSerialization',
1591 'fibonacci'
1592 ])
1593 await expect(
1594 dynamicThreadPool.setDefaultTaskFunction('fibonacci')
1595 ).resolves.toBe(true)
1596 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1597 DEFAULT_TASK_NAME,
1598 'fibonacci',
1599 'jsonIntegerSerialization',
1600 'factorial'
1601 ])
cda9ba34 1602 await dynamicThreadPool.destroy()
30500265
JB
1603 })
1604
90d7d101 1605 it('Verify that multiple task functions worker is working', async () => {
70a4f5ea 1606 const pool = new DynamicClusterPool(
2431bdb4 1607 Math.floor(numberOfWorkers / 2),
70a4f5ea 1608 numberOfWorkers,
90d7d101 1609 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
70a4f5ea
JB
1610 )
1611 const data = { n: 10 }
82888165 1612 const result0 = await pool.execute(data)
30b963d4 1613 expect(result0).toStrictEqual({ ok: 1 })
70a4f5ea 1614 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
30b963d4 1615 expect(result1).toStrictEqual({ ok: 1 })
70a4f5ea
JB
1616 const result2 = await pool.execute(data, 'factorial')
1617 expect(result2).toBe(3628800)
1618 const result3 = await pool.execute(data, 'fibonacci')
024daf59 1619 expect(result3).toBe(55)
5bb5be17
JB
1620 expect(pool.info.executingTasks).toBe(0)
1621 expect(pool.info.executedTasks).toBe(4)
b414b84c 1622 for (const workerNode of pool.workerNodes) {
66979634 1623 expect(workerNode.info.taskFunctionNames).toStrictEqual([
6cd5248f 1624 DEFAULT_TASK_NAME,
b414b84c
JB
1625 'jsonIntegerSerialization',
1626 'factorial',
1627 'fibonacci'
1628 ])
1629 expect(workerNode.taskFunctionsUsage.size).toBe(3)
66979634 1630 for (const name of pool.listTaskFunctionNames()) {
5bb5be17
JB
1631 expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1632 tasks: {
1633 executed: expect.any(Number),
4ba4c7f9 1634 executing: 0,
5bb5be17 1635 failed: 0,
68cbdc84 1636 queued: 0,
463226a4 1637 sequentiallyStolen: 0,
68cbdc84 1638 stolen: 0
5bb5be17
JB
1639 },
1640 runTime: {
1641 history: expect.any(CircularArray)
1642 },
1643 waitTime: {
1644 history: expect.any(CircularArray)
1645 },
1646 elu: {
1647 idle: {
1648 history: expect.any(CircularArray)
1649 },
1650 active: {
1651 history: expect.any(CircularArray)
1652 }
1653 }
1654 })
1655 expect(
4ba4c7f9
JB
1656 workerNode.getTaskFunctionWorkerUsage(name).tasks.executed
1657 ).toBeGreaterThan(0)
5bb5be17 1658 }
dfd7ec01
JB
1659 expect(
1660 workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1661 ).toStrictEqual(
66979634
JB
1662 workerNode.getTaskFunctionWorkerUsage(
1663 workerNode.info.taskFunctionNames[1]
1664 )
dfd7ec01 1665 )
5bb5be17 1666 }
0fe39c97 1667 await pool.destroy()
70a4f5ea 1668 })
52a23942
JB
1669
1670 it('Verify sendKillMessageToWorker()', async () => {
1671 const pool = new DynamicClusterPool(
1672 Math.floor(numberOfWorkers / 2),
1673 numberOfWorkers,
1674 './tests/worker-files/cluster/testWorker.js'
1675 )
1676 const workerNodeKey = 0
1677 await expect(
adee6053 1678 pool.sendKillMessageToWorker(workerNodeKey)
52a23942
JB
1679 ).resolves.toBeUndefined()
1680 await pool.destroy()
1681 })
adee6053
JB
1682
1683 it('Verify sendTaskFunctionOperationToWorker()', async () => {
1684 const pool = new DynamicClusterPool(
1685 Math.floor(numberOfWorkers / 2),
1686 numberOfWorkers,
1687 './tests/worker-files/cluster/testWorker.js'
1688 )
1689 const workerNodeKey = 0
1690 await expect(
1691 pool.sendTaskFunctionOperationToWorker(workerNodeKey, {
1692 taskFunctionOperation: 'add',
1693 taskFunctionName: 'empty',
1694 taskFunction: (() => {}).toString()
1695 })
1696 ).resolves.toBe(true)
1697 expect(
1698 pool.workerNodes[workerNodeKey].info.taskFunctionNames
1699 ).toStrictEqual([DEFAULT_TASK_NAME, 'test', 'empty'])
1700 await pool.destroy()
1701 })
1702
1703 it('Verify sendTaskFunctionOperationToWorkers()', async () => {
1704 const pool = new DynamicClusterPool(
1705 Math.floor(numberOfWorkers / 2),
1706 numberOfWorkers,
1707 './tests/worker-files/cluster/testWorker.js'
1708 )
adee6053
JB
1709 await expect(
1710 pool.sendTaskFunctionOperationToWorkers({
1711 taskFunctionOperation: 'add',
1712 taskFunctionName: 'empty',
1713 taskFunction: (() => {}).toString()
1714 })
1715 ).resolves.toBe(true)
1716 for (const workerNode of pool.workerNodes) {
1717 expect(workerNode.info.taskFunctionNames).toStrictEqual([
1718 DEFAULT_TASK_NAME,
1719 'test',
1720 'empty'
1721 ])
1722 }
1723 await pool.destroy()
1724 })
3ec964d6 1725})