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({ queueMaxSize: 2 })).toThrowError(
 548       new Error(
 549         'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
 550       )
 551     )
 552     expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
 553       new RangeError(
 554         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
 555       )
 556     )
 557     expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError(
 558       new RangeError(
 559         'Invalid worker node tasks queue size: -1 is a negative integer or zero'
 560       )
 561     )
 562     expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError(
 563       new TypeError('Invalid worker node tasks queue size: must be an integer')
 564     )
 565     await pool.destroy()
 566   })
 567
 568   it('Verify that pool info is set', async () => {
 569     let pool = new FixedThreadPool(
 570       numberOfWorkers,
 571       './tests/worker-files/thread/testWorker.js'
 572     )
 573     expect(pool.info).toStrictEqual({
 574       version,
 575       type: PoolTypes.fixed,
 576       worker: WorkerTypes.thread,
 577       ready: true,
 578       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 579       minSize: numberOfWorkers,
 580       maxSize: numberOfWorkers,
 581       workerNodes: numberOfWorkers,
 582       idleWorkerNodes: numberOfWorkers,
 583       busyWorkerNodes: 0,
 584       executedTasks: 0,
 585       executingTasks: 0,
 586       failedTasks: 0
 587     })
 588     await pool.destroy()
 589     pool = new DynamicClusterPool(
 590       Math.floor(numberOfWorkers / 2),
 591       numberOfWorkers,
 592       './tests/worker-files/cluster/testWorker.js'
 593     )
 594     expect(pool.info).toStrictEqual({
 595       version,
 596       type: PoolTypes.dynamic,
 597       worker: WorkerTypes.cluster,
 598       ready: true,
 599       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 600       minSize: Math.floor(numberOfWorkers / 2),
 601       maxSize: numberOfWorkers,
 602       workerNodes: Math.floor(numberOfWorkers / 2),
 603       idleWorkerNodes: Math.floor(numberOfWorkers / 2),
 604       busyWorkerNodes: 0,
 605       executedTasks: 0,
 606       executingTasks: 0,
 607       failedTasks: 0
 608     })
 609     await pool.destroy()
 610   })
 611
 612   it('Verify that pool worker tasks usage are initialized', async () => {
 613     const pool = new FixedClusterPool(
 614       numberOfWorkers,
 615       './tests/worker-files/cluster/testWorker.js'
 616     )
 617     for (const workerNode of pool.workerNodes) {
 618       expect(workerNode.usage).toStrictEqual({
 619         tasks: {
 620           executed: 0,
 621           executing: 0,
 622           queued: 0,
 623           maxQueued: 0,
 624           stolen: 0,
 625           failed: 0
 626         },
 627         runTime: {
 628           history: expect.any(CircularArray)
 629         },
 630         waitTime: {
 631           history: expect.any(CircularArray)
 632         },
 633         elu: {
 634           idle: {
 635             history: expect.any(CircularArray)
 636           },
 637           active: {
 638             history: expect.any(CircularArray)
 639           }
 640         }
 641       })
 642     }
 643     await pool.destroy()
 644   })
 645
 646   it('Verify that pool worker tasks queue are initialized', async () => {
 647     let pool = new FixedClusterPool(
 648       numberOfWorkers,
 649       './tests/worker-files/cluster/testWorker.js'
 650     )
 651     for (const workerNode of pool.workerNodes) {
 652       expect(workerNode.tasksQueue).toBeDefined()
 653       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 654       expect(workerNode.tasksQueue.size).toBe(0)
 655       expect(workerNode.tasksQueue.maxSize).toBe(0)
 656     }
 657     await pool.destroy()
 658     pool = new DynamicThreadPool(
 659       Math.floor(numberOfWorkers / 2),
 660       numberOfWorkers,
 661       './tests/worker-files/thread/testWorker.js'
 662     )
 663     for (const workerNode of pool.workerNodes) {
 664       expect(workerNode.tasksQueue).toBeDefined()
 665       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 666       expect(workerNode.tasksQueue.size).toBe(0)
 667       expect(workerNode.tasksQueue.maxSize).toBe(0)
 668     }
 669   })
 670
 671   it('Verify that pool worker info are initialized', async () => {
 672     let pool = new FixedClusterPool(
 673       numberOfWorkers,
 674       './tests/worker-files/cluster/testWorker.js'
 675     )
 676     for (const workerNode of pool.workerNodes) {
 677       expect(workerNode.info).toStrictEqual({
 678         id: expect.any(Number),
 679         type: WorkerTypes.cluster,
 680         dynamic: false,
 681         ready: true
 682       })
 683     }
 684     await pool.destroy()
 685     pool = new DynamicThreadPool(
 686       Math.floor(numberOfWorkers / 2),
 687       numberOfWorkers,
 688       './tests/worker-files/thread/testWorker.js'
 689     )
 690     for (const workerNode of pool.workerNodes) {
 691       expect(workerNode.info).toStrictEqual({
 692         id: expect.any(Number),
 693         type: WorkerTypes.thread,
 694         dynamic: false,
 695         ready: true
 696       })
 697     }
 698   })
 699
 700   it('Verify that pool worker tasks usage are computed', async () => {
 701     const pool = new FixedClusterPool(
 702       numberOfWorkers,
 703       './tests/worker-files/cluster/testWorker.js'
 704     )
 705     const promises = new Set()
 706     const maxMultiplier = 2
 707     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 708       promises.add(pool.execute())
 709     }
 710     for (const workerNode of pool.workerNodes) {
 711       expect(workerNode.usage).toStrictEqual({
 712         tasks: {
 713           executed: 0,
 714           executing: maxMultiplier,
 715           queued: 0,
 716           maxQueued: 0,
 717           stolen: 0,
 718           failed: 0
 719         },
 720         runTime: {
 721           history: expect.any(CircularArray)
 722         },
 723         waitTime: {
 724           history: expect.any(CircularArray)
 725         },
 726         elu: {
 727           idle: {
 728             history: expect.any(CircularArray)
 729           },
 730           active: {
 731             history: expect.any(CircularArray)
 732           }
 733         }
 734       })
 735     }
 736     await Promise.all(promises)
 737     for (const workerNode of pool.workerNodes) {
 738       expect(workerNode.usage).toStrictEqual({
 739         tasks: {
 740           executed: maxMultiplier,
 741           executing: 0,
 742           queued: 0,
 743           maxQueued: 0,
 744           stolen: 0,
 745           failed: 0
 746         },
 747         runTime: {
 748           history: expect.any(CircularArray)
 749         },
 750         waitTime: {
 751           history: expect.any(CircularArray)
 752         },
 753         elu: {
 754           idle: {
 755             history: expect.any(CircularArray)
 756           },
 757           active: {
 758             history: expect.any(CircularArray)
 759           }
 760         }
 761       })
 762     }
 763     await pool.destroy()
 764   })
 765
 766   it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
 767     const pool = new DynamicThreadPool(
 768       Math.floor(numberOfWorkers / 2),
 769       numberOfWorkers,
 770       './tests/worker-files/thread/testWorker.js'
 771     )
 772     const promises = new Set()
 773     const maxMultiplier = 2
 774     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 775       promises.add(pool.execute())
 776     }
 777     await Promise.all(promises)
 778     for (const workerNode of pool.workerNodes) {
 779       expect(workerNode.usage).toStrictEqual({
 780         tasks: {
 781           executed: expect.any(Number),
 782           executing: 0,
 783           queued: 0,
 784           maxQueued: 0,
 785           stolen: 0,
 786           failed: 0
 787         },
 788         runTime: {
 789           history: expect.any(CircularArray)
 790         },
 791         waitTime: {
 792           history: expect.any(CircularArray)
 793         },
 794         elu: {
 795           idle: {
 796             history: expect.any(CircularArray)
 797           },
 798           active: {
 799             history: expect.any(CircularArray)
 800           }
 801         }
 802       })
 803       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
 804       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
 805         numberOfWorkers * maxMultiplier
 806       )
 807       expect(workerNode.usage.runTime.history.length).toBe(0)
 808       expect(workerNode.usage.waitTime.history.length).toBe(0)
 809       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 810       expect(workerNode.usage.elu.active.history.length).toBe(0)
 811     }
 812     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 813     for (const workerNode of pool.workerNodes) {
 814       expect(workerNode.usage).toStrictEqual({
 815         tasks: {
 816           executed: 0,
 817           executing: 0,
 818           queued: 0,
 819           maxQueued: 0,
 820           stolen: 0,
 821           failed: 0
 822         },
 823         runTime: {
 824           history: expect.any(CircularArray)
 825         },
 826         waitTime: {
 827           history: expect.any(CircularArray)
 828         },
 829         elu: {
 830           idle: {
 831             history: expect.any(CircularArray)
 832           },
 833           active: {
 834             history: expect.any(CircularArray)
 835           }
 836         }
 837       })
 838       expect(workerNode.usage.runTime.history.length).toBe(0)
 839       expect(workerNode.usage.waitTime.history.length).toBe(0)
 840       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 841       expect(workerNode.usage.elu.active.history.length).toBe(0)
 842     }
 843     await pool.destroy()
 844   })
 845
 846   it("Verify that pool event emitter 'ready' event can register a callback", async () => {
 847     const pool = new DynamicClusterPool(
 848       Math.floor(numberOfWorkers / 2),
 849       numberOfWorkers,
 850       './tests/worker-files/cluster/testWorker.js'
 851     )
 852     let poolInfo
 853     let poolReady = 0
 854     pool.emitter.on(PoolEvents.ready, (info) => {
 855       ++poolReady
 856       poolInfo = info
 857     })
 858     await waitPoolEvents(pool, PoolEvents.ready, 1)
 859     expect(poolReady).toBe(1)
 860     expect(poolInfo).toStrictEqual({
 861       version,
 862       type: PoolTypes.dynamic,
 863       worker: WorkerTypes.cluster,
 864       ready: true,
 865       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 866       minSize: expect.any(Number),
 867       maxSize: expect.any(Number),
 868       workerNodes: expect.any(Number),
 869       idleWorkerNodes: expect.any(Number),
 870       busyWorkerNodes: expect.any(Number),
 871       executedTasks: expect.any(Number),
 872       executingTasks: expect.any(Number),
 873       failedTasks: expect.any(Number)
 874     })
 875     await pool.destroy()
 876   })
 877
 878   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 879     const pool = new FixedThreadPool(
 880       numberOfWorkers,
 881       './tests/worker-files/thread/testWorker.js'
 882     )
 883     const promises = new Set()
 884     let poolBusy = 0
 885     let poolInfo
 886     pool.emitter.on(PoolEvents.busy, (info) => {
 887       ++poolBusy
 888       poolInfo = info
 889     })
 890     for (let i = 0; i < numberOfWorkers * 2; i++) {
 891       promises.add(pool.execute())
 892     }
 893     await Promise.all(promises)
 894     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 895     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 896     expect(poolBusy).toBe(numberOfWorkers + 1)
 897     expect(poolInfo).toStrictEqual({
 898       version,
 899       type: PoolTypes.fixed,
 900       worker: WorkerTypes.thread,
 901       ready: expect.any(Boolean),
 902       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 903       minSize: expect.any(Number),
 904       maxSize: expect.any(Number),
 905       workerNodes: expect.any(Number),
 906       idleWorkerNodes: expect.any(Number),
 907       busyWorkerNodes: expect.any(Number),
 908       executedTasks: expect.any(Number),
 909       executingTasks: expect.any(Number),
 910       failedTasks: expect.any(Number)
 911     })
 912     await pool.destroy()
 913   })
 914
 915   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 916     const pool = new DynamicThreadPool(
 917       Math.floor(numberOfWorkers / 2),
 918       numberOfWorkers,
 919       './tests/worker-files/thread/testWorker.js'
 920     )
 921     const promises = new Set()
 922     let poolFull = 0
 923     let poolInfo
 924     pool.emitter.on(PoolEvents.full, (info) => {
 925       ++poolFull
 926       poolInfo = info
 927     })
 928     for (let i = 0; i < numberOfWorkers * 2; i++) {
 929       promises.add(pool.execute())
 930     }
 931     await Promise.all(promises)
 932     expect(poolFull).toBe(1)
 933     expect(poolInfo).toStrictEqual({
 934       version,
 935       type: PoolTypes.dynamic,
 936       worker: WorkerTypes.thread,
 937       ready: expect.any(Boolean),
 938       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 939       minSize: expect.any(Number),
 940       maxSize: expect.any(Number),
 941       workerNodes: expect.any(Number),
 942       idleWorkerNodes: expect.any(Number),
 943       busyWorkerNodes: expect.any(Number),
 944       executedTasks: expect.any(Number),
 945       executingTasks: expect.any(Number),
 946       failedTasks: expect.any(Number)
 947     })
 948     await pool.destroy()
 949   })
 950
 951   it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
 952     const pool = new FixedThreadPool(
 953       numberOfWorkers,
 954       './tests/worker-files/thread/testWorker.js',
 955       {
 956         enableTasksQueue: true
 957       }
 958     )
 959     sinon.stub(pool, 'hasBackPressure').returns(true)
 960     const promises = new Set()
 961     let poolBackPressure = 0
 962     let poolInfo
 963     pool.emitter.on(PoolEvents.backPressure, (info) => {
 964       ++poolBackPressure
 965       poolInfo = info
 966     })
 967     for (let i = 0; i < numberOfWorkers + 1; i++) {
 968       promises.add(pool.execute())
 969     }
 970     await Promise.all(promises)
 971     expect(poolBackPressure).toBe(1)
 972     expect(poolInfo).toStrictEqual({
 973       version,
 974       type: PoolTypes.fixed,
 975       worker: WorkerTypes.thread,
 976       ready: expect.any(Boolean),
 977       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 978       minSize: expect.any(Number),
 979       maxSize: expect.any(Number),
 980       workerNodes: expect.any(Number),
 981       idleWorkerNodes: expect.any(Number),
 982       busyWorkerNodes: expect.any(Number),
 983       executedTasks: expect.any(Number),
 984       executingTasks: expect.any(Number),
 985       maxQueuedTasks: expect.any(Number),
 986       queuedTasks: expect.any(Number),
 987       backPressure: true,
 988       stolenTasks: expect.any(Number),
 989       failedTasks: expect.any(Number)
 990     })
 991     expect(pool.hasBackPressure.called).toBe(true)
 992     await pool.destroy()
 993   })
 994
 995   it('Verify that listTaskFunctions() is working', async () => {
 996     const dynamicThreadPool = new DynamicThreadPool(
 997       Math.floor(numberOfWorkers / 2),
 998       numberOfWorkers,
 999       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
1000     )
1001     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1002     expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
1003       'default',
1004       'jsonIntegerSerialization',
1005       'factorial',
1006       'fibonacci'
1007     ])
1008     const fixedClusterPool = new FixedClusterPool(
1009       numberOfWorkers,
1010       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1011     )
1012     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1013     expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
1014       'default',
1015       'jsonIntegerSerialization',
1016       'factorial',
1017       'fibonacci'
1018     ])
1019   })
1020
1021   it('Verify that multiple task functions worker is working', async () => {
1022     const pool = new DynamicClusterPool(
1023       Math.floor(numberOfWorkers / 2),
1024       numberOfWorkers,
1025       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1026     )
1027     const data = { n: 10 }
1028     const result0 = await pool.execute(data)
1029     expect(result0).toStrictEqual({ ok: 1 })
1030     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1031     expect(result1).toStrictEqual({ ok: 1 })
1032     const result2 = await pool.execute(data, 'factorial')
1033     expect(result2).toBe(3628800)
1034     const result3 = await pool.execute(data, 'fibonacci')
1035     expect(result3).toBe(55)
1036     expect(pool.info.executingTasks).toBe(0)
1037     expect(pool.info.executedTasks).toBe(4)
1038     for (const workerNode of pool.workerNodes) {
1039       expect(workerNode.info.taskFunctions).toStrictEqual([
1040         'default',
1041         'jsonIntegerSerialization',
1042         'factorial',
1043         'fibonacci'
1044       ])
1045       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1046       for (const name of pool.listTaskFunctions()) {
1047         expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1048           tasks: {
1049             executed: expect.any(Number),
1050             executing: expect.any(Number),
1051             failed: 0,
1052             queued: 0,
1053             stolen: 0
1054           },
1055           runTime: {
1056             history: expect.any(CircularArray)
1057           },
1058           waitTime: {
1059             history: expect.any(CircularArray)
1060           },
1061           elu: {
1062             idle: {
1063               history: expect.any(CircularArray)
1064             },
1065             active: {
1066               history: expect.any(CircularArray)
1067             }
1068           }
1069         })
1070         expect(
1071           workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1072         ).toBeGreaterThanOrEqual(0)
1073       }
1074     }
1075   })
1076 })