pools/abstract/abstract-pool.test.js

   1 const { expect } = require('expect')
   2 const sinon = require('sinon')
   3 const {
   4   DynamicClusterPool,
   5   DynamicThreadPool,
   6   FixedClusterPool,
   7   FixedThreadPool,
   8   PoolEvents,
   9   PoolTypes,
  10   WorkerChoiceStrategies,
  11   WorkerTypes
  12 } = require('../../../lib')
  13 const { CircularArray } = require('../../../lib/circular-array')
  14 const { Deque } = require('../../../lib/deque')
  15 const { version } = require('../../../package.json')
  16 const { waitPoolEvents } = require('../../test-utils')
  17
  18 describe('Abstract pool test suite', () => {
  19   const numberOfWorkers = 2
  20   class StubPoolWithIsMain extends FixedThreadPool {
  21     isMain () {
  22       return false
  23     }
  24   }
  25
  26   afterEach(() => {
  27     sinon.restore()
  28   })
  29
  30   it('Simulate pool creation from a non main thread/process', () => {
  31     expect(
  32       () =>
  33         new StubPoolWithIsMain(
  34           numberOfWorkers,
  35           './tests/worker-files/thread/testWorker.js',
  36           {
  37             errorHandler: (e) => console.error(e)
  38           }
  39         )
  40     ).toThrowError(
  41       new Error(
  42         'Cannot start a pool from a worker with the same type as the pool'
  43       )
  44     )
  45   })
  46
  47   it('Verify that pool statuses properties are set', async () => {
  48     const pool = new FixedThreadPool(
  49       numberOfWorkers,
  50       './tests/worker-files/thread/testWorker.js'
  51     )
  52     expect(pool.starting).toBe(false)
  53     expect(pool.started).toBe(true)
  54     await pool.destroy()
  55     expect(pool.started).toBe(false)
  56   })
  57
  58   it('Verify that filePath is checked', () => {
  59     const expectedError = new Error(
  60       'Please specify a file with a worker implementation'
  61     )
  62     expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError(
  63       expectedError
  64     )
  65     expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError(
  66       expectedError
  67     )
  68     expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrowError(
  69       expectedError
  70     )
  71     expect(() => new FixedThreadPool(numberOfWorkers, true)).toThrowError(
  72       expectedError
  73     )
  74     expect(
  75       () => new FixedThreadPool(numberOfWorkers, './dummyWorker.ts')
  76     ).toThrowError(new Error("Cannot find the worker file './dummyWorker.ts'"))
  77   })
  78
  79   it('Verify that numberOfWorkers is checked', () => {
  80     expect(() => new FixedThreadPool()).toThrowError(
  81       new Error(
  82         'Cannot instantiate a pool without specifying the number of workers'
  83       )
  84     )
  85   })
  86
  87   it('Verify that a negative number of workers is checked', () => {
  88     expect(
  89       () =>
  90         new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
  91     ).toThrowError(
  92       new RangeError(
  93         'Cannot instantiate a pool with a negative number of workers'
  94       )
  95     )
  96   })
  97
  98   it('Verify that a non integer number of workers is checked', () => {
  99     expect(
 100       () =>
 101         new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.js')
 102     ).toThrowError(
 103       new TypeError(
 104         'Cannot instantiate a pool with a non safe integer number of workers'
 105       )
 106     )
 107   })
 108
 109   it('Verify that dynamic pool sizing is checked', () => {
 110     expect(
 111       () =>
 112         new DynamicClusterPool(
 113           1,
 114           undefined,
 115           './tests/worker-files/cluster/testWorker.js'
 116         )
 117     ).toThrowError(
 118       new TypeError(
 119         'Cannot instantiate a dynamic pool without specifying the maximum pool size'
 120       )
 121     )
 122     expect(
 123       () =>
 124         new DynamicThreadPool(
 125           0.5,
 126           1,
 127           './tests/worker-files/thread/testWorker.js'
 128         )
 129     ).toThrowError(
 130       new TypeError(
 131         'Cannot instantiate a pool with a non safe integer number of workers'
 132       )
 133     )
 134     expect(
 135       () =>
 136         new DynamicClusterPool(
 137           0,
 138           0.5,
 139           './tests/worker-files/cluster/testWorker.js'
 140         )
 141     ).toThrowError(
 142       new TypeError(
 143         'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
 144       )
 145     )
 146     expect(
 147       () =>
 148         new DynamicThreadPool(2, 1, './tests/worker-files/thread/testWorker.js')
 149     ).toThrowError(
 150       new RangeError(
 151         'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
 152       )
 153     )
 154     expect(
 155       () =>
 156         new DynamicClusterPool(
 157           1,
 158           1,
 159           './tests/worker-files/cluster/testWorker.js'
 160         )
 161     ).toThrowError(
 162       new RangeError(
 163         'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
 164       )
 165     )
 166     expect(
 167       () =>
 168         new DynamicThreadPool(0, 0, './tests/worker-files/thread/testWorker.js')
 169     ).toThrowError(
 170       new RangeError(
 171         'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
 172       )
 173     )
 174   })
 175
 176   it('Verify that pool options are checked', async () => {
 177     let pool = new FixedThreadPool(
 178       numberOfWorkers,
 179       './tests/worker-files/thread/testWorker.js'
 180     )
 181     expect(pool.emitter).toBeDefined()
 182     expect(pool.opts.enableEvents).toBe(true)
 183     expect(pool.opts.restartWorkerOnError).toBe(true)
 184     expect(pool.opts.enableTasksQueue).toBe(false)
 185     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 186     expect(pool.opts.workerChoiceStrategy).toBe(
 187       WorkerChoiceStrategies.ROUND_ROBIN
 188     )
 189     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 190       choiceRetries: 6,
 191       runTime: { median: false },
 192       waitTime: { median: false },
 193       elu: { median: false }
 194     })
 195     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 196       choiceRetries: 6,
 197       runTime: { median: false },
 198       waitTime: { median: false },
 199       elu: { median: false }
 200     })
 201     expect(pool.opts.messageHandler).toBeUndefined()
 202     expect(pool.opts.errorHandler).toBeUndefined()
 203     expect(pool.opts.onlineHandler).toBeUndefined()
 204     expect(pool.opts.exitHandler).toBeUndefined()
 205     await pool.destroy()
 206     const testHandler = () => console.info('test handler executed')
 207     pool = new FixedThreadPool(
 208       numberOfWorkers,
 209       './tests/worker-files/thread/testWorker.js',
 210       {
 211         workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
 212         workerChoiceStrategyOptions: {
 213           runTime: { median: true },
 214           weights: { 0: 300, 1: 200 }
 215         },
 216         enableEvents: false,
 217         restartWorkerOnError: false,
 218         enableTasksQueue: true,
 219         tasksQueueOptions: { concurrency: 2 },
 220         messageHandler: testHandler,
 221         errorHandler: testHandler,
 222         onlineHandler: testHandler,
 223         exitHandler: testHandler
 224       }
 225     )
 226     expect(pool.emitter).toBeUndefined()
 227     expect(pool.opts.enableEvents).toBe(false)
 228     expect(pool.opts.restartWorkerOnError).toBe(false)
 229     expect(pool.opts.enableTasksQueue).toBe(true)
 230     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 231       concurrency: 2,
 232       size: 4
 233     })
 234     expect(pool.opts.workerChoiceStrategy).toBe(
 235       WorkerChoiceStrategies.LEAST_USED
 236     )
 237     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 238       choiceRetries: 6,
 239       runTime: { median: true },
 240       waitTime: { median: false },
 241       elu: { median: false },
 242       weights: { 0: 300, 1: 200 }
 243     })
 244     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 245       choiceRetries: 6,
 246       runTime: { median: true },
 247       waitTime: { median: false },
 248       elu: { median: false },
 249       weights: { 0: 300, 1: 200 }
 250     })
 251     expect(pool.opts.messageHandler).toStrictEqual(testHandler)
 252     expect(pool.opts.errorHandler).toStrictEqual(testHandler)
 253     expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
 254     expect(pool.opts.exitHandler).toStrictEqual(testHandler)
 255     await pool.destroy()
 256   })
 257
 258   it('Verify that pool options are validated', async () => {
 259     expect(
 260       () =>
 261         new FixedThreadPool(
 262           numberOfWorkers,
 263           './tests/worker-files/thread/testWorker.js',
 264           {
 265             workerChoiceStrategy: 'invalidStrategy'
 266           }
 267         )
 268     ).toThrowError(
 269       new Error("Invalid worker choice strategy 'invalidStrategy'")
 270     )
 271     expect(
 272       () =>
 273         new FixedThreadPool(
 274           numberOfWorkers,
 275           './tests/worker-files/thread/testWorker.js',
 276           {
 277             workerChoiceStrategyOptions: {
 278               choiceRetries: 'invalidChoiceRetries'
 279             }
 280           }
 281         )
 282     ).toThrowError(
 283       new TypeError(
 284         'Invalid worker choice strategy options: choice retries must be an integer'
 285       )
 286     )
 287     expect(
 288       () =>
 289         new FixedThreadPool(
 290           numberOfWorkers,
 291           './tests/worker-files/thread/testWorker.js',
 292           {
 293             workerChoiceStrategyOptions: {
 294               choiceRetries: -1
 295             }
 296           }
 297         )
 298     ).toThrowError(
 299       new RangeError(
 300         "Invalid worker choice strategy options: choice retries '-1' must be greater or equal than zero"
 301       )
 302     )
 303     expect(
 304       () =>
 305         new FixedThreadPool(
 306           numberOfWorkers,
 307           './tests/worker-files/thread/testWorker.js',
 308           {
 309             workerChoiceStrategyOptions: { weights: {} }
 310           }
 311         )
 312     ).toThrowError(
 313       new Error(
 314         'Invalid worker choice strategy options: must have a weight for each worker node'
 315       )
 316     )
 317     expect(
 318       () =>
 319         new FixedThreadPool(
 320           numberOfWorkers,
 321           './tests/worker-files/thread/testWorker.js',
 322           {
 323             workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
 324           }
 325         )
 326     ).toThrowError(
 327       new Error(
 328         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 329       )
 330     )
 331     expect(
 332       () =>
 333         new FixedThreadPool(
 334           numberOfWorkers,
 335           './tests/worker-files/thread/testWorker.js',
 336           {
 337             enableTasksQueue: true,
 338             tasksQueueOptions: 'invalidTasksQueueOptions'
 339           }
 340         )
 341     ).toThrowError(
 342       new TypeError('Invalid tasks queue options: must be a plain object')
 343     )
 344     expect(
 345       () =>
 346         new FixedThreadPool(
 347           numberOfWorkers,
 348           './tests/worker-files/thread/testWorker.js',
 349           {
 350             enableTasksQueue: true,
 351             tasksQueueOptions: { concurrency: 0 }
 352           }
 353         )
 354     ).toThrowError(
 355       new RangeError(
 356         'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
 357       )
 358     )
 359     expect(
 360       () =>
 361         new FixedThreadPool(
 362           numberOfWorkers,
 363           './tests/worker-files/thread/testWorker.js',
 364           {
 365             enableTasksQueue: true,
 366             tasksQueueOptions: { concurrency: -1 }
 367           }
 368         )
 369     ).toThrowError(
 370       new RangeError(
 371         'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
 372       )
 373     )
 374     expect(
 375       () =>
 376         new FixedThreadPool(
 377           numberOfWorkers,
 378           './tests/worker-files/thread/testWorker.js',
 379           {
 380             enableTasksQueue: true,
 381             tasksQueueOptions: { concurrency: 0.2 }
 382           }
 383         )
 384     ).toThrowError(
 385       new TypeError('Invalid worker node tasks concurrency: must be an integer')
 386     )
 387     expect(
 388       () =>
 389         new FixedThreadPool(
 390           numberOfWorkers,
 391           './tests/worker-files/thread/testWorker.js',
 392           {
 393             enableTasksQueue: true,
 394             tasksQueueOptions: { queueMaxSize: 2 }
 395           }
 396         )
 397     ).toThrowError(
 398       new Error(
 399         'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
 400       )
 401     )
 402     expect(
 403       () =>
 404         new FixedThreadPool(
 405           numberOfWorkers,
 406           './tests/worker-files/thread/testWorker.js',
 407           {
 408             enableTasksQueue: true,
 409             tasksQueueOptions: { size: 0 }
 410           }
 411         )
 412     ).toThrowError(
 413       new RangeError(
 414         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
 415       )
 416     )
 417     expect(
 418       () =>
 419         new FixedThreadPool(
 420           numberOfWorkers,
 421           './tests/worker-files/thread/testWorker.js',
 422           {
 423             enableTasksQueue: true,
 424             tasksQueueOptions: { size: -1 }
 425           }
 426         )
 427     ).toThrowError(
 428       new RangeError(
 429         'Invalid worker node tasks queue size: -1 is a negative integer or zero'
 430       )
 431     )
 432     expect(
 433       () =>
 434         new FixedThreadPool(
 435           numberOfWorkers,
 436           './tests/worker-files/thread/testWorker.js',
 437           {
 438             enableTasksQueue: true,
 439             tasksQueueOptions: { size: 0.2 }
 440           }
 441         )
 442     ).toThrowError(
 443       new TypeError('Invalid worker node tasks queue size: must be an integer')
 444     )
 445   })
 446
 447   it('Verify that pool worker choice strategy options can be set', async () => {
 448     const pool = new FixedThreadPool(
 449       numberOfWorkers,
 450       './tests/worker-files/thread/testWorker.js',
 451       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 452     )
 453     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 454       choiceRetries: 6,
 455       runTime: { median: false },
 456       waitTime: { median: false },
 457       elu: { median: false }
 458     })
 459     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 460       choiceRetries: 6,
 461       runTime: { median: false },
 462       waitTime: { median: false },
 463       elu: { median: false }
 464     })
 465     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 466       .workerChoiceStrategies) {
 467       expect(workerChoiceStrategy.opts).toStrictEqual({
 468         choiceRetries: 6,
 469         runTime: { median: false },
 470         waitTime: { median: false },
 471         elu: { median: false }
 472       })
 473     }
 474     expect(
 475       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 476     ).toStrictEqual({
 477       runTime: {
 478         aggregate: true,
 479         average: true,
 480         median: false
 481       },
 482       waitTime: {
 483         aggregate: false,
 484         average: false,
 485         median: false
 486       },
 487       elu: {
 488         aggregate: true,
 489         average: true,
 490         median: false
 491       }
 492     })
 493     pool.setWorkerChoiceStrategyOptions({
 494       runTime: { median: true },
 495       elu: { median: true }
 496     })
 497     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 498       choiceRetries: 6,
 499       runTime: { median: true },
 500       waitTime: { median: false },
 501       elu: { median: true }
 502     })
 503     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 504       choiceRetries: 6,
 505       runTime: { median: true },
 506       waitTime: { median: false },
 507       elu: { median: true }
 508     })
 509     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 510       .workerChoiceStrategies) {
 511       expect(workerChoiceStrategy.opts).toStrictEqual({
 512         choiceRetries: 6,
 513         runTime: { median: true },
 514         waitTime: { median: false },
 515         elu: { median: true }
 516       })
 517     }
 518     expect(
 519       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 520     ).toStrictEqual({
 521       runTime: {
 522         aggregate: true,
 523         average: false,
 524         median: true
 525       },
 526       waitTime: {
 527         aggregate: false,
 528         average: false,
 529         median: false
 530       },
 531       elu: {
 532         aggregate: true,
 533         average: false,
 534         median: true
 535       }
 536     })
 537     pool.setWorkerChoiceStrategyOptions({
 538       runTime: { median: false },
 539       elu: { median: false }
 540     })
 541     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 542       choiceRetries: 6,
 543       runTime: { median: false },
 544       waitTime: { median: false },
 545       elu: { median: false }
 546     })
 547     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 548       choiceRetries: 6,
 549       runTime: { median: false },
 550       waitTime: { median: false },
 551       elu: { median: false }
 552     })
 553     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 554       .workerChoiceStrategies) {
 555       expect(workerChoiceStrategy.opts).toStrictEqual({
 556         choiceRetries: 6,
 557         runTime: { median: false },
 558         waitTime: { median: false },
 559         elu: { median: false }
 560       })
 561     }
 562     expect(
 563       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 564     ).toStrictEqual({
 565       runTime: {
 566         aggregate: true,
 567         average: true,
 568         median: false
 569       },
 570       waitTime: {
 571         aggregate: false,
 572         average: false,
 573         median: false
 574       },
 575       elu: {
 576         aggregate: true,
 577         average: true,
 578         median: false
 579       }
 580     })
 581     expect(() =>
 582       pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
 583     ).toThrowError(
 584       new TypeError(
 585         'Invalid worker choice strategy options: must be a plain object'
 586       )
 587     )
 588     expect(() =>
 589       pool.setWorkerChoiceStrategyOptions({
 590         choiceRetries: 'invalidChoiceRetries'
 591       })
 592     ).toThrowError(
 593       new TypeError(
 594         'Invalid worker choice strategy options: choice retries must be an integer'
 595       )
 596     )
 597     expect(() =>
 598       pool.setWorkerChoiceStrategyOptions({ choiceRetries: -1 })
 599     ).toThrowError(
 600       new RangeError(
 601         "Invalid worker choice strategy options: choice retries '-1' must be greater or equal than zero"
 602       )
 603     )
 604     expect(() =>
 605       pool.setWorkerChoiceStrategyOptions({ weights: {} })
 606     ).toThrowError(
 607       new Error(
 608         'Invalid worker choice strategy options: must have a weight for each worker node'
 609       )
 610     )
 611     expect(() =>
 612       pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
 613     ).toThrowError(
 614       new Error(
 615         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 616       )
 617     )
 618     await pool.destroy()
 619   })
 620
 621   it('Verify that pool tasks queue can be enabled/disabled', async () => {
 622     const pool = new FixedThreadPool(
 623       numberOfWorkers,
 624       './tests/worker-files/thread/testWorker.js'
 625     )
 626     expect(pool.opts.enableTasksQueue).toBe(false)
 627     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 628     pool.enableTasksQueue(true)
 629     expect(pool.opts.enableTasksQueue).toBe(true)
 630     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 631       concurrency: 1,
 632       size: 4
 633     })
 634     pool.enableTasksQueue(true, { concurrency: 2 })
 635     expect(pool.opts.enableTasksQueue).toBe(true)
 636     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 637       concurrency: 2,
 638       size: 4
 639     })
 640     pool.enableTasksQueue(false)
 641     expect(pool.opts.enableTasksQueue).toBe(false)
 642     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 643     await pool.destroy()
 644   })
 645
 646   it('Verify that pool tasks queue options can be set', async () => {
 647     const pool = new FixedThreadPool(
 648       numberOfWorkers,
 649       './tests/worker-files/thread/testWorker.js',
 650       { enableTasksQueue: true }
 651     )
 652     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 653       concurrency: 1,
 654       size: 4
 655     })
 656     pool.setTasksQueueOptions({ concurrency: 2 })
 657     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 658       concurrency: 2,
 659       size: 4
 660     })
 661     expect(() =>
 662       pool.setTasksQueueOptions('invalidTasksQueueOptions')
 663     ).toThrowError(
 664       new TypeError('Invalid tasks queue options: must be a plain object')
 665     )
 666     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
 667       new RangeError(
 668         'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
 669       )
 670     )
 671     expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError(
 672       new RangeError(
 673         'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
 674       )
 675     )
 676     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
 677       new TypeError('Invalid worker node tasks concurrency: must be an integer')
 678     )
 679     expect(() => pool.setTasksQueueOptions({ queueMaxSize: 2 })).toThrowError(
 680       new Error(
 681         'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
 682       )
 683     )
 684     expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
 685       new RangeError(
 686         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
 687       )
 688     )
 689     expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError(
 690       new RangeError(
 691         'Invalid worker node tasks queue size: -1 is a negative integer or zero'
 692       )
 693     )
 694     expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError(
 695       new TypeError('Invalid worker node tasks queue size: must be an integer')
 696     )
 697     await pool.destroy()
 698   })
 699
 700   it('Verify that pool info is set', async () => {
 701     let pool = new FixedThreadPool(
 702       numberOfWorkers,
 703       './tests/worker-files/thread/testWorker.js'
 704     )
 705     expect(pool.info).toStrictEqual({
 706       version,
 707       type: PoolTypes.fixed,
 708       worker: WorkerTypes.thread,
 709       ready: true,
 710       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 711       minSize: numberOfWorkers,
 712       maxSize: numberOfWorkers,
 713       workerNodes: numberOfWorkers,
 714       idleWorkerNodes: numberOfWorkers,
 715       busyWorkerNodes: 0,
 716       executedTasks: 0,
 717       executingTasks: 0,
 718       failedTasks: 0
 719     })
 720     await pool.destroy()
 721     pool = new DynamicClusterPool(
 722       Math.floor(numberOfWorkers / 2),
 723       numberOfWorkers,
 724       './tests/worker-files/cluster/testWorker.js'
 725     )
 726     expect(pool.info).toStrictEqual({
 727       version,
 728       type: PoolTypes.dynamic,
 729       worker: WorkerTypes.cluster,
 730       ready: true,
 731       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 732       minSize: Math.floor(numberOfWorkers / 2),
 733       maxSize: numberOfWorkers,
 734       workerNodes: Math.floor(numberOfWorkers / 2),
 735       idleWorkerNodes: Math.floor(numberOfWorkers / 2),
 736       busyWorkerNodes: 0,
 737       executedTasks: 0,
 738       executingTasks: 0,
 739       failedTasks: 0
 740     })
 741     await pool.destroy()
 742   })
 743
 744   it('Verify that pool worker tasks usage are initialized', async () => {
 745     const pool = new FixedClusterPool(
 746       numberOfWorkers,
 747       './tests/worker-files/cluster/testWorker.js'
 748     )
 749     for (const workerNode of pool.workerNodes) {
 750       expect(workerNode.usage).toStrictEqual({
 751         tasks: {
 752           executed: 0,
 753           executing: 0,
 754           queued: 0,
 755           maxQueued: 0,
 756           stolen: 0,
 757           failed: 0
 758         },
 759         runTime: {
 760           history: expect.any(CircularArray)
 761         },
 762         waitTime: {
 763           history: expect.any(CircularArray)
 764         },
 765         elu: {
 766           idle: {
 767             history: expect.any(CircularArray)
 768           },
 769           active: {
 770             history: expect.any(CircularArray)
 771           }
 772         }
 773       })
 774     }
 775     await pool.destroy()
 776   })
 777
 778   it('Verify that pool worker tasks queue are initialized', async () => {
 779     let pool = new FixedClusterPool(
 780       numberOfWorkers,
 781       './tests/worker-files/cluster/testWorker.js'
 782     )
 783     for (const workerNode of pool.workerNodes) {
 784       expect(workerNode.tasksQueue).toBeDefined()
 785       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 786       expect(workerNode.tasksQueue.size).toBe(0)
 787       expect(workerNode.tasksQueue.maxSize).toBe(0)
 788     }
 789     await pool.destroy()
 790     pool = new DynamicThreadPool(
 791       Math.floor(numberOfWorkers / 2),
 792       numberOfWorkers,
 793       './tests/worker-files/thread/testWorker.js'
 794     )
 795     for (const workerNode of pool.workerNodes) {
 796       expect(workerNode.tasksQueue).toBeDefined()
 797       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 798       expect(workerNode.tasksQueue.size).toBe(0)
 799       expect(workerNode.tasksQueue.maxSize).toBe(0)
 800     }
 801   })
 802
 803   it('Verify that pool worker info are initialized', async () => {
 804     let pool = new FixedClusterPool(
 805       numberOfWorkers,
 806       './tests/worker-files/cluster/testWorker.js'
 807     )
 808     for (const workerNode of pool.workerNodes) {
 809       expect(workerNode.info).toStrictEqual({
 810         id: expect.any(Number),
 811         type: WorkerTypes.cluster,
 812         dynamic: false,
 813         ready: true
 814       })
 815     }
 816     await pool.destroy()
 817     pool = new DynamicThreadPool(
 818       Math.floor(numberOfWorkers / 2),
 819       numberOfWorkers,
 820       './tests/worker-files/thread/testWorker.js'
 821     )
 822     for (const workerNode of pool.workerNodes) {
 823       expect(workerNode.info).toStrictEqual({
 824         id: expect.any(Number),
 825         type: WorkerTypes.thread,
 826         dynamic: false,
 827         ready: true
 828       })
 829     }
 830   })
 831
 832   it('Verify that pool execute() arguments are checked', async () => {
 833     const pool = new FixedClusterPool(
 834       numberOfWorkers,
 835       './tests/worker-files/cluster/testWorker.js'
 836     )
 837     await expect(pool.execute(undefined, 0)).rejects.toThrowError(
 838       new TypeError('name argument must be a string')
 839     )
 840     await expect(pool.execute(undefined, '')).rejects.toThrowError(
 841       new TypeError('name argument must not be an empty string')
 842     )
 843     await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
 844       new TypeError('transferList argument must be an array')
 845     )
 846     await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
 847       "Task function 'unknown' not found"
 848     )
 849     await pool.destroy()
 850     await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
 851       new Error('Cannot execute a task on destroyed pool')
 852     )
 853   })
 854
 855   it('Verify that pool worker tasks usage are computed', async () => {
 856     const pool = new FixedClusterPool(
 857       numberOfWorkers,
 858       './tests/worker-files/cluster/testWorker.js'
 859     )
 860     const promises = new Set()
 861     const maxMultiplier = 2
 862     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 863       promises.add(pool.execute())
 864     }
 865     for (const workerNode of pool.workerNodes) {
 866       expect(workerNode.usage).toStrictEqual({
 867         tasks: {
 868           executed: 0,
 869           executing: maxMultiplier,
 870           queued: 0,
 871           maxQueued: 0,
 872           stolen: 0,
 873           failed: 0
 874         },
 875         runTime: {
 876           history: expect.any(CircularArray)
 877         },
 878         waitTime: {
 879           history: expect.any(CircularArray)
 880         },
 881         elu: {
 882           idle: {
 883             history: expect.any(CircularArray)
 884           },
 885           active: {
 886             history: expect.any(CircularArray)
 887           }
 888         }
 889       })
 890     }
 891     await Promise.all(promises)
 892     for (const workerNode of pool.workerNodes) {
 893       expect(workerNode.usage).toStrictEqual({
 894         tasks: {
 895           executed: maxMultiplier,
 896           executing: 0,
 897           queued: 0,
 898           maxQueued: 0,
 899           stolen: 0,
 900           failed: 0
 901         },
 902         runTime: {
 903           history: expect.any(CircularArray)
 904         },
 905         waitTime: {
 906           history: expect.any(CircularArray)
 907         },
 908         elu: {
 909           idle: {
 910             history: expect.any(CircularArray)
 911           },
 912           active: {
 913             history: expect.any(CircularArray)
 914           }
 915         }
 916       })
 917     }
 918     await pool.destroy()
 919   })
 920
 921   it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
 922     const pool = new DynamicThreadPool(
 923       Math.floor(numberOfWorkers / 2),
 924       numberOfWorkers,
 925       './tests/worker-files/thread/testWorker.js'
 926     )
 927     const promises = new Set()
 928     const maxMultiplier = 2
 929     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 930       promises.add(pool.execute())
 931     }
 932     await Promise.all(promises)
 933     for (const workerNode of pool.workerNodes) {
 934       expect(workerNode.usage).toStrictEqual({
 935         tasks: {
 936           executed: expect.any(Number),
 937           executing: 0,
 938           queued: 0,
 939           maxQueued: 0,
 940           stolen: 0,
 941           failed: 0
 942         },
 943         runTime: {
 944           history: expect.any(CircularArray)
 945         },
 946         waitTime: {
 947           history: expect.any(CircularArray)
 948         },
 949         elu: {
 950           idle: {
 951             history: expect.any(CircularArray)
 952           },
 953           active: {
 954             history: expect.any(CircularArray)
 955           }
 956         }
 957       })
 958       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
 959       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
 960         numberOfWorkers * maxMultiplier
 961       )
 962       expect(workerNode.usage.runTime.history.length).toBe(0)
 963       expect(workerNode.usage.waitTime.history.length).toBe(0)
 964       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 965       expect(workerNode.usage.elu.active.history.length).toBe(0)
 966     }
 967     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 968     for (const workerNode of pool.workerNodes) {
 969       expect(workerNode.usage).toStrictEqual({
 970         tasks: {
 971           executed: 0,
 972           executing: 0,
 973           queued: 0,
 974           maxQueued: 0,
 975           stolen: 0,
 976           failed: 0
 977         },
 978         runTime: {
 979           history: expect.any(CircularArray)
 980         },
 981         waitTime: {
 982           history: expect.any(CircularArray)
 983         },
 984         elu: {
 985           idle: {
 986             history: expect.any(CircularArray)
 987           },
 988           active: {
 989             history: expect.any(CircularArray)
 990           }
 991         }
 992       })
 993       expect(workerNode.usage.runTime.history.length).toBe(0)
 994       expect(workerNode.usage.waitTime.history.length).toBe(0)
 995       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 996       expect(workerNode.usage.elu.active.history.length).toBe(0)
 997     }
 998     await pool.destroy()
 999   })
