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