fix: put back properties init in constructor
[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 },
3d6f0f73
JB
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) {
55ddaf0f
JB
489 expect(workerChoiceStrategy.opts).toStrictEqual(
490 expect.objectContaining({
491 retries:
492 pool.info.maxSize +
493 Object.keys(workerChoiceStrategy.opts.weights).length,
494 runTime: { median: false },
495 waitTime: { median: false },
496 elu: { median: false }
00e1bdeb 497 })
55ddaf0f 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) {
fc7633dd
JB
541 expect(workerChoiceStrategy.opts).toStrictEqual(
542 expect.objectContaining({
543 retries:
544 pool.info.maxSize +
545 Object.keys(workerChoiceStrategy.opts.weights).length,
546 runTime: { median: true },
547 waitTime: { median: false },
548 elu: { median: true }
00e1bdeb 549 })
fc7633dd 550 )
a20f0ba5 551 }
87de9ff5
JB
552 expect(
553 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
554 ).toStrictEqual({
932fc8be
JB
555 runTime: {
556 aggregate: true,
557 average: false,
558 median: true
559 },
560 waitTime: {
561 aggregate: false,
562 average: false,
563 median: false
564 },
5df69fab 565 elu: {
9adcefab 566 aggregate: true,
5df69fab 567 average: false,
9adcefab 568 median: true
5df69fab 569 }
86bf340d 570 })
9adcefab
JB
571 pool.setWorkerChoiceStrategyOptions({
572 runTime: { median: false },
573 elu: { median: false }
574 })
a20f0ba5 575 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
8990357d 576 runTime: { median: false },
8990357d
JB
577 elu: { median: false }
578 })
579 expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
00e1bdeb
JB
580 retries:
581 pool.info.maxSize +
582 Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
9adcefab 583 runTime: { median: false },
8990357d 584 waitTime: { median: false },
00e1bdeb
JB
585 elu: { median: false },
586 weights: expect.objectContaining({
587 0: expect.any(Number),
588 [pool.info.maxSize - 1]: expect.any(Number)
589 })
a20f0ba5
JB
590 })
591 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
592 .workerChoiceStrategies) {
69bba5e2
JB
593 expect(workerChoiceStrategy.opts).toStrictEqual(
594 expect.objectContaining({
595 retries:
596 pool.info.maxSize +
597 Object.keys(workerChoiceStrategy.opts.weights).length,
598 runTime: { median: false },
599 waitTime: { median: false },
600 elu: { median: false }
00e1bdeb 601 })
69bba5e2 602 )
a20f0ba5 603 }
87de9ff5
JB
604 expect(
605 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
606 ).toStrictEqual({
932fc8be
JB
607 runTime: {
608 aggregate: true,
609 average: true,
610 median: false
611 },
612 waitTime: {
613 aggregate: false,
614 average: false,
615 median: false
616 },
5df69fab 617 elu: {
9adcefab
JB
618 aggregate: true,
619 average: true,
5df69fab
JB
620 median: false
621 }
86bf340d 622 })
1f95d544
JB
623 expect(() =>
624 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
948faff7 625 ).toThrow(
8735b4e5
JB
626 new TypeError(
627 'Invalid worker choice strategy options: must be a plain object'
628 )
1f95d544 629 )
948faff7 630 expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow(
8735b4e5
JB
631 new Error(
632 'Invalid worker choice strategy options: must have a weight for each worker node'
633 )
1f95d544
JB
634 )
635 expect(() =>
636 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
948faff7 637 ).toThrow(
8735b4e5
JB
638 new Error(
639 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
640 )
1f95d544 641 )
a20f0ba5
JB
642 await pool.destroy()
643 })
644
2431bdb4 645 it('Verify that pool tasks queue can be enabled/disabled', async () => {
a20f0ba5
JB
646 const pool = new FixedThreadPool(
647 numberOfWorkers,
b2fd3f4a 648 './tests/worker-files/thread/testWorker.mjs'
a20f0ba5
JB
649 )
650 expect(pool.opts.enableTasksQueue).toBe(false)
651 expect(pool.opts.tasksQueueOptions).toBeUndefined()
652 pool.enableTasksQueue(true)
653 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
654 expect(pool.opts.tasksQueueOptions).toStrictEqual({
655 concurrency: 1,
2324f8c9 656 size: Math.pow(numberOfWorkers, 2),
dbd73092 657 taskStealing: true,
32b141fd 658 tasksStealingOnBackPressure: true,
568d0075 659 tasksFinishedTimeout: 2000
20c6f652 660 })
a20f0ba5
JB
661 pool.enableTasksQueue(true, { concurrency: 2 })
662 expect(pool.opts.enableTasksQueue).toBe(true)
20c6f652
JB
663 expect(pool.opts.tasksQueueOptions).toStrictEqual({
664 concurrency: 2,
2324f8c9 665 size: Math.pow(numberOfWorkers, 2),
dbd73092 666 taskStealing: true,
32b141fd 667 tasksStealingOnBackPressure: true,
568d0075 668 tasksFinishedTimeout: 2000
20c6f652 669 })
a20f0ba5
JB
670 pool.enableTasksQueue(false)
671 expect(pool.opts.enableTasksQueue).toBe(false)
672 expect(pool.opts.tasksQueueOptions).toBeUndefined()
673 await pool.destroy()
674 })
675
2431bdb4 676 it('Verify that pool tasks queue options can be set', async () => {
a20f0ba5
JB
677 const pool = new FixedThreadPool(
678 numberOfWorkers,
b2fd3f4a 679 './tests/worker-files/thread/testWorker.mjs',
a20f0ba5
JB
680 { enableTasksQueue: true }
681 )
20c6f652
JB
682 expect(pool.opts.tasksQueueOptions).toStrictEqual({
683 concurrency: 1,
2324f8c9 684 size: Math.pow(numberOfWorkers, 2),
dbd73092 685 taskStealing: true,
32b141fd 686 tasksStealingOnBackPressure: true,
568d0075 687 tasksFinishedTimeout: 2000
20c6f652 688 })
d6ca1416 689 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
690 expect(workerNode.tasksQueueBackPressureSize).toBe(
691 pool.opts.tasksQueueOptions.size
692 )
d6ca1416
JB
693 }
694 pool.setTasksQueueOptions({
695 concurrency: 2,
2324f8c9 696 size: 2,
d6ca1416 697 taskStealing: false,
32b141fd 698 tasksStealingOnBackPressure: false,
568d0075 699 tasksFinishedTimeout: 3000
d6ca1416 700 })
20c6f652
JB
701 expect(pool.opts.tasksQueueOptions).toStrictEqual({
702 concurrency: 2,
2324f8c9 703 size: 2,
d6ca1416 704 taskStealing: false,
32b141fd 705 tasksStealingOnBackPressure: false,
568d0075 706 tasksFinishedTimeout: 3000
d6ca1416
JB
707 })
708 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
709 expect(workerNode.tasksQueueBackPressureSize).toBe(
710 pool.opts.tasksQueueOptions.size
711 )
d6ca1416
JB
712 }
713 pool.setTasksQueueOptions({
714 concurrency: 1,
715 taskStealing: true,
716 tasksStealingOnBackPressure: true
717 })
718 expect(pool.opts.tasksQueueOptions).toStrictEqual({
719 concurrency: 1,
2324f8c9 720 size: Math.pow(numberOfWorkers, 2),
dbd73092 721 taskStealing: true,
32b141fd 722 tasksStealingOnBackPressure: true,
568d0075 723 tasksFinishedTimeout: 2000
20c6f652 724 })
d6ca1416 725 for (const workerNode of pool.workerNodes) {
2324f8c9
JB
726 expect(workerNode.tasksQueueBackPressureSize).toBe(
727 pool.opts.tasksQueueOptions.size
728 )
d6ca1416 729 }
948faff7 730 expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions')).toThrow(
8735b4e5
JB
731 new TypeError('Invalid tasks queue options: must be a plain object')
732 )
948faff7 733 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrow(
e695d66f 734 new RangeError(
20c6f652 735 'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
8735b4e5
JB
736 )
737 )
948faff7 738 expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrow(
e695d66f 739 new RangeError(
20c6f652 740 'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
8735b4e5 741 )
a20f0ba5 742 )
948faff7 743 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrow(
20c6f652
JB
744 new TypeError('Invalid worker node tasks concurrency: must be an integer')
745 )
948faff7 746 expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrow(
20c6f652 747 new RangeError(
68dbcdc0 748 'Invalid worker node tasks queue size: 0 is a negative integer or zero'
20c6f652
JB
749 )
750 )
948faff7 751 expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrow(
20c6f652 752 new RangeError(
68dbcdc0 753 'Invalid worker node tasks queue size: -1 is a negative integer or zero'
20c6f652
JB
754 )
755 )
948faff7 756 expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrow(
68dbcdc0 757 new TypeError('Invalid worker node tasks queue size: must be an integer')
f0d7f803 758 )
a20f0ba5
JB
759 await pool.destroy()
760 })
761
6b27d407
JB
762 it('Verify that pool info is set', async () => {
763 let pool = new FixedThreadPool(
764 numberOfWorkers,
b2fd3f4a 765 './tests/worker-files/thread/testWorker.mjs'
6b27d407 766 )
2dca6cab
JB
767 expect(pool.info).toStrictEqual({
768 version,
769 type: PoolTypes.fixed,
770 worker: WorkerTypes.thread,
47352846 771 started: true,
2dca6cab
JB
772 ready: true,
773 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
774 minSize: numberOfWorkers,
775 maxSize: numberOfWorkers,
776 workerNodes: numberOfWorkers,
777 idleWorkerNodes: numberOfWorkers,
778 busyWorkerNodes: 0,
779 executedTasks: 0,
780 executingTasks: 0,
2dca6cab
JB
781 failedTasks: 0
782 })
6b27d407
JB
783 await pool.destroy()
784 pool = new DynamicClusterPool(
2431bdb4 785 Math.floor(numberOfWorkers / 2),
6b27d407 786 numberOfWorkers,
ecdfbdc0 787 './tests/worker-files/cluster/testWorker.js'
6b27d407 788 )
2dca6cab
JB
789 expect(pool.info).toStrictEqual({
790 version,
791 type: PoolTypes.dynamic,
792 worker: WorkerTypes.cluster,
47352846 793 started: true,
2dca6cab
JB
794 ready: true,
795 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
796 minSize: Math.floor(numberOfWorkers / 2),
797 maxSize: numberOfWorkers,
798 workerNodes: Math.floor(numberOfWorkers / 2),
799 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
800 busyWorkerNodes: 0,
801 executedTasks: 0,
802 executingTasks: 0,
2dca6cab
JB
803 failedTasks: 0
804 })
6b27d407
JB
805 await pool.destroy()
806 })
807
2431bdb4 808 it('Verify that pool worker tasks usage are initialized', async () => {
bf9549ae
JB
809 const pool = new FixedClusterPool(
810 numberOfWorkers,
811 './tests/worker-files/cluster/testWorker.js'
812 )
f06e48d8 813 for (const workerNode of pool.workerNodes) {
47352846 814 expect(workerNode).toBeInstanceOf(WorkerNode)
465b2940 815 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
816 tasks: {
817 executed: 0,
818 executing: 0,
819 queued: 0,
df593701 820 maxQueued: 0,
463226a4 821 sequentiallyStolen: 0,
68cbdc84 822 stolen: 0,
a4e07f72
JB
823 failed: 0
824 },
825 runTime: {
4ba4c7f9 826 history: new CircularArray()
a4e07f72
JB
827 },
828 waitTime: {
4ba4c7f9 829 history: new CircularArray()
a4e07f72 830 },
5df69fab
JB
831 elu: {
832 idle: {
4ba4c7f9 833 history: new CircularArray()
5df69fab
JB
834 },
835 active: {
4ba4c7f9 836 history: new CircularArray()
f7510105 837 }
5df69fab 838 }
86bf340d 839 })
f06e48d8
JB
840 }
841 await pool.destroy()
842 })
843
2431bdb4
JB
844 it('Verify that pool worker tasks queue are initialized', async () => {
845 let pool = new FixedClusterPool(
f06e48d8
JB
846 numberOfWorkers,
847 './tests/worker-files/cluster/testWorker.js'
848 )
849 for (const workerNode of pool.workerNodes) {
47352846 850 expect(workerNode).toBeInstanceOf(WorkerNode)
574b351d 851 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
4d8bf9e4 852 expect(workerNode.tasksQueue.size).toBe(0)
9c16fb4b 853 expect(workerNode.tasksQueue.maxSize).toBe(0)
bf9549ae 854 }
fd7ebd49 855 await pool.destroy()
2431bdb4
JB
856 pool = new DynamicThreadPool(
857 Math.floor(numberOfWorkers / 2),
858 numberOfWorkers,
b2fd3f4a 859 './tests/worker-files/thread/testWorker.mjs'
2431bdb4
JB
860 )
861 for (const workerNode of pool.workerNodes) {
47352846 862 expect(workerNode).toBeInstanceOf(WorkerNode)
574b351d 863 expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
2431bdb4
JB
864 expect(workerNode.tasksQueue.size).toBe(0)
865 expect(workerNode.tasksQueue.maxSize).toBe(0)
866 }
213cbac6 867 await pool.destroy()
2431bdb4
JB
868 })
869
870 it('Verify that pool worker info are initialized', async () => {
871 let pool = new FixedClusterPool(
872 numberOfWorkers,
873 './tests/worker-files/cluster/testWorker.js'
874 )
2dca6cab 875 for (const workerNode of pool.workerNodes) {
47352846 876 expect(workerNode).toBeInstanceOf(WorkerNode)
2dca6cab
JB
877 expect(workerNode.info).toStrictEqual({
878 id: expect.any(Number),
879 type: WorkerTypes.cluster,
880 dynamic: false,
881 ready: true
882 })
883 }
2431bdb4
JB
884 await pool.destroy()
885 pool = new DynamicThreadPool(
886 Math.floor(numberOfWorkers / 2),
887 numberOfWorkers,
b2fd3f4a 888 './tests/worker-files/thread/testWorker.mjs'
2431bdb4 889 )
2dca6cab 890 for (const workerNode of pool.workerNodes) {
47352846 891 expect(workerNode).toBeInstanceOf(WorkerNode)
2dca6cab
JB
892 expect(workerNode.info).toStrictEqual({
893 id: expect.any(Number),
894 type: WorkerTypes.thread,
895 dynamic: false,
7884d183 896 ready: true
2dca6cab
JB
897 })
898 }
213cbac6 899 await pool.destroy()
bf9549ae
JB
900 })
901
711623b8
JB
902 it('Verify that pool statuses are checked at start or destroy', async () => {
903 const pool = new FixedThreadPool(
904 numberOfWorkers,
905 './tests/worker-files/thread/testWorker.mjs'
906 )
907 expect(pool.info.started).toBe(true)
908 expect(pool.info.ready).toBe(true)
909 expect(() => pool.start()).toThrow(
910 new Error('Cannot start an already started pool')
911 )
912 await pool.destroy()
913 expect(pool.info.started).toBe(false)
914 expect(pool.info.ready).toBe(false)
915 await expect(pool.destroy()).rejects.toThrow(
916 new Error('Cannot destroy an already destroyed pool')
917 )
918 })
919
47352846
JB
920 it('Verify that pool can be started after initialization', async () => {
921 const pool = new FixedClusterPool(
922 numberOfWorkers,
923 './tests/worker-files/cluster/testWorker.js',
924 {
925 startWorkers: false
926 }
927 )
928 expect(pool.info.started).toBe(false)
929 expect(pool.info.ready).toBe(false)
55082af9 930 expect(pool.readyEventEmitted).toBe(false)
47352846 931 expect(pool.workerNodes).toStrictEqual([])
948faff7 932 await expect(pool.execute()).rejects.toThrow(
47352846
JB
933 new Error('Cannot execute a task on not started pool')
934 )
935 pool.start()
936 expect(pool.info.started).toBe(true)
937 expect(pool.info.ready).toBe(true)
55082af9
JB
938 await waitPoolEvents(pool, PoolEvents.ready, 1)
939 expect(pool.readyEventEmitted).toBe(true)
47352846
JB
940 expect(pool.workerNodes.length).toBe(numberOfWorkers)
941 for (const workerNode of pool.workerNodes) {
942 expect(workerNode).toBeInstanceOf(WorkerNode)
943 }
944 await pool.destroy()
945 })
946
9d2d0da1
JB
947 it('Verify that pool execute() arguments are checked', async () => {
948 const pool = new FixedClusterPool(
949 numberOfWorkers,
950 './tests/worker-files/cluster/testWorker.js'
951 )
948faff7 952 await expect(pool.execute(undefined, 0)).rejects.toThrow(
9d2d0da1
JB
953 new TypeError('name argument must be a string')
954 )
948faff7 955 await expect(pool.execute(undefined, '')).rejects.toThrow(
9d2d0da1
JB
956 new TypeError('name argument must not be an empty string')
957 )
948faff7 958 await expect(pool.execute(undefined, undefined, {})).rejects.toThrow(
9d2d0da1
JB
959 new TypeError('transferList argument must be an array')
960 )
961 await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
962 "Task function 'unknown' not found"
963 )
964 await pool.destroy()
948faff7 965 await expect(pool.execute()).rejects.toThrow(
47352846 966 new Error('Cannot execute a task on not started pool')
9d2d0da1
JB
967 )
968 })
969
2431bdb4 970 it('Verify that pool worker tasks usage are computed', async () => {
bf9549ae
JB
971 const pool = new FixedClusterPool(
972 numberOfWorkers,
973 './tests/worker-files/cluster/testWorker.js'
974 )
09c2d0d3 975 const promises = new Set()
fc027381
JB
976 const maxMultiplier = 2
977 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 978 promises.add(pool.execute())
bf9549ae 979 }
f06e48d8 980 for (const workerNode of pool.workerNodes) {
465b2940 981 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
982 tasks: {
983 executed: 0,
984 executing: maxMultiplier,
985 queued: 0,
df593701 986 maxQueued: 0,
463226a4 987 sequentiallyStolen: 0,
68cbdc84 988 stolen: 0,
a4e07f72
JB
989 failed: 0
990 },
991 runTime: {
a4e07f72
JB
992 history: expect.any(CircularArray)
993 },
994 waitTime: {
a4e07f72
JB
995 history: expect.any(CircularArray)
996 },
5df69fab
JB
997 elu: {
998 idle: {
5df69fab
JB
999 history: expect.any(CircularArray)
1000 },
1001 active: {
5df69fab 1002 history: expect.any(CircularArray)
f7510105 1003 }
5df69fab 1004 }
86bf340d 1005 })
bf9549ae
JB
1006 }
1007 await Promise.all(promises)
f06e48d8 1008 for (const workerNode of pool.workerNodes) {
465b2940 1009 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
1010 tasks: {
1011 executed: maxMultiplier,
1012 executing: 0,
1013 queued: 0,
df593701 1014 maxQueued: 0,
463226a4 1015 sequentiallyStolen: 0,
68cbdc84 1016 stolen: 0,
a4e07f72
JB
1017 failed: 0
1018 },
1019 runTime: {
a4e07f72
JB
1020 history: expect.any(CircularArray)
1021 },
1022 waitTime: {
a4e07f72
JB
1023 history: expect.any(CircularArray)
1024 },
5df69fab
JB
1025 elu: {
1026 idle: {
5df69fab
JB
1027 history: expect.any(CircularArray)
1028 },
1029 active: {
5df69fab 1030 history: expect.any(CircularArray)
f7510105 1031 }
5df69fab 1032 }
86bf340d 1033 })
bf9549ae 1034 }
fd7ebd49 1035 await pool.destroy()
bf9549ae
JB
1036 })
1037
2431bdb4 1038 it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
7fd82a1c 1039 const pool = new DynamicThreadPool(
2431bdb4 1040 Math.floor(numberOfWorkers / 2),
8f4878b7 1041 numberOfWorkers,
b2fd3f4a 1042 './tests/worker-files/thread/testWorker.mjs'
9e619829 1043 )
09c2d0d3 1044 const promises = new Set()
ee9f5295
JB
1045 const maxMultiplier = 2
1046 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 1047 promises.add(pool.execute())
9e619829
JB
1048 }
1049 await Promise.all(promises)
f06e48d8 1050 for (const workerNode of pool.workerNodes) {
465b2940 1051 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
1052 tasks: {
1053 executed: expect.any(Number),
1054 executing: 0,
1055 queued: 0,
df593701 1056 maxQueued: 0,
463226a4 1057 sequentiallyStolen: 0,
68cbdc84 1058 stolen: 0,
a4e07f72
JB
1059 failed: 0
1060 },
1061 runTime: {
a4e07f72
JB
1062 history: expect.any(CircularArray)
1063 },
1064 waitTime: {
a4e07f72
JB
1065 history: expect.any(CircularArray)
1066 },
5df69fab
JB
1067 elu: {
1068 idle: {
5df69fab
JB
1069 history: expect.any(CircularArray)
1070 },
1071 active: {
5df69fab 1072 history: expect.any(CircularArray)
f7510105 1073 }
5df69fab 1074 }
86bf340d 1075 })
465b2940 1076 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
94407def
JB
1077 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1078 numberOfWorkers * maxMultiplier
1079 )
b97d82d8
JB
1080 expect(workerNode.usage.runTime.history.length).toBe(0)
1081 expect(workerNode.usage.waitTime.history.length).toBe(0)
1082 expect(workerNode.usage.elu.idle.history.length).toBe(0)
1083 expect(workerNode.usage.elu.active.history.length).toBe(0)
9e619829
JB
1084 }
1085 pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
f06e48d8 1086 for (const workerNode of pool.workerNodes) {
465b2940 1087 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
1088 tasks: {
1089 executed: 0,
1090 executing: 0,
1091 queued: 0,
df593701 1092 maxQueued: 0,
463226a4 1093 sequentiallyStolen: 0,
68cbdc84 1094 stolen: 0,
a4e07f72
JB
1095 failed: 0
1096 },
1097 runTime: {
a4e07f72
JB
1098 history: expect.any(CircularArray)
1099 },
1100 waitTime: {
a4e07f72
JB
1101 history: expect.any(CircularArray)
1102 },
5df69fab
JB
1103 elu: {
1104 idle: {
5df69fab
JB
1105 history: expect.any(CircularArray)
1106 },
1107 active: {
5df69fab 1108 history: expect.any(CircularArray)
f7510105 1109 }
5df69fab 1110 }
86bf340d 1111 })
465b2940
JB
1112 expect(workerNode.usage.runTime.history.length).toBe(0)
1113 expect(workerNode.usage.waitTime.history.length).toBe(0)
b97d82d8
JB
1114 expect(workerNode.usage.elu.idle.history.length).toBe(0)
1115 expect(workerNode.usage.elu.active.history.length).toBe(0)
ee11a4a2 1116 }
fd7ebd49 1117 await pool.destroy()
ee11a4a2
JB
1118 })
1119
a1763c54
JB
1120 it("Verify that pool event emitter 'ready' event can register a callback", async () => {
1121 const pool = new DynamicClusterPool(
2431bdb4 1122 Math.floor(numberOfWorkers / 2),
164d950a 1123 numberOfWorkers,
a1763c54 1124 './tests/worker-files/cluster/testWorker.js'
164d950a 1125 )
c726f66c 1126 expect(pool.emitter.eventNames()).toStrictEqual([])
d46660cd 1127 let poolInfo
a1763c54 1128 let poolReady = 0
041dc05b 1129 pool.emitter.on(PoolEvents.ready, info => {
a1763c54 1130 ++poolReady
d46660cd
JB
1131 poolInfo = info
1132 })
a1763c54 1133 await waitPoolEvents(pool, PoolEvents.ready, 1)
c726f66c 1134 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.ready])
a1763c54 1135 expect(poolReady).toBe(1)
d46660cd 1136 expect(poolInfo).toStrictEqual({
23ccf9d7 1137 version,
d46660cd 1138 type: PoolTypes.dynamic,
a1763c54 1139 worker: WorkerTypes.cluster,
47352846 1140 started: true,
a1763c54 1141 ready: true,
2431bdb4
JB
1142 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1143 minSize: expect.any(Number),
1144 maxSize: expect.any(Number),
1145 workerNodes: expect.any(Number),
1146 idleWorkerNodes: expect.any(Number),
1147 busyWorkerNodes: expect.any(Number),
1148 executedTasks: expect.any(Number),
1149 executingTasks: expect.any(Number),
2431bdb4
JB
1150 failedTasks: expect.any(Number)
1151 })
1152 await pool.destroy()
1153 })
1154
a1763c54
JB
1155 it("Verify that pool event emitter 'busy' event can register a callback", async () => {
1156 const pool = new FixedThreadPool(
2431bdb4 1157 numberOfWorkers,
b2fd3f4a 1158 './tests/worker-files/thread/testWorker.mjs'
2431bdb4 1159 )
c726f66c 1160 expect(pool.emitter.eventNames()).toStrictEqual([])
a1763c54
JB
1161 const promises = new Set()
1162 let poolBusy = 0
2431bdb4 1163 let poolInfo
041dc05b 1164 pool.emitter.on(PoolEvents.busy, info => {
a1763c54 1165 ++poolBusy
2431bdb4
JB
1166 poolInfo = info
1167 })
c726f66c 1168 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
a1763c54
JB
1169 for (let i = 0; i < numberOfWorkers * 2; i++) {
1170 promises.add(pool.execute())
1171 }
1172 await Promise.all(promises)
1173 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
1174 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
1175 expect(poolBusy).toBe(numberOfWorkers + 1)
2431bdb4
JB
1176 expect(poolInfo).toStrictEqual({
1177 version,
a1763c54
JB
1178 type: PoolTypes.fixed,
1179 worker: WorkerTypes.thread,
47352846
JB
1180 started: true,
1181 ready: true,
2431bdb4 1182 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
1183 minSize: expect.any(Number),
1184 maxSize: expect.any(Number),
1185 workerNodes: expect.any(Number),
1186 idleWorkerNodes: expect.any(Number),
1187 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1188 executedTasks: expect.any(Number),
1189 executingTasks: expect.any(Number),
a4e07f72 1190 failedTasks: expect.any(Number)
d46660cd 1191 })
164d950a
JB
1192 await pool.destroy()
1193 })
1194
a1763c54
JB
1195 it("Verify that pool event emitter 'full' event can register a callback", async () => {
1196 const pool = new DynamicThreadPool(
1197 Math.floor(numberOfWorkers / 2),
7c0ba920 1198 numberOfWorkers,
b2fd3f4a 1199 './tests/worker-files/thread/testWorker.mjs'
7c0ba920 1200 )
c726f66c 1201 expect(pool.emitter.eventNames()).toStrictEqual([])
09c2d0d3 1202 const promises = new Set()
a1763c54 1203 let poolFull = 0
d46660cd 1204 let poolInfo
041dc05b 1205 pool.emitter.on(PoolEvents.full, info => {
a1763c54 1206 ++poolFull
d46660cd
JB
1207 poolInfo = info
1208 })
c726f66c 1209 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.full])
7c0ba920 1210 for (let i = 0; i < numberOfWorkers * 2; i++) {
f5d14e90 1211 promises.add(pool.execute())
7c0ba920 1212 }
cf597bc5 1213 await Promise.all(promises)
33e6bb4c 1214 expect(poolFull).toBe(1)
d46660cd 1215 expect(poolInfo).toStrictEqual({
23ccf9d7 1216 version,
a1763c54 1217 type: PoolTypes.dynamic,
d46660cd 1218 worker: WorkerTypes.thread,
47352846
JB
1219 started: true,
1220 ready: true,
2431bdb4 1221 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
8735b4e5
JB
1222 minSize: expect.any(Number),
1223 maxSize: expect.any(Number),
1224 workerNodes: expect.any(Number),
1225 idleWorkerNodes: expect.any(Number),
1226 busyWorkerNodes: expect.any(Number),
1227 executedTasks: expect.any(Number),
1228 executingTasks: expect.any(Number),
1229 failedTasks: expect.any(Number)
1230 })
1231 await pool.destroy()
1232 })
1233
3e8611a8 1234 it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
b1aae695 1235 const pool = new FixedThreadPool(
8735b4e5 1236 numberOfWorkers,
b2fd3f4a 1237 './tests/worker-files/thread/testWorker.mjs',
8735b4e5
JB
1238 {
1239 enableTasksQueue: true
1240 }
1241 )
a074ffee 1242 stub(pool, 'hasBackPressure').returns(true)
c726f66c 1243 expect(pool.emitter.eventNames()).toStrictEqual([])
8735b4e5
JB
1244 const promises = new Set()
1245 let poolBackPressure = 0
1246 let poolInfo
041dc05b 1247 pool.emitter.on(PoolEvents.backPressure, info => {
8735b4e5
JB
1248 ++poolBackPressure
1249 poolInfo = info
1250 })
c726f66c 1251 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.backPressure])
033f1776 1252 for (let i = 0; i < numberOfWorkers + 1; i++) {
8735b4e5
JB
1253 promises.add(pool.execute())
1254 }
1255 await Promise.all(promises)
033f1776 1256 expect(poolBackPressure).toBe(1)
8735b4e5
JB
1257 expect(poolInfo).toStrictEqual({
1258 version,
3e8611a8 1259 type: PoolTypes.fixed,
8735b4e5 1260 worker: WorkerTypes.thread,
47352846
JB
1261 started: true,
1262 ready: true,
8735b4e5 1263 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
1264 minSize: expect.any(Number),
1265 maxSize: expect.any(Number),
1266 workerNodes: expect.any(Number),
1267 idleWorkerNodes: expect.any(Number),
1268 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
1269 executedTasks: expect.any(Number),
1270 executingTasks: expect.any(Number),
3e8611a8
JB
1271 maxQueuedTasks: expect.any(Number),
1272 queuedTasks: expect.any(Number),
1273 backPressure: true,
68cbdc84 1274 stolenTasks: expect.any(Number),
a4e07f72 1275 failedTasks: expect.any(Number)
d46660cd 1276 })
e458c82e 1277 expect(pool.hasBackPressure.callCount).toBe(5)
fd7ebd49 1278 await pool.destroy()
7c0ba920 1279 })
70a4f5ea 1280
85b2561d
JB
1281 it('Verify that destroy() waits for queued tasks to finish', async () => {
1282 const tasksFinishedTimeout = 2500
1283 const pool = new FixedThreadPool(
1284 numberOfWorkers,
1285 './tests/worker-files/thread/asyncWorker.mjs',
1286 {
1287 enableTasksQueue: true,
1288 tasksQueueOptions: { tasksFinishedTimeout }
1289 }
1290 )
1291 const maxMultiplier = 4
1292 let tasksFinished = 0
1293 for (const workerNode of pool.workerNodes) {
1294 workerNode.on('taskFinished', () => {
1295 ++tasksFinished
1296 })
1297 }
1298 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1299 pool.execute()
1300 }
1301 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1302 const startTime = performance.now()
1303 await pool.destroy()
1304 const elapsedTime = performance.now() - startTime
1305 expect(tasksFinished).toBe(numberOfWorkers * maxMultiplier)
1306 expect(elapsedTime).toBeGreaterThanOrEqual(2000)
6d41131f 1307 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 100)
85b2561d
JB
1308 })
1309
1310 it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
1311 const tasksFinishedTimeout = 1000
1312 const pool = new FixedThreadPool(
1313 numberOfWorkers,
1314 './tests/worker-files/thread/asyncWorker.mjs',
1315 {
1316 enableTasksQueue: true,
1317 tasksQueueOptions: { tasksFinishedTimeout }
1318 }
1319 )
1320 const maxMultiplier = 4
1321 let tasksFinished = 0
1322 for (const workerNode of pool.workerNodes) {
1323 workerNode.on('taskFinished', () => {
1324 ++tasksFinished
1325 })
1326 }
1327 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1328 pool.execute()
1329 }
1330 expect(pool.info.queuedTasks).toBeGreaterThan(0)
1331 const startTime = performance.now()
1332 await pool.destroy()
1333 const elapsedTime = performance.now() - startTime
1334 expect(tasksFinished).toBe(0)
c004ecec 1335 expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 600)
85b2561d
JB
1336 })
1337
f18fd12b
JB
1338 it('Verify that pool asynchronous resource track tasks execution', async () => {
1339 let taskAsyncId
1340 let initCalls = 0
1341 let beforeCalls = 0
1342 let afterCalls = 0
1343 let resolveCalls = 0
1344 const hook = createHook({
1345 init (asyncId, type) {
1346 if (type === 'poolifier:task') {
1347 initCalls++
1348 taskAsyncId = asyncId
1349 }
1350 },
1351 before (asyncId) {
1352 if (asyncId === taskAsyncId) beforeCalls++
1353 },
1354 after (asyncId) {
1355 if (asyncId === taskAsyncId) afterCalls++
1356 },
1357 promiseResolve () {
1358 if (executionAsyncId() === taskAsyncId) resolveCalls++
1359 }
1360 })
f18fd12b
JB
1361 const pool = new FixedThreadPool(
1362 numberOfWorkers,
1363 './tests/worker-files/thread/testWorker.mjs'
1364 )
8954c0a3 1365 hook.enable()
f18fd12b
JB
1366 await pool.execute()
1367 hook.disable()
1368 expect(initCalls).toBe(1)
1369 expect(beforeCalls).toBe(1)
1370 expect(afterCalls).toBe(1)
1371 expect(resolveCalls).toBe(1)
8954c0a3 1372 await pool.destroy()
f18fd12b
JB
1373 })
1374
9eae3c69
JB
1375 it('Verify that hasTaskFunction() is working', async () => {
1376 const dynamicThreadPool = new DynamicThreadPool(
1377 Math.floor(numberOfWorkers / 2),
1378 numberOfWorkers,
b2fd3f4a 1379 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
9eae3c69
JB
1380 )
1381 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1382 expect(dynamicThreadPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1383 expect(dynamicThreadPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1384 true
1385 )
1386 expect(dynamicThreadPool.hasTaskFunction('factorial')).toBe(true)
1387 expect(dynamicThreadPool.hasTaskFunction('fibonacci')).toBe(true)
1388 expect(dynamicThreadPool.hasTaskFunction('unknown')).toBe(false)
1389 await dynamicThreadPool.destroy()
1390 const fixedClusterPool = new FixedClusterPool(
1391 numberOfWorkers,
1392 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1393 )
1394 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1395 expect(fixedClusterPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1396 expect(fixedClusterPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1397 true
1398 )
1399 expect(fixedClusterPool.hasTaskFunction('factorial')).toBe(true)
1400 expect(fixedClusterPool.hasTaskFunction('fibonacci')).toBe(true)
1401 expect(fixedClusterPool.hasTaskFunction('unknown')).toBe(false)
1402 await fixedClusterPool.destroy()
1403 })
1404
1405 it('Verify that addTaskFunction() is working', async () => {
1406 const dynamicThreadPool = new DynamicThreadPool(
1407 Math.floor(numberOfWorkers / 2),
1408 numberOfWorkers,
b2fd3f4a 1409 './tests/worker-files/thread/testWorker.mjs'
9eae3c69
JB
1410 )
1411 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
3feeab69
JB
1412 await expect(
1413 dynamicThreadPool.addTaskFunction(0, () => {})
948faff7 1414 ).rejects.toThrow(new TypeError('name argument must be a string'))
3feeab69
JB
1415 await expect(
1416 dynamicThreadPool.addTaskFunction('', () => {})
948faff7 1417 ).rejects.toThrow(
3feeab69
JB
1418 new TypeError('name argument must not be an empty string')
1419 )
948faff7
JB
1420 await expect(dynamicThreadPool.addTaskFunction('test', 0)).rejects.toThrow(
1421 new TypeError('fn argument must be a function')
1422 )
1423 await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow(
1424 new TypeError('fn argument must be a function')
1425 )
9eae3c69
JB
1426 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1427 DEFAULT_TASK_NAME,
1428 'test'
1429 ])
1430 const echoTaskFunction = data => {
1431 return data
1432 }
1433 await expect(
1434 dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1435 ).resolves.toBe(true)
1436 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1437 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1438 echoTaskFunction
1439 )
1440 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1441 DEFAULT_TASK_NAME,
1442 'test',
1443 'echo'
1444 ])
1445 const taskFunctionData = { test: 'test' }
1446 const echoResult = await dynamicThreadPool.execute(taskFunctionData, 'echo')
1447 expect(echoResult).toStrictEqual(taskFunctionData)
adee6053
JB
1448 for (const workerNode of dynamicThreadPool.workerNodes) {
1449 expect(workerNode.getTaskFunctionWorkerUsage('echo')).toStrictEqual({
1450 tasks: {
1451 executed: expect.any(Number),
1452 executing: 0,
1453 queued: 0,
463226a4 1454 sequentiallyStolen: 0,
5ad42e34 1455 stolen: 0,
adee6053
JB
1456 failed: 0
1457 },
1458 runTime: {
1459 history: new CircularArray()
1460 },
1461 waitTime: {
1462 history: new CircularArray()
1463 },
1464 elu: {
1465 idle: {
1466 history: new CircularArray()
1467 },
1468 active: {
1469 history: new CircularArray()
1470 }
1471 }
1472 })
1473 }
9eae3c69
JB
1474 await dynamicThreadPool.destroy()
1475 })
1476
1477 it('Verify that removeTaskFunction() is working', async () => {
1478 const dynamicThreadPool = new DynamicThreadPool(
1479 Math.floor(numberOfWorkers / 2),
1480 numberOfWorkers,
b2fd3f4a 1481 './tests/worker-files/thread/testWorker.mjs'
9eae3c69
JB
1482 )
1483 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1484 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1485 DEFAULT_TASK_NAME,
1486 'test'
1487 ])
948faff7 1488 await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow(
16248b23 1489 new Error('Cannot remove a task function not handled on the pool side')
9eae3c69
JB
1490 )
1491 const echoTaskFunction = data => {
1492 return data
1493 }
1494 await dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1495 expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1496 expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1497 echoTaskFunction
1498 )
1499 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1500 DEFAULT_TASK_NAME,
1501 'test',
1502 'echo'
1503 ])
1504 await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe(
1505 true
1506 )
1507 expect(dynamicThreadPool.taskFunctions.size).toBe(0)
1508 expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined()
1509 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1510 DEFAULT_TASK_NAME,
1511 'test'
1512 ])
1513 await dynamicThreadPool.destroy()
1514 })
1515
30500265 1516 it('Verify that listTaskFunctionNames() is working', async () => {
90d7d101
JB
1517 const dynamicThreadPool = new DynamicThreadPool(
1518 Math.floor(numberOfWorkers / 2),
1519 numberOfWorkers,
b2fd3f4a 1520 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
90d7d101
JB
1521 )
1522 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
66979634 1523 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
6cd5248f 1524 DEFAULT_TASK_NAME,
90d7d101
JB
1525 'jsonIntegerSerialization',
1526 'factorial',
1527 'fibonacci'
1528 ])
9eae3c69 1529 await dynamicThreadPool.destroy()
90d7d101
JB
1530 const fixedClusterPool = new FixedClusterPool(
1531 numberOfWorkers,
1532 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1533 )
1534 await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
66979634 1535 expect(fixedClusterPool.listTaskFunctionNames()).toStrictEqual([
6cd5248f 1536 DEFAULT_TASK_NAME,
90d7d101
JB
1537 'jsonIntegerSerialization',
1538 'factorial',
1539 'fibonacci'
1540 ])
0fe39c97 1541 await fixedClusterPool.destroy()
90d7d101
JB
1542 })
1543
9eae3c69 1544 it('Verify that setDefaultTaskFunction() is working', async () => {
30500265
JB
1545 const dynamicThreadPool = new DynamicThreadPool(
1546 Math.floor(numberOfWorkers / 2),
1547 numberOfWorkers,
b2fd3f4a 1548 './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
30500265
JB
1549 )
1550 await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
711623b8 1551 const workerId = dynamicThreadPool.workerNodes[0].info.id
948faff7 1552 await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow(
b0b55f57 1553 new Error(
711623b8 1554 `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'`
b0b55f57
JB
1555 )
1556 )
1557 await expect(
1558 dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME)
948faff7 1559 ).rejects.toThrow(
b0b55f57 1560 new Error(
711623b8 1561 `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
1562 )
1563 )
1564 await expect(
1565 dynamicThreadPool.setDefaultTaskFunction('unknown')
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 to a non-existing task function'`
b0b55f57
JB
1569 )
1570 )
9eae3c69
JB
1571 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1572 DEFAULT_TASK_NAME,
1573 'jsonIntegerSerialization',
1574 'factorial',
1575 'fibonacci'
1576 ])
1577 await expect(
1578 dynamicThreadPool.setDefaultTaskFunction('factorial')
1579 ).resolves.toBe(true)
1580 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1581 DEFAULT_TASK_NAME,
1582 'factorial',
1583 'jsonIntegerSerialization',
1584 'fibonacci'
1585 ])
1586 await expect(
1587 dynamicThreadPool.setDefaultTaskFunction('fibonacci')
1588 ).resolves.toBe(true)
1589 expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1590 DEFAULT_TASK_NAME,
1591 'fibonacci',
1592 'jsonIntegerSerialization',
1593 'factorial'
1594 ])
cda9ba34 1595 await dynamicThreadPool.destroy()
30500265
JB
1596 })
1597
90d7d101 1598 it('Verify that multiple task functions worker is working', async () => {
70a4f5ea 1599 const pool = new DynamicClusterPool(
2431bdb4 1600 Math.floor(numberOfWorkers / 2),
70a4f5ea 1601 numberOfWorkers,
90d7d101 1602 './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
70a4f5ea
JB
1603 )
1604 const data = { n: 10 }
82888165 1605 const result0 = await pool.execute(data)
30b963d4 1606 expect(result0).toStrictEqual({ ok: 1 })
70a4f5ea 1607 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
30b963d4 1608 expect(result1).toStrictEqual({ ok: 1 })
70a4f5ea
JB
1609 const result2 = await pool.execute(data, 'factorial')
1610 expect(result2).toBe(3628800)
1611 const result3 = await pool.execute(data, 'fibonacci')
024daf59 1612 expect(result3).toBe(55)
5bb5be17
JB
1613 expect(pool.info.executingTasks).toBe(0)
1614 expect(pool.info.executedTasks).toBe(4)
b414b84c 1615 for (const workerNode of pool.workerNodes) {
66979634 1616 expect(workerNode.info.taskFunctionNames).toStrictEqual([
6cd5248f 1617 DEFAULT_TASK_NAME,
b414b84c
JB
1618 'jsonIntegerSerialization',
1619 'factorial',
1620 'fibonacci'
1621 ])
1622 expect(workerNode.taskFunctionsUsage.size).toBe(3)
66979634 1623 for (const name of pool.listTaskFunctionNames()) {
5bb5be17
JB
1624 expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1625 tasks: {
1626 executed: expect.any(Number),
4ba4c7f9 1627 executing: 0,
5bb5be17 1628 failed: 0,
68cbdc84 1629 queued: 0,
463226a4 1630 sequentiallyStolen: 0,
68cbdc84 1631 stolen: 0
5bb5be17
JB
1632 },
1633 runTime: {
1634 history: expect.any(CircularArray)
1635 },
1636 waitTime: {
1637 history: expect.any(CircularArray)
1638 },
1639 elu: {
1640 idle: {
1641 history: expect.any(CircularArray)
1642 },
1643 active: {
1644 history: expect.any(CircularArray)
1645 }
1646 }
1647 })
1648 expect(
4ba4c7f9
JB
1649 workerNode.getTaskFunctionWorkerUsage(name).tasks.executed
1650 ).toBeGreaterThan(0)
5bb5be17 1651 }
dfd7ec01
JB
1652 expect(
1653 workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1654 ).toStrictEqual(
66979634
JB
1655 workerNode.getTaskFunctionWorkerUsage(
1656 workerNode.info.taskFunctionNames[1]
1657 )
dfd7ec01 1658 )
5bb5be17 1659 }
0fe39c97 1660 await pool.destroy()
70a4f5ea 1661 })
52a23942
JB
1662
1663 it('Verify sendKillMessageToWorker()', async () => {
1664 const pool = new DynamicClusterPool(
1665 Math.floor(numberOfWorkers / 2),
1666 numberOfWorkers,
1667 './tests/worker-files/cluster/testWorker.js'
1668 )
1669 const workerNodeKey = 0
1670 await expect(
adee6053 1671 pool.sendKillMessageToWorker(workerNodeKey)
52a23942 1672 ).resolves.toBeUndefined()
85b2561d
JB
1673 await expect(
1674 pool.sendKillMessageToWorker(numberOfWorkers)
1675 ).rejects.toStrictEqual(
1676 new Error(`Invalid worker node key '${numberOfWorkers}'`)
1677 )
52a23942
JB
1678 await pool.destroy()
1679 })
adee6053
JB
1680
1681 it('Verify sendTaskFunctionOperationToWorker()', async () => {
1682 const pool = new DynamicClusterPool(
1683 Math.floor(numberOfWorkers / 2),
1684 numberOfWorkers,
1685 './tests/worker-files/cluster/testWorker.js'
1686 )
1687 const workerNodeKey = 0
1688 await expect(
1689 pool.sendTaskFunctionOperationToWorker(workerNodeKey, {
1690 taskFunctionOperation: 'add',
1691 taskFunctionName: 'empty',
1692 taskFunction: (() => {}).toString()
1693 })
1694 ).resolves.toBe(true)
1695 expect(
1696 pool.workerNodes[workerNodeKey].info.taskFunctionNames
1697 ).toStrictEqual([DEFAULT_TASK_NAME, 'test', 'empty'])
1698 await pool.destroy()
1699 })
1700
1701 it('Verify sendTaskFunctionOperationToWorkers()', async () => {
1702 const pool = new DynamicClusterPool(
1703 Math.floor(numberOfWorkers / 2),
1704 numberOfWorkers,
1705 './tests/worker-files/cluster/testWorker.js'
1706 )
adee6053
JB
1707 await expect(
1708 pool.sendTaskFunctionOperationToWorkers({
1709 taskFunctionOperation: 'add',
1710 taskFunctionName: 'empty',
1711 taskFunction: (() => {}).toString()
1712 })
1713 ).resolves.toBe(true)
1714 for (const workerNode of pool.workerNodes) {
1715 expect(workerNode.info.taskFunctionNames).toStrictEqual([
1716 DEFAULT_TASK_NAME,
1717 'test',
1718 'empty'
1719 ])
1720 }
1721 await pool.destroy()
1722 })
3ec964d6 1723})