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