tests/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 { DEFAULT_TASK_NAME } = require('../../../lib/utils')
  16 const { version } = require('../../../package.json')
  17 const { waitPoolEvents } = require('../../test-utils')
  18
  19 describe('Abstract pool test suite', () => {
  20   const numberOfWorkers = 2
  21   class StubPoolWithIsMain extends FixedThreadPool {
  22     isMain () {
  23       return false
  24     }
  25   }
  26
  27   afterEach(() => {
  28     sinon.restore()
  29   })
  30
  31   it('Simulate pool creation from a non main thread/process', () => {
  32     expect(
  33       () =>
  34         new StubPoolWithIsMain(
  35           numberOfWorkers,
  36           './tests/worker-files/thread/testWorker.js',
  37           {
  38             errorHandler: (e) => console.error(e)
  39           }
  40         )
  41     ).toThrowError(
  42       new Error(
  43         'Cannot start a pool from a worker with the same type as the pool'
  44       )
  45     )
  46   })
  47
  48   it('Verify that pool statuses properties are set', async () => {
  49     const pool = new FixedThreadPool(
  50       numberOfWorkers,
  51       './tests/worker-files/thread/testWorker.js'
  52     )
  53     expect(pool.starting).toBe(false)
  54     expect(pool.started).toBe(true)
  55     await pool.destroy()
  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 DynamicThreadPool(0, 0, './tests/worker-files/thread/testWorker.js')
 157     ).toThrowError(
 158       new RangeError(
 159         'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
 160       )
 161     )
 162     expect(
 163       () =>
 164         new DynamicClusterPool(
 165           1,
 166           1,
 167           './tests/worker-files/cluster/testWorker.js'
 168         )
 169     ).toThrowError(
 170       new RangeError(
 171         'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
 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       retries: 6,
 191       runTime: { median: false },
 192       waitTime: { median: false },
 193       elu: { median: false }
 194     })
 195     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 196       retries: 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       retries: 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       retries: 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               retries: 'invalidChoiceRetries'
 279             }
 280           }
 281         )
 282     ).toThrowError(
 283       new TypeError(
 284         'Invalid worker choice strategy options: 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               retries: -1
 295             }
 296           }
 297         )
 298     ).toThrowError(
 299       new RangeError(
 300         "Invalid worker choice strategy options: 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       retries: 6,
 455       runTime: { median: false },
 456       waitTime: { median: false },
 457       elu: { median: false }
 458     })
 459     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 460       retries: 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         retries: 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       retries: 6,
 499       runTime: { median: true },
 500       waitTime: { median: false },
 501       elu: { median: true }
 502     })
 503     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 504       retries: 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         retries: 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       retries: 6,
 543       runTime: { median: false },
 544       waitTime: { median: false },
 545       elu: { median: false }
 546     })
 547     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 548       retries: 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         retries: 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         retries: 'invalidChoiceRetries'
 591       })
 592     ).toThrowError(
 593       new TypeError(
 594         'Invalid worker choice strategy options: retries must be an integer'
 595       )
 596     )
 597     expect(() =>
 598       pool.setWorkerChoiceStrategyOptions({ retries: -1 })
 599     ).toThrowError(
 600       new RangeError(
 601         "Invalid worker choice strategy options: 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     await pool.destroy()
 802   })
 803
 804   it('Verify that pool worker info are initialized', async () => {
 805     let pool = new FixedClusterPool(
 806       numberOfWorkers,
 807       './tests/worker-files/cluster/testWorker.js'
 808     )
 809     for (const workerNode of pool.workerNodes) {
 810       expect(workerNode.info).toStrictEqual({
 811         id: expect.any(Number),
 812         type: WorkerTypes.cluster,
 813         dynamic: false,
 814         ready: true
 815       })
 816     }
 817     await pool.destroy()
 818     pool = new DynamicThreadPool(
 819       Math.floor(numberOfWorkers / 2),
 820       numberOfWorkers,
 821       './tests/worker-files/thread/testWorker.js'
 822     )
 823     for (const workerNode of pool.workerNodes) {
 824       expect(workerNode.info).toStrictEqual({
 825         id: expect.any(Number),
 826         type: WorkerTypes.thread,
 827         dynamic: false,
 828         ready: true
 829       })
 830     }
 831     await pool.destroy()
 832   })
 833
 834   it('Verify that pool execute() arguments are checked', async () => {
 835     const pool = new FixedClusterPool(
 836       numberOfWorkers,
 837       './tests/worker-files/cluster/testWorker.js'
 838     )
 839     await expect(pool.execute(undefined, 0)).rejects.toThrowError(
 840       new TypeError('name argument must be a string')
 841     )
 842     await expect(pool.execute(undefined, '')).rejects.toThrowError(
 843       new TypeError('name argument must not be an empty string')
 844     )
 845     await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
 846       new TypeError('transferList argument must be an array')
 847     )
 848     await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
 849       "Task function 'unknown' not found"
 850     )
 851     await pool.destroy()
 852     await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
 853       new Error('Cannot execute a task on destroyed pool')
 854     )
 855   })
 856
 857   it('Verify that pool worker tasks usage are computed', async () => {
 858     const pool = new FixedClusterPool(
 859       numberOfWorkers,
 860       './tests/worker-files/cluster/testWorker.js'
 861     )
 862     const promises = new Set()
 863     const maxMultiplier = 2
 864     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 865       promises.add(pool.execute())
 866     }
 867     for (const workerNode of pool.workerNodes) {
 868       expect(workerNode.usage).toStrictEqual({
 869         tasks: {
 870           executed: 0,
 871           executing: maxMultiplier,
 872           queued: 0,
 873           maxQueued: 0,
 874           stolen: 0,
 875           failed: 0
 876         },
 877         runTime: {
 878           history: expect.any(CircularArray)
 879         },
 880         waitTime: {
 881           history: expect.any(CircularArray)
 882         },
 883         elu: {
 884           idle: {
 885             history: expect.any(CircularArray)
 886           },
 887           active: {
 888             history: expect.any(CircularArray)
 889           }
 890         }
 891       })
 892     }
 893     await Promise.all(promises)
 894     for (const workerNode of pool.workerNodes) {
 895       expect(workerNode.usage).toStrictEqual({
 896         tasks: {
 897           executed: maxMultiplier,
 898           executing: 0,
 899           queued: 0,
 900           maxQueued: 0,
 901           stolen: 0,
 902           failed: 0
 903         },
 904         runTime: {
 905           history: expect.any(CircularArray)
 906         },
 907         waitTime: {
 908           history: expect.any(CircularArray)
 909         },
 910         elu: {
 911           idle: {
 912             history: expect.any(CircularArray)
 913           },
 914           active: {
 915             history: expect.any(CircularArray)
 916           }
 917         }
 918       })
 919     }
 920     await pool.destroy()
 921   })
 922
 923   it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
 924     const pool = new DynamicThreadPool(
 925       Math.floor(numberOfWorkers / 2),
 926       numberOfWorkers,
 927       './tests/worker-files/thread/testWorker.js'
 928     )
 929     const promises = new Set()
 930     const maxMultiplier = 2
 931     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 932       promises.add(pool.execute())
 933     }
 934     await Promise.all(promises)
 935     for (const workerNode of pool.workerNodes) {
 936       expect(workerNode.usage).toStrictEqual({
 937         tasks: {
 938           executed: expect.any(Number),
 939           executing: 0,
 940           queued: 0,
 941           maxQueued: 0,
 942           stolen: 0,
 943           failed: 0
 944         },
 945         runTime: {
 946           history: expect.any(CircularArray)
 947         },
 948         waitTime: {
 949           history: expect.any(CircularArray)
 950         },
 951         elu: {
 952           idle: {
 953             history: expect.any(CircularArray)
 954           },
 955           active: {
 956             history: expect.any(CircularArray)
 957           }
 958         }
 959       })
 960       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
 961       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
 962         numberOfWorkers * maxMultiplier
 963       )
 964       expect(workerNode.usage.runTime.history.length).toBe(0)
 965       expect(workerNode.usage.waitTime.history.length).toBe(0)
 966       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 967       expect(workerNode.usage.elu.active.history.length).toBe(0)
 968     }
 969     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 970     for (const workerNode of pool.workerNodes) {
 971       expect(workerNode.usage).toStrictEqual({
 972         tasks: {
 973           executed: 0,
 974           executing: 0,
 975           queued: 0,
 976           maxQueued: 0,
 977           stolen: 0,
 978           failed: 0
 979         },
 980         runTime: {
 981           history: expect.any(CircularArray)
 982         },
 983         waitTime: {
 984           history: expect.any(CircularArray)
 985         },
 986         elu: {
 987           idle: {
 988             history: expect.any(CircularArray)
 989           },
 990           active: {
 991             history: expect.any(CircularArray)
 992           }
 993         }
 994       })
 995       expect(workerNode.usage.runTime.history.length).toBe(0)
 996       expect(workerNode.usage.waitTime.history.length).toBe(0)
 997       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 998       expect(workerNode.usage.elu.active.history.length).toBe(0)
 999     }
