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 { Queue } = require('../../../lib/queue')
  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({ concurrency: 2 })
 216     expect(pool.opts.workerChoiceStrategy).toBe(
 217       WorkerChoiceStrategies.LEAST_USED
 218     )
 219     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 220       choiceRetries: 6,
 221       runTime: { median: true },
 222       waitTime: { median: false },
 223       elu: { median: false },
 224       weights: { 0: 300, 1: 200 }
 225     })
 226     expect(pool.workerChoiceStrategyContext.opts).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.opts.messageHandler).toStrictEqual(testHandler)
 234     expect(pool.opts.errorHandler).toStrictEqual(testHandler)
 235     expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
 236     expect(pool.opts.exitHandler).toStrictEqual(testHandler)
 237     await pool.destroy()
 238   })
 239
 240   it('Verify that pool options are validated', async () => {
 241     expect(
 242       () =>
 243         new FixedThreadPool(
 244           numberOfWorkers,
 245           './tests/worker-files/thread/testWorker.js',
 246           {
 247             workerChoiceStrategy: 'invalidStrategy'
 248           }
 249         )
 250     ).toThrowError(
 251       new Error("Invalid worker choice strategy 'invalidStrategy'")
 252     )
 253     expect(
 254       () =>
 255         new FixedThreadPool(
 256           numberOfWorkers,
 257           './tests/worker-files/thread/testWorker.js',
 258           {
 259             workerChoiceStrategyOptions: { weights: {} }
 260           }
 261         )
 262     ).toThrowError(
 263       new Error(
 264         'Invalid worker choice strategy options: must have a weight for each worker node'
 265       )
 266     )
 267     expect(
 268       () =>
 269         new FixedThreadPool(
 270           numberOfWorkers,
 271           './tests/worker-files/thread/testWorker.js',
 272           {
 273             workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
 274           }
 275         )
 276     ).toThrowError(
 277       new Error(
 278         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 279       )
 280     )
 281     expect(
 282       () =>
 283         new FixedThreadPool(
 284           numberOfWorkers,
 285           './tests/worker-files/thread/testWorker.js',
 286           {
 287             enableTasksQueue: true,
 288             tasksQueueOptions: { concurrency: 0 }
 289           }
 290         )
 291     ).toThrowError(
 292       new RangeError(
 293         'Invalid worker tasks concurrency: 0 is a negative integer or zero'
 294       )
 295     )
 296     expect(
 297       () =>
 298         new FixedThreadPool(
 299           numberOfWorkers,
 300           './tests/worker-files/thread/testWorker.js',
 301           {
 302             enableTasksQueue: true,
 303             tasksQueueOptions: 'invalidTasksQueueOptions'
 304           }
 305         )
 306     ).toThrowError(
 307       new TypeError('Invalid tasks queue options: must be a plain object')
 308     )
 309     expect(
 310       () =>
 311         new FixedThreadPool(
 312           numberOfWorkers,
 313           './tests/worker-files/thread/testWorker.js',
 314           {
 315             enableTasksQueue: true,
 316             tasksQueueOptions: { concurrency: 0.2 }
 317           }
 318         )
 319     ).toThrowError(
 320       new TypeError('Invalid worker tasks concurrency: must be an integer')
 321     )
 322   })
 323
 324   it('Verify that pool worker choice strategy options can be set', async () => {
 325     const pool = new FixedThreadPool(
 326       numberOfWorkers,
 327       './tests/worker-files/thread/testWorker.js',
 328       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 329     )
 330     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 331       choiceRetries: 6,
 332       runTime: { median: false },
 333       waitTime: { median: false },
 334       elu: { median: false }
 335     })
 336     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 337       choiceRetries: 6,
 338       runTime: { median: false },
 339       waitTime: { median: false },
 340       elu: { median: false }
 341     })
 342     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 343       .workerChoiceStrategies) {
 344       expect(workerChoiceStrategy.opts).toStrictEqual({
 345         choiceRetries: 6,
 346         runTime: { median: false },
 347         waitTime: { median: false },
 348         elu: { median: false }
 349       })
 350     }
 351     expect(
 352       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 353     ).toStrictEqual({
 354       runTime: {
 355         aggregate: true,
 356         average: true,
 357         median: false
 358       },
 359       waitTime: {
 360         aggregate: false,
 361         average: false,
 362         median: false
 363       },
 364       elu: {
 365         aggregate: true,
 366         average: true,
 367         median: false
 368       }
 369     })
 370     pool.setWorkerChoiceStrategyOptions({
 371       runTime: { median: true },
 372       elu: { median: true }
 373     })
 374     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 375       choiceRetries: 6,
 376       runTime: { median: true },
 377       waitTime: { median: false },
 378       elu: { median: true }
 379     })
 380     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 381       choiceRetries: 6,
 382       runTime: { median: true },
 383       waitTime: { median: false },
 384       elu: { median: true }
 385     })
 386     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 387       .workerChoiceStrategies) {
 388       expect(workerChoiceStrategy.opts).toStrictEqual({
 389         choiceRetries: 6,
 390         runTime: { median: true },
 391         waitTime: { median: false },
 392         elu: { median: true }
 393       })
 394     }
 395     expect(
 396       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 397     ).toStrictEqual({
 398       runTime: {
 399         aggregate: true,
 400         average: false,
 401         median: true
 402       },
 403       waitTime: {
 404         aggregate: false,
 405         average: false,
 406         median: false
 407       },
 408       elu: {
 409         aggregate: true,
 410         average: false,
 411         median: true
 412       }
 413     })
 414     pool.setWorkerChoiceStrategyOptions({
 415       runTime: { median: false },
 416       elu: { median: false }
 417     })
 418     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 419       choiceRetries: 6,
 420       runTime: { median: false },
 421       waitTime: { median: false },
 422       elu: { median: false }
 423     })
 424     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 425       choiceRetries: 6,
 426       runTime: { median: false },
 427       waitTime: { median: false },
 428       elu: { median: false }
 429     })
 430     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 431       .workerChoiceStrategies) {
 432       expect(workerChoiceStrategy.opts).toStrictEqual({
 433         choiceRetries: 6,
 434         runTime: { median: false },
 435         waitTime: { median: false },
 436         elu: { median: false }
 437       })
 438     }
 439     expect(
 440       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 441     ).toStrictEqual({
 442       runTime: {
 443         aggregate: true,
 444         average: true,
 445         median: false
 446       },
 447       waitTime: {
 448         aggregate: false,
 449         average: false,
 450         median: false
 451       },
 452       elu: {
 453         aggregate: true,
 454         average: true,
 455         median: false
 456       }
 457     })
 458     expect(() =>
 459       pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
 460     ).toThrowError(
 461       new TypeError(
 462         'Invalid worker choice strategy options: must be a plain object'
 463       )
 464     )
 465     expect(() =>
 466       pool.setWorkerChoiceStrategyOptions({ weights: {} })
 467     ).toThrowError(
 468       new Error(
 469         'Invalid worker choice strategy options: must have a weight for each worker node'
 470       )
 471     )
 472     expect(() =>
 473       pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
 474     ).toThrowError(
 475       new Error(
 476         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 477       )
 478     )
 479     await pool.destroy()
 480   })
 481
 482   it('Verify that pool tasks queue can be enabled/disabled', async () => {
 483     const pool = new FixedThreadPool(
 484       numberOfWorkers,
 485       './tests/worker-files/thread/testWorker.js'
 486     )
 487     expect(pool.opts.enableTasksQueue).toBe(false)
 488     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 489     pool.enableTasksQueue(true)
 490     expect(pool.opts.enableTasksQueue).toBe(true)
 491     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 492     pool.enableTasksQueue(true, { concurrency: 2 })
 493     expect(pool.opts.enableTasksQueue).toBe(true)
 494     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 495     pool.enableTasksQueue(false)
 496     expect(pool.opts.enableTasksQueue).toBe(false)
 497     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 498     await pool.destroy()
 499   })
 500
 501   it('Verify that pool tasks queue options can be set', async () => {
 502     const pool = new FixedThreadPool(
 503       numberOfWorkers,
 504       './tests/worker-files/thread/testWorker.js',
 505       { enableTasksQueue: true }
 506     )
 507     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 508     pool.setTasksQueueOptions({ concurrency: 2 })
 509     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 510     expect(() =>
 511       pool.setTasksQueueOptions('invalidTasksQueueOptions')
 512     ).toThrowError(
 513       new TypeError('Invalid tasks queue options: must be a plain object')
 514     )
 515     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
 516       new RangeError(
 517         'Invalid worker tasks concurrency: 0 is a negative integer or zero'
 518       )
 519     )
 520     expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError(
 521       new RangeError(
 522         'Invalid worker tasks concurrency: -1 is a negative integer or zero'
 523       )
 524     )
 525     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
 526       new TypeError('Invalid worker tasks concurrency: must be an integer')
 527     )
 528     await pool.destroy()
 529   })
 530
 531   it('Verify that pool info is set', async () => {
 532     let pool = new FixedThreadPool(
 533       numberOfWorkers,
 534       './tests/worker-files/thread/testWorker.js'
 535     )
 536     expect(pool.info).toStrictEqual({
 537       version,
 538       type: PoolTypes.fixed,
 539       worker: WorkerTypes.thread,
 540       ready: true,
 541       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 542       minSize: numberOfWorkers,
 543       maxSize: numberOfWorkers,
 544       workerNodes: numberOfWorkers,
 545       idleWorkerNodes: numberOfWorkers,
 546       busyWorkerNodes: 0,
 547       executedTasks: 0,
 548       executingTasks: 0,
 549       failedTasks: 0
 550     })
 551     await pool.destroy()
 552     pool = new DynamicClusterPool(
 553       Math.floor(numberOfWorkers / 2),
 554       numberOfWorkers,
 555       './tests/worker-files/cluster/testWorker.js'
 556     )
 557     expect(pool.info).toStrictEqual({
 558       version,
 559       type: PoolTypes.dynamic,
 560       worker: WorkerTypes.cluster,
 561       ready: true,
 562       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 563       minSize: Math.floor(numberOfWorkers / 2),
 564       maxSize: numberOfWorkers,
 565       workerNodes: Math.floor(numberOfWorkers / 2),
 566       idleWorkerNodes: Math.floor(numberOfWorkers / 2),
 567       busyWorkerNodes: 0,
 568       executedTasks: 0,
 569       executingTasks: 0,
 570       failedTasks: 0
 571     })
 572     await pool.destroy()
 573   })
 574
 575   it('Verify that pool worker tasks usage are initialized', async () => {
 576     const pool = new FixedClusterPool(
 577       numberOfWorkers,
 578       './tests/worker-files/cluster/testWorker.js'
 579     )
 580     for (const workerNode of pool.workerNodes) {
 581       expect(workerNode.usage).toStrictEqual({
 582         tasks: {
 583           executed: 0,
 584           executing: 0,
 585           queued: 0,
 586           maxQueued: 0,
 587           failed: 0
 588         },
 589         runTime: {
 590           history: expect.any(CircularArray)
 591         },
 592         waitTime: {
 593           history: expect.any(CircularArray)
 594         },
 595         elu: {
 596           idle: {
 597             history: expect.any(CircularArray)
 598           },
 599           active: {
 600             history: expect.any(CircularArray)
 601           }
 602         }
 603       })
 604     }
 605     await pool.destroy()
 606   })
 607
 608   it('Verify that pool worker tasks queue are initialized', async () => {
 609     let pool = new FixedClusterPool(
 610       numberOfWorkers,
 611       './tests/worker-files/cluster/testWorker.js'
 612     )
 613     for (const workerNode of pool.workerNodes) {
 614       expect(workerNode.tasksQueue).toBeDefined()
 615       expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
 616       expect(workerNode.tasksQueue.size).toBe(0)
 617       expect(workerNode.tasksQueue.maxSize).toBe(0)
 618     }
 619     await pool.destroy()
 620     pool = new DynamicThreadPool(
 621       Math.floor(numberOfWorkers / 2),
 622       numberOfWorkers,
 623       './tests/worker-files/thread/testWorker.js'
 624     )
 625     for (const workerNode of pool.workerNodes) {
 626       expect(workerNode.tasksQueue).toBeDefined()
 627       expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
 628       expect(workerNode.tasksQueue.size).toBe(0)
 629       expect(workerNode.tasksQueue.maxSize).toBe(0)
 630     }
 631   })
 632
 633   it('Verify that pool worker info are initialized', async () => {
 634     let pool = new FixedClusterPool(
 635       numberOfWorkers,
 636       './tests/worker-files/cluster/testWorker.js'
 637     )
 638     for (const workerNode of pool.workerNodes) {
 639       expect(workerNode.info).toStrictEqual({
 640         id: expect.any(Number),
 641         type: WorkerTypes.cluster,
 642         dynamic: false,
 643         ready: true
 644       })
 645     }
 646     await pool.destroy()
 647     pool = new DynamicThreadPool(
 648       Math.floor(numberOfWorkers / 2),
 649       numberOfWorkers,
 650       './tests/worker-files/thread/testWorker.js'
 651     )
 652     for (const workerNode of pool.workerNodes) {
 653       expect(workerNode.info).toStrictEqual({
 654         id: expect.any(Number),
 655         type: WorkerTypes.thread,
 656         dynamic: false,
 657         ready: true
 658       })
 659     }
 660   })
 661
 662   it('Verify that pool worker tasks usage are computed', async () => {
 663     const pool = new FixedClusterPool(
 664       numberOfWorkers,
 665       './tests/worker-files/cluster/testWorker.js'
 666     )
 667     const promises = new Set()
 668     const maxMultiplier = 2
 669     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 670       promises.add(pool.execute())
 671     }
 672     for (const workerNode of pool.workerNodes) {
 673       expect(workerNode.usage).toStrictEqual({
 674         tasks: {
 675           executed: 0,
 676           executing: maxMultiplier,
 677           queued: 0,
 678           maxQueued: 0,
 679           failed: 0
 680         },
 681         runTime: {
 682           history: expect.any(CircularArray)
 683         },
 684         waitTime: {
 685           history: expect.any(CircularArray)
 686         },
 687         elu: {
 688           idle: {
 689             history: expect.any(CircularArray)
 690           },
 691           active: {
 692             history: expect.any(CircularArray)
 693           }
 694         }
 695       })
 696     }
 697     await Promise.all(promises)
 698     for (const workerNode of pool.workerNodes) {
 699       expect(workerNode.usage).toStrictEqual({
 700         tasks: {
 701           executed: maxMultiplier,
 702           executing: 0,
 703           queued: 0,
 704           maxQueued: 0,
 705           failed: 0
 706         },
 707         runTime: {
 708           history: expect.any(CircularArray)
 709         },
 710         waitTime: {
 711           history: expect.any(CircularArray)
 712         },
 713         elu: {
 714           idle: {
 715             history: expect.any(CircularArray)
 716           },
 717           active: {
 718             history: expect.any(CircularArray)
 719           }
 720         }
 721       })
 722     }
 723     await pool.destroy()
 724   })
 725
 726   it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
 727     const pool = new DynamicThreadPool(
 728       Math.floor(numberOfWorkers / 2),
 729       numberOfWorkers,
 730       './tests/worker-files/thread/testWorker.js'
 731     )
 732     const promises = new Set()
 733     const maxMultiplier = 2
 734     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 735       promises.add(pool.execute())
 736     }
 737     await Promise.all(promises)
 738     for (const workerNode of pool.workerNodes) {
 739       expect(workerNode.usage).toStrictEqual({
 740         tasks: {
 741           executed: expect.any(Number),
 742           executing: 0,
 743           queued: 0,
 744           maxQueued: 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       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
 763       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
 764         numberOfWorkers * maxMultiplier
 765       )
 766       expect(workerNode.usage.runTime.history.length).toBe(0)
 767       expect(workerNode.usage.waitTime.history.length).toBe(0)
 768       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 769       expect(workerNode.usage.elu.active.history.length).toBe(0)
 770     }
 771     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 772     for (const workerNode of pool.workerNodes) {
 773       expect(workerNode.usage).toStrictEqual({
 774         tasks: {
 775           executed: 0,
 776           executing: 0,
 777           queued: 0,
 778           maxQueued: 0,
 779           failed: 0
 780         },
 781         runTime: {
 782           history: expect.any(CircularArray)
 783         },
 784         waitTime: {
 785           history: expect.any(CircularArray)
 786         },
 787         elu: {
 788           idle: {
 789             history: expect.any(CircularArray)
 790           },
 791           active: {
 792             history: expect.any(CircularArray)
 793           }
 794         }
 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     await pool.destroy()
 802   })
 803
 804   it("Verify that pool event emitter 'ready' event can register a callback", async () => {
 805     const pool = new DynamicClusterPool(
 806       Math.floor(numberOfWorkers / 2),
 807       numberOfWorkers,
 808       './tests/worker-files/cluster/testWorker.js'
 809     )
 810     let poolInfo
 811     let poolReady = 0
 812     pool.emitter.on(PoolEvents.ready, (info) => {
 813       ++poolReady
 814       poolInfo = info
 815     })
 816     await waitPoolEvents(pool, PoolEvents.ready, 1)
 817     expect(poolReady).toBe(1)
 818     expect(poolInfo).toStrictEqual({
 819       version,
 820       type: PoolTypes.dynamic,
 821       worker: WorkerTypes.cluster,
 822       ready: true,
 823       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 824       minSize: expect.any(Number),
 825       maxSize: expect.any(Number),
 826       workerNodes: expect.any(Number),
 827       idleWorkerNodes: expect.any(Number),
 828       busyWorkerNodes: expect.any(Number),
 829       executedTasks: expect.any(Number),
 830       executingTasks: expect.any(Number),
 831       failedTasks: expect.any(Number)
 832     })
 833     await pool.destroy()
 834   })
 835
 836   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 837     const pool = new FixedThreadPool(
 838       numberOfWorkers,
 839       './tests/worker-files/thread/testWorker.js'
 840     )
 841     const promises = new Set()
 842     let poolBusy = 0
 843     let poolInfo
 844     pool.emitter.on(PoolEvents.busy, (info) => {
 845       ++poolBusy
 846       poolInfo = info
 847     })
 848     for (let i = 0; i < numberOfWorkers * 2; i++) {
 849       promises.add(pool.execute())
 850     }
 851     await Promise.all(promises)
 852     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 853     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 854     expect(poolBusy).toBe(numberOfWorkers + 1)
 855     expect(poolInfo).toStrictEqual({
 856       version,
 857       type: PoolTypes.fixed,
 858       worker: WorkerTypes.thread,
 859       ready: expect.any(Boolean),
 860       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 861       minSize: expect.any(Number),
 862       maxSize: expect.any(Number),
 863       workerNodes: expect.any(Number),
 864       idleWorkerNodes: expect.any(Number),
 865       busyWorkerNodes: expect.any(Number),
 866       executedTasks: expect.any(Number),
 867       executingTasks: expect.any(Number),
 868       failedTasks: expect.any(Number)
 869     })
 870     await pool.destroy()
 871   })
 872
 873   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 874     const pool = new DynamicThreadPool(
 875       Math.floor(numberOfWorkers / 2),
 876       numberOfWorkers,
 877       './tests/worker-files/thread/testWorker.js'
 878     )
 879     const promises = new Set()
 880     let poolFull = 0
 881     let poolInfo
 882     pool.emitter.on(PoolEvents.full, (info) => {
 883       ++poolFull
 884       poolInfo = info
 885     })
 886     for (let i = 0; i < numberOfWorkers * 2; i++) {
 887       promises.add(pool.execute())
 888     }
 889     await Promise.all(promises)
 890     expect(poolFull).toBe(1)
 891     expect(poolInfo).toStrictEqual({
 892       version,
 893       type: PoolTypes.dynamic,
 894       worker: WorkerTypes.thread,
 895       ready: expect.any(Boolean),
 896       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 897       minSize: expect.any(Number),
 898       maxSize: expect.any(Number),
 899       workerNodes: expect.any(Number),
 900       idleWorkerNodes: expect.any(Number),
 901       busyWorkerNodes: expect.any(Number),
 902       executedTasks: expect.any(Number),
 903       executingTasks: expect.any(Number),
 904       failedTasks: expect.any(Number)
 905     })
 906     await pool.destroy()
 907   })
 908
 909   it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
 910     const pool = new FixedThreadPool(
 911       numberOfWorkers,
 912       './tests/worker-files/thread/testWorker.js',
 913       {
 914         enableTasksQueue: true
 915       }
 916     )
 917     sinon.stub(pool, 'hasBackPressure').returns(true)
 918     const promises = new Set()
 919     let poolBackPressure = 0
 920     let poolInfo
 921     pool.emitter.on(PoolEvents.backPressure, (info) => {
 922       ++poolBackPressure
 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(poolBackPressure).toBe(2)
 930     expect(poolInfo).toStrictEqual({
 931       version,
 932       type: PoolTypes.fixed,
 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       maxQueuedTasks: expect.any(Number),
 944       queuedTasks: expect.any(Number),
 945       backPressure: true,
 946       failedTasks: expect.any(Number)
 947     })
 948     expect(pool.hasBackPressure.called).toBe(true)
 949     await pool.destroy()
 950   })
 951
 952   it('Verify that listTaskFunctions() is working', async () => {
 953     const dynamicThreadPool = new DynamicThreadPool(
 954       Math.floor(numberOfWorkers / 2),
 955       numberOfWorkers,
 956       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
 957     )
 958     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
 959     expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
 960       'default',
 961       'jsonIntegerSerialization',
 962       'factorial',
 963       'fibonacci'
 964     ])
 965     const fixedClusterPool = new FixedClusterPool(
 966       numberOfWorkers,
 967       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
 968     )
 969     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
 970     expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
 971       'default',
 972       'jsonIntegerSerialization',
 973       'factorial',
 974       'fibonacci'
 975     ])
 976   })
 977
 978   it('Verify that multiple task functions worker is working', async () => {
 979     const pool = new DynamicClusterPool(
 980       Math.floor(numberOfWorkers / 2),
 981       numberOfWorkers,
 982       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
 983     )
 984     const data = { n: 10 }
 985     const result0 = await pool.execute(data)
 986     expect(result0).toStrictEqual({ ok: 1 })
 987     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
 988     expect(result1).toStrictEqual({ ok: 1 })
 989     const result2 = await pool.execute(data, 'factorial')
 990     expect(result2).toBe(3628800)
 991     const result3 = await pool.execute(data, 'fibonacci')
 992     expect(result3).toBe(55)
 993     expect(pool.info.executingTasks).toBe(0)
 994     expect(pool.info.executedTasks).toBe(4)
 995     for (const workerNode of pool.workerNodes) {
 996       expect(workerNode.info.taskFunctions).toStrictEqual([
 997         'default',
 998         'jsonIntegerSerialization',
 999         'factorial',
1000         'fibonacci'
1001       ])
1002       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1003       for (const name of pool.listTaskFunctions()) {
1004         expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1005           tasks: {
1006             executed: expect.any(Number),
1007             executing: expect.any(Number),
1008             failed: 0,
1009             queued: 0
1010           },
1011           runTime: {
1012             history: expect.any(CircularArray)
1013           },
1014           waitTime: {
1015             history: expect.any(CircularArray)
1016           },
1017           elu: {
1018             idle: {
1019               history: expect.any(CircularArray)
1020             },
1021             active: {
1022               history: expect.any(CircularArray)
1023             }
1024           }
1025         })
1026         expect(
1027           workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1028         ).toBeGreaterThanOrEqual(0)
1029       }
1030     }
1031   })
1032 })