1000
1001   it("Verify that pool event emitter 'ready' event can register a callback", async () => {
1002     const pool = new DynamicClusterPool(
1003       Math.floor(numberOfWorkers / 2),
1004       numberOfWorkers,
1005       './tests/worker-files/cluster/testWorker.js'
1006     )
1007     let poolInfo
1008     let poolReady = 0
1009     pool.emitter.on(PoolEvents.ready, (info) => {
1010       ++poolReady
1011       poolInfo = info
1012     })
1013     await waitPoolEvents(pool, PoolEvents.ready, 1)
1014     expect(poolReady).toBe(1)
1015     expect(poolInfo).toStrictEqual({
1016       version,
1017       type: PoolTypes.dynamic,
1018       worker: WorkerTypes.cluster,
1019       ready: true,
1020       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1021       minSize: expect.any(Number),
1022       maxSize: expect.any(Number),
1023       workerNodes: expect.any(Number),
1024       idleWorkerNodes: expect.any(Number),
1025       busyWorkerNodes: expect.any(Number),
1026       executedTasks: expect.any(Number),
1027       executingTasks: expect.any(Number),
1028       failedTasks: expect.any(Number)
1029     })
1030     await pool.destroy()
1031   })
1032
1033   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
1034     const pool = new FixedThreadPool(
1035       numberOfWorkers,
1036       './tests/worker-files/thread/testWorker.js'
1037     )
1038     const promises = new Set()
1039     let poolBusy = 0
1040     let poolInfo
1041     pool.emitter.on(PoolEvents.busy, (info) => {
1042       ++poolBusy
1043       poolInfo = info
1044     })
1045     for (let i = 0; i < numberOfWorkers * 2; i++) {
1046       promises.add(pool.execute())
1047     }
1048     await Promise.all(promises)
1049     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
1050     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
1051     expect(poolBusy).toBe(numberOfWorkers + 1)
1052     expect(poolInfo).toStrictEqual({
1053       version,
1054       type: PoolTypes.fixed,
1055       worker: WorkerTypes.thread,
1056       ready: expect.any(Boolean),
1057       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1058       minSize: expect.any(Number),
1059       maxSize: expect.any(Number),
1060       workerNodes: expect.any(Number),
1061       idleWorkerNodes: expect.any(Number),
1062       busyWorkerNodes: expect.any(Number),
1063       executedTasks: expect.any(Number),
1064       executingTasks: expect.any(Number),
1065       failedTasks: expect.any(Number)
1066     })
1067     await pool.destroy()
1068   })
1069
1070   it("Verify that pool event emitter 'full' event can register a callback", async () => {
1071     const pool = new DynamicThreadPool(
1072       Math.floor(numberOfWorkers / 2),
1073       numberOfWorkers,
1074       './tests/worker-files/thread/testWorker.js'
1075     )
1076     const promises = new Set()
1077     let poolFull = 0
1078     let poolInfo
1079     pool.emitter.on(PoolEvents.full, (info) => {
1080       ++poolFull
1081       poolInfo = info
1082     })
1083     for (let i = 0; i < numberOfWorkers * 2; i++) {
1084       promises.add(pool.execute())
1085     }
1086     await Promise.all(promises)
1087     expect(poolFull).toBe(1)
1088     expect(poolInfo).toStrictEqual({
1089       version,
1090       type: PoolTypes.dynamic,
1091       worker: WorkerTypes.thread,
1092       ready: expect.any(Boolean),
1093       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1094       minSize: expect.any(Number),
1095       maxSize: expect.any(Number),
1096       workerNodes: expect.any(Number),
1097       idleWorkerNodes: expect.any(Number),
1098       busyWorkerNodes: expect.any(Number),
1099       executedTasks: expect.any(Number),
1100       executingTasks: expect.any(Number),
1101       failedTasks: expect.any(Number)
1102     })
1103     await pool.destroy()
1104   })
1105
1106   it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
1107     const pool = new FixedThreadPool(
1108       numberOfWorkers,
1109       './tests/worker-files/thread/testWorker.js',
1110       {
1111         enableTasksQueue: true
1112       }
1113     )
1114     sinon.stub(pool, 'hasBackPressure').returns(true)
1115     const promises = new Set()
1116     let poolBackPressure = 0
1117     let poolInfo
1118     pool.emitter.on(PoolEvents.backPressure, (info) => {
1119       ++poolBackPressure
1120       poolInfo = info
1121     })
1122     for (let i = 0; i < numberOfWorkers + 1; i++) {
1123       promises.add(pool.execute())
1124     }
1125     await Promise.all(promises)
1126     expect(poolBackPressure).toBe(1)
1127     expect(poolInfo).toStrictEqual({
1128       version,
1129       type: PoolTypes.fixed,
1130       worker: WorkerTypes.thread,
1131       ready: expect.any(Boolean),
1132       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1133       minSize: expect.any(Number),
1134       maxSize: expect.any(Number),
1135       workerNodes: expect.any(Number),
1136       idleWorkerNodes: expect.any(Number),
1137       busyWorkerNodes: expect.any(Number),
1138       executedTasks: expect.any(Number),
1139       executingTasks: expect.any(Number),
1140       maxQueuedTasks: expect.any(Number),
1141       queuedTasks: expect.any(Number),
1142       backPressure: true,
1143       stolenTasks: expect.any(Number),
1144       failedTasks: expect.any(Number)
1145     })
1146     expect(pool.hasBackPressure.called).toBe(true)
1147     await pool.destroy()
1148   })
1149
1150   it('Verify that listTaskFunctions() is working', async () => {
1151     const dynamicThreadPool = new DynamicThreadPool(
1152       Math.floor(numberOfWorkers / 2),
1153       numberOfWorkers,
1154       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
1155     )
1156     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1157     expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
1158       'default',
1159       'jsonIntegerSerialization',
1160       'factorial',
1161       'fibonacci'
1162     ])
1163     const fixedClusterPool = new FixedClusterPool(
1164       numberOfWorkers,
1165       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1166     )
1167     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1168     expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
1169       'default',
1170       'jsonIntegerSerialization',
1171       'factorial',
1172       'fibonacci'
1173     ])
1174   })
1175
1176   it('Verify that multiple task functions worker is working', async () => {
1177     const pool = new DynamicClusterPool(
1178       Math.floor(numberOfWorkers / 2),
1179       numberOfWorkers,
1180       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1181     )
1182     const data = { n: 10 }
1183     const result0 = await pool.execute(data)
1184     expect(result0).toStrictEqual({ ok: 1 })
1185     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1186     expect(result1).toStrictEqual({ ok: 1 })
1187     const result2 = await pool.execute(data, 'factorial')
1188     expect(result2).toBe(3628800)
1189     const result3 = await pool.execute(data, 'fibonacci')
1190     expect(result3).toBe(55)
1191     expect(pool.info.executingTasks).toBe(0)
1192     expect(pool.info.executedTasks).toBe(4)
1193     for (const workerNode of pool.workerNodes) {
1194       expect(workerNode.info.taskFunctions).toStrictEqual([
1195         'default',
1196         'jsonIntegerSerialization',
1197         'factorial',
1198         'fibonacci'
1199       ])
1200       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1201       for (const name of pool.listTaskFunctions()) {
1202         expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1203           tasks: {
1204             executed: expect.any(Number),
1205             executing: expect.any(Number),
1206             failed: 0,
1207             queued: 0,
1208             stolen: 0
1209           },
1210           runTime: {
1211             history: expect.any(CircularArray)
1212           },
1213           waitTime: {
1214             history: expect.any(CircularArray)
1215           },
1216           elu: {
1217             idle: {
1218               history: expect.any(CircularArray)
1219             },
1220             active: {
1221               history: expect.any(CircularArray)
1222             }
1223           }
1224         })
1225         expect(
1226           workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1227         ).toBeGreaterThanOrEqual(0)
1228       }
1229     }
1230   })
1231 })