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