Merge branch 'master' of github.com:poolifier/poolifier
[poolifier.git] / tests / pools / abstract / abstract-pool.test.js
CommitLineData
a61a0724 1const { expect } = require('expect')
e843b904 2const {
70a4f5ea 3 DynamicClusterPool,
9e619829 4 DynamicThreadPool,
aee46736 5 FixedClusterPool,
e843b904 6 FixedThreadPool,
aee46736 7 PoolEvents,
6b27d407 8 WorkerChoiceStrategies,
184855e6
JB
9 PoolTypes,
10 WorkerTypes
cdace0e5 11} = require('../../../lib')
78099a15 12const { CircularArray } = require('../../../lib/circular-array')
29ee7e9a 13const { Queue } = require('../../../lib/queue')
23ccf9d7 14const { version } = require('../../../package.json')
2431bdb4 15const { waitPoolEvents } = require('../../test-utils')
e1ffb94f
JB
16
17describe('Abstract pool test suite', () => {
fc027381 18 const numberOfWorkers = 2
a8884ffd 19 class StubPoolWithIsMain extends FixedThreadPool {
e1ffb94f
JB
20 isMain () {
21 return false
22 }
3ec964d6 23 }
3ec964d6 24
3ec964d6 25 it('Simulate pool creation from a non main thread/process', () => {
8d3782fa
JB
26 expect(
27 () =>
a8884ffd 28 new StubPoolWithIsMain(
7c0ba920 29 numberOfWorkers,
8d3782fa
JB
30 './tests/worker-files/thread/testWorker.js',
31 {
32 errorHandler: e => console.error(e)
33 }
34 )
d4aeae5a 35 ).toThrowError('Cannot start a pool from a worker!')
3ec964d6 36 })
c510fea7
APA
37
38 it('Verify that filePath is checked', () => {
292ad316
JB
39 const expectedError = new Error(
40 'Please specify a file with a worker implementation'
41 )
7c0ba920 42 expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError(
292ad316 43 expectedError
8d3782fa 44 )
7c0ba920 45 expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError(
292ad316 46 expectedError
8d3782fa
JB
47 )
48 })
49
50 it('Verify that numberOfWorkers is checked', () => {
51 expect(() => new FixedThreadPool()).toThrowError(
d4aeae5a 52 'Cannot instantiate a pool without specifying the number of workers'
8d3782fa
JB
53 )
54 })
55
56 it('Verify that a negative number of workers is checked', () => {
57 expect(
58 () =>
59 new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
60 ).toThrowError(
473c717a
JB
61 new RangeError(
62 'Cannot instantiate a pool with a negative number of workers'
63 )
8d3782fa
JB
64 )
65 })
66
67 it('Verify that a non integer number of workers is checked', () => {
68 expect(
69 () =>
70 new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.js')
71 ).toThrowError(
473c717a 72 new TypeError(
0d80593b 73 'Cannot instantiate a pool with a non safe integer number of workers'
8d3782fa
JB
74 )
75 )
c510fea7 76 })
7c0ba920 77
2431bdb4
JB
78 it('Verify dynamic pool sizing', () => {
79 expect(
80 () =>
81 new DynamicThreadPool(2, 1, './tests/worker-files/thread/testWorker.js')
82 ).toThrowError(
83 new RangeError(
84 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
85 )
86 )
87 expect(
88 () =>
89 new DynamicThreadPool(1, 1, './tests/worker-files/thread/testWorker.js')
90 ).toThrowError(
91 new RangeError(
92 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
93 )
94 )
21f710aa
JB
95 expect(
96 () =>
97 new DynamicThreadPool(0, 0, './tests/worker-files/thread/testWorker.js')
98 ).toThrowError(
99 new RangeError(
100 'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
101 )
102 )
2431bdb4
JB
103 })
104
fd7ebd49 105 it('Verify that pool options are checked', async () => {
7c0ba920
JB
106 let pool = new FixedThreadPool(
107 numberOfWorkers,
108 './tests/worker-files/thread/testWorker.js'
109 )
7c0ba920 110 expect(pool.emitter).toBeDefined()
1f68cede
JB
111 expect(pool.opts.enableEvents).toBe(true)
112 expect(pool.opts.restartWorkerOnError).toBe(true)
ff733df7 113 expect(pool.opts.enableTasksQueue).toBe(false)
d4aeae5a 114 expect(pool.opts.tasksQueueOptions).toBeUndefined()
e843b904
JB
115 expect(pool.opts.workerChoiceStrategy).toBe(
116 WorkerChoiceStrategies.ROUND_ROBIN
117 )
da309861 118 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
932fc8be 119 runTime: { median: false },
5df69fab
JB
120 waitTime: { median: false },
121 elu: { median: false }
da309861 122 })
35cf1c03
JB
123 expect(pool.opts.messageHandler).toBeUndefined()
124 expect(pool.opts.errorHandler).toBeUndefined()
125 expect(pool.opts.onlineHandler).toBeUndefined()
126 expect(pool.opts.exitHandler).toBeUndefined()
fd7ebd49 127 await pool.destroy()
35cf1c03 128 const testHandler = () => console.log('test handler executed')
7c0ba920
JB
129 pool = new FixedThreadPool(
130 numberOfWorkers,
131 './tests/worker-files/thread/testWorker.js',
132 {
e4543b14 133 workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
49be33fe 134 workerChoiceStrategyOptions: {
932fc8be 135 runTime: { median: true },
fc027381 136 weights: { 0: 300, 1: 200 }
49be33fe 137 },
35cf1c03 138 enableEvents: false,
1f68cede 139 restartWorkerOnError: false,
ff733df7 140 enableTasksQueue: true,
d4aeae5a 141 tasksQueueOptions: { concurrency: 2 },
35cf1c03
JB
142 messageHandler: testHandler,
143 errorHandler: testHandler,
144 onlineHandler: testHandler,
145 exitHandler: testHandler
7c0ba920
JB
146 }
147 )
7c0ba920 148 expect(pool.emitter).toBeUndefined()
1f68cede
JB
149 expect(pool.opts.enableEvents).toBe(false)
150 expect(pool.opts.restartWorkerOnError).toBe(false)
ff733df7 151 expect(pool.opts.enableTasksQueue).toBe(true)
d4aeae5a 152 expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
e843b904 153 expect(pool.opts.workerChoiceStrategy).toBe(
e4543b14 154 WorkerChoiceStrategies.LEAST_USED
e843b904 155 )
da309861 156 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
932fc8be 157 runTime: { median: true },
fc027381 158 weights: { 0: 300, 1: 200 }
da309861 159 })
35cf1c03
JB
160 expect(pool.opts.messageHandler).toStrictEqual(testHandler)
161 expect(pool.opts.errorHandler).toStrictEqual(testHandler)
162 expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
163 expect(pool.opts.exitHandler).toStrictEqual(testHandler)
fd7ebd49 164 await pool.destroy()
7c0ba920
JB
165 })
166
a20f0ba5 167 it('Verify that pool options are validated', async () => {
d4aeae5a
JB
168 expect(
169 () =>
170 new FixedThreadPool(
171 numberOfWorkers,
172 './tests/worker-files/thread/testWorker.js',
173 {
f0d7f803 174 workerChoiceStrategy: 'invalidStrategy'
d4aeae5a
JB
175 }
176 )
f0d7f803 177 ).toThrowError("Invalid worker choice strategy 'invalidStrategy'")
d4aeae5a
JB
178 expect(
179 () =>
180 new FixedThreadPool(
181 numberOfWorkers,
182 './tests/worker-files/thread/testWorker.js',
183 {
f0d7f803 184 workerChoiceStrategyOptions: 'invalidOptions'
d4aeae5a
JB
185 }
186 )
f0d7f803
JB
187 ).toThrowError(
188 'Invalid worker choice strategy options: must be a plain object'
189 )
49be33fe
JB
190 expect(
191 () =>
192 new FixedThreadPool(
193 numberOfWorkers,
194 './tests/worker-files/thread/testWorker.js',
195 {
196 workerChoiceStrategyOptions: { weights: {} }
197 }
198 )
199 ).toThrowError(
200 'Invalid worker choice strategy options: must have a weight for each worker node'
201 )
f0d7f803
JB
202 expect(
203 () =>
204 new FixedThreadPool(
205 numberOfWorkers,
206 './tests/worker-files/thread/testWorker.js',
207 {
208 workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
209 }
210 )
211 ).toThrowError(
212 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
213 )
214 expect(
215 () =>
216 new FixedThreadPool(
217 numberOfWorkers,
218 './tests/worker-files/thread/testWorker.js',
219 {
220 enableTasksQueue: true,
221 tasksQueueOptions: { concurrency: 0 }
222 }
223 )
224 ).toThrowError("Invalid worker tasks concurrency '0'")
225 expect(
226 () =>
227 new FixedThreadPool(
228 numberOfWorkers,
229 './tests/worker-files/thread/testWorker.js',
230 {
231 enableTasksQueue: true,
232 tasksQueueOptions: 'invalidTasksQueueOptions'
233 }
234 )
235 ).toThrowError('Invalid tasks queue options: must be a plain object')
236 expect(
237 () =>
238 new FixedThreadPool(
239 numberOfWorkers,
240 './tests/worker-files/thread/testWorker.js',
241 {
242 enableTasksQueue: true,
243 tasksQueueOptions: { concurrency: 0.2 }
244 }
245 )
246 ).toThrowError('Invalid worker tasks concurrency: must be an integer')
d4aeae5a
JB
247 })
248
2431bdb4 249 it('Verify that pool worker choice strategy options can be set', async () => {
a20f0ba5
JB
250 const pool = new FixedThreadPool(
251 numberOfWorkers,
252 './tests/worker-files/thread/testWorker.js',
253 { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
254 )
255 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
932fc8be 256 runTime: { median: false },
5df69fab
JB
257 waitTime: { median: false },
258 elu: { median: false }
a20f0ba5
JB
259 })
260 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
261 .workerChoiceStrategies) {
86bf340d 262 expect(workerChoiceStrategy.opts).toStrictEqual({
932fc8be 263 runTime: { median: false },
5df69fab
JB
264 waitTime: { median: false },
265 elu: { median: false }
86bf340d 266 })
a20f0ba5 267 }
87de9ff5
JB
268 expect(
269 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
270 ).toStrictEqual({
932fc8be
JB
271 runTime: {
272 aggregate: true,
273 average: true,
274 median: false
275 },
276 waitTime: {
277 aggregate: false,
278 average: false,
279 median: false
280 },
5df69fab 281 elu: {
9adcefab
JB
282 aggregate: true,
283 average: true,
5df69fab
JB
284 median: false
285 }
86bf340d 286 })
9adcefab
JB
287 pool.setWorkerChoiceStrategyOptions({
288 runTime: { median: true },
289 elu: { median: true }
290 })
a20f0ba5 291 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
9adcefab
JB
292 runTime: { median: true },
293 elu: { median: true }
a20f0ba5
JB
294 })
295 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
296 .workerChoiceStrategies) {
932fc8be 297 expect(workerChoiceStrategy.opts).toStrictEqual({
9adcefab
JB
298 runTime: { median: true },
299 elu: { median: true }
932fc8be 300 })
a20f0ba5 301 }
87de9ff5
JB
302 expect(
303 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
304 ).toStrictEqual({
932fc8be
JB
305 runTime: {
306 aggregate: true,
307 average: false,
308 median: true
309 },
310 waitTime: {
311 aggregate: false,
312 average: false,
313 median: false
314 },
5df69fab 315 elu: {
9adcefab 316 aggregate: true,
5df69fab 317 average: false,
9adcefab 318 median: true
5df69fab 319 }
86bf340d 320 })
9adcefab
JB
321 pool.setWorkerChoiceStrategyOptions({
322 runTime: { median: false },
323 elu: { median: false }
324 })
a20f0ba5 325 expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
9adcefab
JB
326 runTime: { median: false },
327 elu: { median: false }
a20f0ba5
JB
328 })
329 for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
330 .workerChoiceStrategies) {
932fc8be 331 expect(workerChoiceStrategy.opts).toStrictEqual({
9adcefab
JB
332 runTime: { median: false },
333 elu: { median: false }
932fc8be 334 })
a20f0ba5 335 }
87de9ff5
JB
336 expect(
337 pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
338 ).toStrictEqual({
932fc8be
JB
339 runTime: {
340 aggregate: true,
341 average: true,
342 median: false
343 },
344 waitTime: {
345 aggregate: false,
346 average: false,
347 median: false
348 },
5df69fab 349 elu: {
9adcefab
JB
350 aggregate: true,
351 average: true,
5df69fab
JB
352 median: false
353 }
86bf340d 354 })
1f95d544
JB
355 expect(() =>
356 pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
357 ).toThrowError(
358 'Invalid worker choice strategy options: must be a plain object'
359 )
360 expect(() =>
361 pool.setWorkerChoiceStrategyOptions({ weights: {} })
362 ).toThrowError(
363 'Invalid worker choice strategy options: must have a weight for each worker node'
364 )
365 expect(() =>
366 pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
367 ).toThrowError(
368 "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
369 )
a20f0ba5
JB
370 await pool.destroy()
371 })
372
2431bdb4 373 it('Verify that pool tasks queue can be enabled/disabled', async () => {
a20f0ba5
JB
374 const pool = new FixedThreadPool(
375 numberOfWorkers,
376 './tests/worker-files/thread/testWorker.js'
377 )
378 expect(pool.opts.enableTasksQueue).toBe(false)
379 expect(pool.opts.tasksQueueOptions).toBeUndefined()
380 pool.enableTasksQueue(true)
381 expect(pool.opts.enableTasksQueue).toBe(true)
382 expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
383 pool.enableTasksQueue(true, { concurrency: 2 })
384 expect(pool.opts.enableTasksQueue).toBe(true)
385 expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
386 pool.enableTasksQueue(false)
387 expect(pool.opts.enableTasksQueue).toBe(false)
388 expect(pool.opts.tasksQueueOptions).toBeUndefined()
389 await pool.destroy()
390 })
391
2431bdb4 392 it('Verify that pool tasks queue options can be set', async () => {
a20f0ba5
JB
393 const pool = new FixedThreadPool(
394 numberOfWorkers,
395 './tests/worker-files/thread/testWorker.js',
396 { enableTasksQueue: true }
397 )
398 expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
399 pool.setTasksQueueOptions({ concurrency: 2 })
400 expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
f0d7f803
JB
401 expect(() =>
402 pool.setTasksQueueOptions('invalidTasksQueueOptions')
403 ).toThrowError('Invalid tasks queue options: must be a plain object')
a20f0ba5
JB
404 expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
405 "Invalid worker tasks concurrency '0'"
406 )
f0d7f803
JB
407 expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
408 'Invalid worker tasks concurrency: must be an integer'
409 )
a20f0ba5
JB
410 await pool.destroy()
411 })
412
6b27d407
JB
413 it('Verify that pool info is set', async () => {
414 let pool = new FixedThreadPool(
415 numberOfWorkers,
416 './tests/worker-files/thread/testWorker.js'
417 )
418 expect(pool.info).toStrictEqual({
23ccf9d7 419 version,
6b27d407 420 type: PoolTypes.fixed,
184855e6 421 worker: WorkerTypes.thread,
2431bdb4
JB
422 ready: false,
423 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
6b27d407
JB
424 minSize: numberOfWorkers,
425 maxSize: numberOfWorkers,
426 workerNodes: numberOfWorkers,
427 idleWorkerNodes: numberOfWorkers,
428 busyWorkerNodes: 0,
a4e07f72
JB
429 executedTasks: 0,
430 executingTasks: 0,
6b27d407 431 queuedTasks: 0,
a4e07f72
JB
432 maxQueuedTasks: 0,
433 failedTasks: 0
6b27d407
JB
434 })
435 await pool.destroy()
436 pool = new DynamicClusterPool(
2431bdb4 437 Math.floor(numberOfWorkers / 2),
6b27d407 438 numberOfWorkers,
ecdfbdc0 439 './tests/worker-files/cluster/testWorker.js'
6b27d407
JB
440 )
441 expect(pool.info).toStrictEqual({
23ccf9d7 442 version,
6b27d407 443 type: PoolTypes.dynamic,
184855e6 444 worker: WorkerTypes.cluster,
2431bdb4
JB
445 ready: false,
446 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
447 minSize: Math.floor(numberOfWorkers / 2),
448 maxSize: numberOfWorkers,
449 workerNodes: Math.floor(numberOfWorkers / 2),
450 idleWorkerNodes: Math.floor(numberOfWorkers / 2),
6b27d407 451 busyWorkerNodes: 0,
a4e07f72
JB
452 executedTasks: 0,
453 executingTasks: 0,
6b27d407 454 queuedTasks: 0,
a4e07f72
JB
455 maxQueuedTasks: 0,
456 failedTasks: 0
6b27d407
JB
457 })
458 await pool.destroy()
459 })
460
2431bdb4 461 it('Verify that pool worker tasks usage are initialized', async () => {
bf9549ae
JB
462 const pool = new FixedClusterPool(
463 numberOfWorkers,
464 './tests/worker-files/cluster/testWorker.js'
465 )
f06e48d8 466 for (const workerNode of pool.workerNodes) {
465b2940 467 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
468 tasks: {
469 executed: 0,
470 executing: 0,
471 queued: 0,
df593701 472 maxQueued: 0,
a4e07f72
JB
473 failed: 0
474 },
475 runTime: {
a4e07f72
JB
476 history: expect.any(CircularArray)
477 },
478 waitTime: {
a4e07f72
JB
479 history: expect.any(CircularArray)
480 },
5df69fab
JB
481 elu: {
482 idle: {
5df69fab
JB
483 history: expect.any(CircularArray)
484 },
485 active: {
5df69fab 486 history: expect.any(CircularArray)
f7510105 487 }
5df69fab 488 }
86bf340d 489 })
f06e48d8
JB
490 }
491 await pool.destroy()
492 })
493
2431bdb4
JB
494 it('Verify that pool worker tasks queue are initialized', async () => {
495 let pool = new FixedClusterPool(
f06e48d8
JB
496 numberOfWorkers,
497 './tests/worker-files/cluster/testWorker.js'
498 )
499 for (const workerNode of pool.workerNodes) {
500 expect(workerNode.tasksQueue).toBeDefined()
29ee7e9a 501 expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
4d8bf9e4 502 expect(workerNode.tasksQueue.size).toBe(0)
9c16fb4b 503 expect(workerNode.tasksQueue.maxSize).toBe(0)
bf9549ae 504 }
fd7ebd49 505 await pool.destroy()
2431bdb4
JB
506 pool = new DynamicThreadPool(
507 Math.floor(numberOfWorkers / 2),
508 numberOfWorkers,
509 './tests/worker-files/thread/testWorker.js'
510 )
511 for (const workerNode of pool.workerNodes) {
512 expect(workerNode.tasksQueue).toBeDefined()
513 expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
514 expect(workerNode.tasksQueue.size).toBe(0)
515 expect(workerNode.tasksQueue.maxSize).toBe(0)
516 }
517 })
518
519 it('Verify that pool worker info are initialized', async () => {
520 let pool = new FixedClusterPool(
521 numberOfWorkers,
522 './tests/worker-files/cluster/testWorker.js'
523 )
524 for (const workerNode of pool.workerNodes) {
525 expect(workerNode.info).toStrictEqual({
526 id: expect.any(Number),
527 type: WorkerTypes.cluster,
528 dynamic: false,
529 ready: false
530 })
531 }
532 await pool.destroy()
533 pool = new DynamicThreadPool(
534 Math.floor(numberOfWorkers / 2),
535 numberOfWorkers,
536 './tests/worker-files/thread/testWorker.js'
537 )
538 for (const workerNode of pool.workerNodes) {
539 expect(workerNode.info).toStrictEqual({
540 id: expect.any(Number),
541 type: WorkerTypes.thread,
542 dynamic: false,
543 ready: false
544 })
545 }
bf9549ae
JB
546 })
547
2431bdb4 548 it('Verify that pool worker tasks usage are computed', async () => {
bf9549ae
JB
549 const pool = new FixedClusterPool(
550 numberOfWorkers,
551 './tests/worker-files/cluster/testWorker.js'
552 )
09c2d0d3 553 const promises = new Set()
fc027381
JB
554 const maxMultiplier = 2
555 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 556 promises.add(pool.execute())
bf9549ae 557 }
f06e48d8 558 for (const workerNode of pool.workerNodes) {
465b2940 559 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
560 tasks: {
561 executed: 0,
562 executing: maxMultiplier,
563 queued: 0,
df593701 564 maxQueued: 0,
a4e07f72
JB
565 failed: 0
566 },
567 runTime: {
a4e07f72
JB
568 history: expect.any(CircularArray)
569 },
570 waitTime: {
a4e07f72
JB
571 history: expect.any(CircularArray)
572 },
5df69fab
JB
573 elu: {
574 idle: {
5df69fab
JB
575 history: expect.any(CircularArray)
576 },
577 active: {
5df69fab 578 history: expect.any(CircularArray)
f7510105 579 }
5df69fab 580 }
86bf340d 581 })
bf9549ae
JB
582 }
583 await Promise.all(promises)
f06e48d8 584 for (const workerNode of pool.workerNodes) {
465b2940 585 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
586 tasks: {
587 executed: maxMultiplier,
588 executing: 0,
589 queued: 0,
df593701 590 maxQueued: 0,
a4e07f72
JB
591 failed: 0
592 },
593 runTime: {
a4e07f72
JB
594 history: expect.any(CircularArray)
595 },
596 waitTime: {
a4e07f72
JB
597 history: expect.any(CircularArray)
598 },
5df69fab
JB
599 elu: {
600 idle: {
5df69fab
JB
601 history: expect.any(CircularArray)
602 },
603 active: {
5df69fab 604 history: expect.any(CircularArray)
f7510105 605 }
5df69fab 606 }
86bf340d 607 })
bf9549ae 608 }
fd7ebd49 609 await pool.destroy()
bf9549ae
JB
610 })
611
2431bdb4 612 it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
7fd82a1c 613 const pool = new DynamicThreadPool(
2431bdb4 614 Math.floor(numberOfWorkers / 2),
8f4878b7 615 numberOfWorkers,
9e619829
JB
616 './tests/worker-files/thread/testWorker.js'
617 )
09c2d0d3 618 const promises = new Set()
ee9f5295
JB
619 const maxMultiplier = 2
620 for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
09c2d0d3 621 promises.add(pool.execute())
9e619829
JB
622 }
623 await Promise.all(promises)
f06e48d8 624 for (const workerNode of pool.workerNodes) {
465b2940 625 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
626 tasks: {
627 executed: expect.any(Number),
628 executing: 0,
629 queued: 0,
df593701 630 maxQueued: 0,
a4e07f72
JB
631 failed: 0
632 },
633 runTime: {
a4e07f72
JB
634 history: expect.any(CircularArray)
635 },
636 waitTime: {
a4e07f72
JB
637 history: expect.any(CircularArray)
638 },
5df69fab
JB
639 elu: {
640 idle: {
5df69fab
JB
641 history: expect.any(CircularArray)
642 },
643 active: {
5df69fab 644 history: expect.any(CircularArray)
f7510105 645 }
5df69fab 646 }
86bf340d 647 })
465b2940
JB
648 expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
649 expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(maxMultiplier)
9e619829
JB
650 }
651 pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
f06e48d8 652 for (const workerNode of pool.workerNodes) {
465b2940 653 expect(workerNode.usage).toStrictEqual({
a4e07f72
JB
654 tasks: {
655 executed: 0,
656 executing: 0,
657 queued: 0,
df593701 658 maxQueued: 0,
a4e07f72
JB
659 failed: 0
660 },
661 runTime: {
a4e07f72
JB
662 history: expect.any(CircularArray)
663 },
664 waitTime: {
a4e07f72
JB
665 history: expect.any(CircularArray)
666 },
5df69fab
JB
667 elu: {
668 idle: {
5df69fab
JB
669 history: expect.any(CircularArray)
670 },
671 active: {
5df69fab 672 history: expect.any(CircularArray)
f7510105 673 }
5df69fab 674 }
86bf340d 675 })
465b2940
JB
676 expect(workerNode.usage.runTime.history.length).toBe(0)
677 expect(workerNode.usage.waitTime.history.length).toBe(0)
ee11a4a2 678 }
fd7ebd49 679 await pool.destroy()
ee11a4a2
JB
680 })
681
164d950a
JB
682 it("Verify that pool event emitter 'full' event can register a callback", async () => {
683 const pool = new DynamicThreadPool(
2431bdb4 684 Math.floor(numberOfWorkers / 2),
164d950a
JB
685 numberOfWorkers,
686 './tests/worker-files/thread/testWorker.js'
687 )
09c2d0d3 688 const promises = new Set()
164d950a 689 let poolFull = 0
d46660cd
JB
690 let poolInfo
691 pool.emitter.on(PoolEvents.full, info => {
692 ++poolFull
693 poolInfo = info
694 })
164d950a 695 for (let i = 0; i < numberOfWorkers * 2; i++) {
f5d14e90 696 promises.add(pool.execute())
164d950a
JB
697 }
698 await Promise.all(promises)
2431bdb4
JB
699 // The `full` event is triggered when the number of submitted tasks at once reach the maximum number of workers in the dynamic pool.
700 // So in total numberOfWorkers * 2 - 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool with min = (max = numberOfWorkers) / 2.
701 expect(poolFull).toBe(numberOfWorkers * 2 - 1)
d46660cd 702 expect(poolInfo).toStrictEqual({
23ccf9d7 703 version,
d46660cd
JB
704 type: PoolTypes.dynamic,
705 worker: WorkerTypes.thread,
2431bdb4
JB
706 ready: expect.any(Boolean),
707 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
708 minSize: expect.any(Number),
709 maxSize: expect.any(Number),
710 workerNodes: expect.any(Number),
711 idleWorkerNodes: expect.any(Number),
712 busyWorkerNodes: expect.any(Number),
713 executedTasks: expect.any(Number),
714 executingTasks: expect.any(Number),
715 queuedTasks: expect.any(Number),
716 maxQueuedTasks: expect.any(Number),
717 failedTasks: expect.any(Number)
718 })
719 await pool.destroy()
720 })
721
722 it("Verify that pool event emitter 'ready' event can register a callback", async () => {
723 const pool = new FixedClusterPool(
724 numberOfWorkers,
725 './tests/worker-files/cluster/testWorker.js'
726 )
727 let poolReady = 0
728 let poolInfo
729 pool.emitter.on(PoolEvents.ready, info => {
730 ++poolReady
731 poolInfo = info
732 })
733 await waitPoolEvents(pool, PoolEvents.ready, 1)
734 expect(poolReady).toBe(1)
735 expect(poolInfo).toStrictEqual({
736 version,
737 type: PoolTypes.fixed,
738 worker: WorkerTypes.cluster,
739 ready: true,
740 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
741 minSize: expect.any(Number),
742 maxSize: expect.any(Number),
743 workerNodes: expect.any(Number),
744 idleWorkerNodes: expect.any(Number),
745 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
746 executedTasks: expect.any(Number),
747 executingTasks: expect.any(Number),
d46660cd 748 queuedTasks: expect.any(Number),
a4e07f72
JB
749 maxQueuedTasks: expect.any(Number),
750 failedTasks: expect.any(Number)
d46660cd 751 })
164d950a
JB
752 await pool.destroy()
753 })
754
cf597bc5 755 it("Verify that pool event emitter 'busy' event can register a callback", async () => {
7c0ba920
JB
756 const pool = new FixedThreadPool(
757 numberOfWorkers,
758 './tests/worker-files/thread/testWorker.js'
759 )
09c2d0d3 760 const promises = new Set()
7c0ba920 761 let poolBusy = 0
d46660cd
JB
762 let poolInfo
763 pool.emitter.on(PoolEvents.busy, info => {
764 ++poolBusy
765 poolInfo = info
766 })
7c0ba920 767 for (let i = 0; i < numberOfWorkers * 2; i++) {
f5d14e90 768 promises.add(pool.execute())
7c0ba920 769 }
cf597bc5 770 await Promise.all(promises)
14916bf9
JB
771 // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
772 // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
773 expect(poolBusy).toBe(numberOfWorkers + 1)
d46660cd 774 expect(poolInfo).toStrictEqual({
23ccf9d7 775 version,
d46660cd
JB
776 type: PoolTypes.fixed,
777 worker: WorkerTypes.thread,
2431bdb4
JB
778 ready: expect.any(Boolean),
779 strategy: WorkerChoiceStrategies.ROUND_ROBIN,
d46660cd
JB
780 minSize: expect.any(Number),
781 maxSize: expect.any(Number),
782 workerNodes: expect.any(Number),
783 idleWorkerNodes: expect.any(Number),
784 busyWorkerNodes: expect.any(Number),
a4e07f72
JB
785 executedTasks: expect.any(Number),
786 executingTasks: expect.any(Number),
d46660cd 787 queuedTasks: expect.any(Number),
a4e07f72
JB
788 maxQueuedTasks: expect.any(Number),
789 failedTasks: expect.any(Number)
d46660cd 790 })
fd7ebd49 791 await pool.destroy()
7c0ba920 792 })
70a4f5ea
JB
793
794 it('Verify that multiple tasks worker is working', async () => {
795 const pool = new DynamicClusterPool(
2431bdb4 796 Math.floor(numberOfWorkers / 2),
70a4f5ea 797 numberOfWorkers,
70a4f5ea
JB
798 './tests/worker-files/cluster/testMultiTasksWorker.js'
799 )
800 const data = { n: 10 }
82888165 801 const result0 = await pool.execute(data)
30b963d4 802 expect(result0).toStrictEqual({ ok: 1 })
70a4f5ea 803 const result1 = await pool.execute(data, 'jsonIntegerSerialization')
30b963d4 804 expect(result1).toStrictEqual({ ok: 1 })
70a4f5ea
JB
805 const result2 = await pool.execute(data, 'factorial')
806 expect(result2).toBe(3628800)
807 const result3 = await pool.execute(data, 'fibonacci')
024daf59 808 expect(result3).toBe(55)
70a4f5ea 809 })
3ec964d6 810})