1000     await pool.destroy()
1001   })
1002
1003   it("Verify that pool event emitter 'ready' event can register a callback", async () => {
1004     const pool = new DynamicClusterPool(
1005       Math.floor(numberOfWorkers / 2),
1006       numberOfWorkers,
1007       './tests/worker-files/cluster/testWorker.js'
1008     )
1009     let poolInfo
1010     let poolReady = 0
1011     pool.emitter.on(PoolEvents.ready, (info) => {
1012       ++poolReady
1013       poolInfo = info
1014     })
1015     await waitPoolEvents(pool, PoolEvents.ready, 1)
1016     expect(poolReady).toBe(1)
1017     expect(poolInfo).toStrictEqual({
1018       version,
1019       type: PoolTypes.dynamic,
1020       worker: WorkerTypes.cluster,
1021       ready: true,
1022       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1023       minSize: expect.any(Number),
1024       maxSize: expect.any(Number),
1025       workerNodes: expect.any(Number),
1026       idleWorkerNodes: expect.any(Number),
1027       busyWorkerNodes: expect.any(Number),
1028       executedTasks: expect.any(Number),
1029       executingTasks: expect.any(Number),
1030       failedTasks: expect.any(Number)
1031     })
1032     await pool.destroy()
1033   })
1034
1035   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
1036     const pool = new FixedThreadPool(
1037       numberOfWorkers,
1038       './tests/worker-files/thread/testWorker.js'
1039     )
1040     const promises = new Set()
1041     let poolBusy = 0
1042     let poolInfo
1043     pool.emitter.on(PoolEvents.busy, (info) => {
1044       ++poolBusy
1045       poolInfo = info
1046     })
1047     for (let i = 0; i < numberOfWorkers * 2; i++) {
1048       promises.add(pool.execute())
1049     }
1050     await Promise.all(promises)
1051     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
1052     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
1053     expect(poolBusy).toBe(numberOfWorkers + 1)
1054     expect(poolInfo).toStrictEqual({
1055       version,
1056       type: PoolTypes.fixed,
1057       worker: WorkerTypes.thread,
1058       ready: expect.any(Boolean),
1059       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1060       minSize: expect.any(Number),
1061       maxSize: expect.any(Number),
1062       workerNodes: expect.any(Number),
1063       idleWorkerNodes: expect.any(Number),
1064       busyWorkerNodes: expect.any(Number),
1065       executedTasks: expect.any(Number),
1066       executingTasks: expect.any(Number),
1067       failedTasks: expect.any(Number)
1068     })
1069     await pool.destroy()
1070   })
1071
1072   it("Verify that pool event emitter 'full' event can register a callback", async () => {
1073     const pool = new DynamicThreadPool(
1074       Math.floor(numberOfWorkers / 2),
1075       numberOfWorkers,
1076       './tests/worker-files/thread/testWorker.js'
1077     )
1078     const promises = new Set()
1079     let poolFull = 0
1080     let poolInfo
1081     pool.emitter.on(PoolEvents.full, (info) => {
1082       ++poolFull
1083       poolInfo = info
1084     })
1085     for (let i = 0; i < numberOfWorkers * 2; i++) {
1086       promises.add(pool.execute())
1087     }
1088     await Promise.all(promises)
1089     expect(poolFull).toBe(1)
1090     expect(poolInfo).toStrictEqual({
1091       version,
1092       type: PoolTypes.dynamic,
1093       worker: WorkerTypes.thread,
1094       ready: expect.any(Boolean),
1095       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1096       minSize: expect.any(Number),
1097       maxSize: expect.any(Number),
1098       workerNodes: expect.any(Number),
1099       idleWorkerNodes: expect.any(Number),
1100       busyWorkerNodes: expect.any(Number),
1101       executedTasks: expect.any(Number),
1102       executingTasks: expect.any(Number),
1103       failedTasks: expect.any(Number)
1104     })
1105     await pool.destroy()
1106   })
1107
1108   it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
1109     const pool = new FixedThreadPool(
1110       numberOfWorkers,
1111       './tests/worker-files/thread/testWorker.js',
1112       {
1113         enableTasksQueue: true
1114       }
1115     )
1116     sinon.stub(pool, 'hasBackPressure').returns(true)
1117     const promises = new Set()
1118     let poolBackPressure = 0
1119     let poolInfo
1120     pool.emitter.on(PoolEvents.backPressure, (info) => {
1121       ++poolBackPressure
1122       poolInfo = info
1123     })
1124     for (let i = 0; i < numberOfWorkers + 1; i++) {
1125       promises.add(pool.execute())
1126     }
1127     await Promise.all(promises)
1128     expect(poolBackPressure).toBe(1)
1129     expect(poolInfo).toStrictEqual({
1130       version,
1131       type: PoolTypes.fixed,
1132       worker: WorkerTypes.thread,
1133       ready: expect.any(Boolean),
1134       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1135       minSize: expect.any(Number),
1136       maxSize: expect.any(Number),
1137       workerNodes: expect.any(Number),
1138       idleWorkerNodes: expect.any(Number),
1139       busyWorkerNodes: expect.any(Number),
1140       executedTasks: expect.any(Number),
1141       executingTasks: expect.any(Number),
1142       maxQueuedTasks: expect.any(Number),
1143       queuedTasks: expect.any(Number),
1144       backPressure: true,
1145       stolenTasks: expect.any(Number),
1146       failedTasks: expect.any(Number)
1147     })
1148     expect(pool.hasBackPressure.called).toBe(true)
1149     await pool.destroy()
1150   })
1151
1152   it('Verify that listTaskFunctions() is working', async () => {
1153     const dynamicThreadPool = new DynamicThreadPool(
1154       Math.floor(numberOfWorkers / 2),
1155       numberOfWorkers,
1156       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
1157     )
1158     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1159     expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
1160       DEFAULT_TASK_NAME,
1161       'jsonIntegerSerialization',
1162       'factorial',
1163       'fibonacci'
1164     ])
1165     const fixedClusterPool = new FixedClusterPool(
1166       numberOfWorkers,
1167       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1168     )
1169     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1170     expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
1171       DEFAULT_TASK_NAME,
1172       'jsonIntegerSerialization',
1173       'factorial',
1174       'fibonacci'
1175     ])
1176     await dynamicThreadPool.destroy()
1177     await fixedClusterPool.destroy()
1178   })
1179
1180   it('Verify that multiple task functions worker is working', async () => {
1181     const pool = new DynamicClusterPool(
1182       Math.floor(numberOfWorkers / 2),
1183       numberOfWorkers,
1184       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1185     )
1186     const data = { n: 10 }
1187     const result0 = await pool.execute(data)
1188     expect(result0).toStrictEqual({ ok: 1 })
1189     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1190     expect(result1).toStrictEqual({ ok: 1 })
1191     const result2 = await pool.execute(data, 'factorial')
1192     expect(result2).toBe(3628800)
1193     const result3 = await pool.execute(data, 'fibonacci')
1194     expect(result3).toBe(55)
1195     expect(pool.info.executingTasks).toBe(0)
1196     expect(pool.info.executedTasks).toBe(4)
1197     for (const workerNode of pool.workerNodes) {
1198       expect(workerNode.info.taskFunctions).toStrictEqual([
1199         DEFAULT_TASK_NAME,
1200         'jsonIntegerSerialization',
1201         'factorial',
1202         'fibonacci'
1203       ])
1204       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1205       for (const name of pool.listTaskFunctions()) {
1206         expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1207           tasks: {
1208             executed: expect.any(Number),
1209             executing: expect.any(Number),
1210             failed: 0,
1211             queued: 0,
1212             stolen: 0
1213           },
1214           runTime: {
1215             history: expect.any(CircularArray)
1216           },
1217           waitTime: {
1218             history: expect.any(CircularArray)
1219           },
1220           elu: {
1221             idle: {
1222               history: expect.any(CircularArray)
1223             },
1224             active: {
1225               history: expect.any(CircularArray)
1226             }
1227           }
1228         })
1229         expect(
1230           workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1231         ).toBeGreaterThanOrEqual(0)
1232       }
1233     }
1234     await pool.destroy()
1235   })
1236 })