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: {
 267               choiceRetries: 'invalidChoiceRetries'
 268             }
 269           }
 270         )
 271     ).toThrowError(
 272       new TypeError(
 273         'Invalid worker choice strategy options: choice retries must be an integer'
 274       )
 275     )
 276     expect(
 277       () =>
 278         new FixedThreadPool(
 279           numberOfWorkers,
 280           './tests/worker-files/thread/testWorker.js',
 281           {
 282             workerChoiceStrategyOptions: {
 283               choiceRetries: -1
 284             }
 285           }
 286         )
 287     ).toThrowError(
 288       new RangeError(
 289         "Invalid worker choice strategy options: choice retries '-1' must be greater or equal than zero"
 290       )
 291     )
 292     expect(
 293       () =>
 294         new FixedThreadPool(
 295           numberOfWorkers,
 296           './tests/worker-files/thread/testWorker.js',
 297           {
 298             workerChoiceStrategyOptions: { weights: {} }
 299           }
 300         )
 301     ).toThrowError(
 302       new Error(
 303         'Invalid worker choice strategy options: must have a weight for each worker node'
 304       )
 305     )
 306     expect(
 307       () =>
 308         new FixedThreadPool(
 309           numberOfWorkers,
 310           './tests/worker-files/thread/testWorker.js',
 311           {
 312             workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
 313           }
 314         )
 315     ).toThrowError(
 316       new Error(
 317         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 318       )
 319     )
 320     expect(
 321       () =>
 322         new FixedThreadPool(
 323           numberOfWorkers,
 324           './tests/worker-files/thread/testWorker.js',
 325           {
 326             enableTasksQueue: true,
 327             tasksQueueOptions: { concurrency: 0 }
 328           }
 329         )
 330     ).toThrowError(
 331       new RangeError(
 332         'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
 333       )
 334     )
 335     expect(
 336       () =>
 337         new FixedThreadPool(
 338           numberOfWorkers,
 339           './tests/worker-files/thread/testWorker.js',
 340           {
 341             enableTasksQueue: true,
 342             tasksQueueOptions: 'invalidTasksQueueOptions'
 343           }
 344         )
 345     ).toThrowError(
 346       new TypeError('Invalid tasks queue options: must be a plain object')
 347     )
 348     expect(
 349       () =>
 350         new FixedThreadPool(
 351           numberOfWorkers,
 352           './tests/worker-files/thread/testWorker.js',
 353           {
 354             enableTasksQueue: true,
 355             tasksQueueOptions: { concurrency: 0.2 }
 356           }
 357         )
 358     ).toThrowError(
 359       new TypeError('Invalid worker node tasks concurrency: must be an integer')
 360     )
 361   })
 362
 363   it('Verify that pool worker choice strategy options can be set', async () => {
 364     const pool = new FixedThreadPool(
 365       numberOfWorkers,
 366       './tests/worker-files/thread/testWorker.js',
 367       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 368     )
 369     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 370       choiceRetries: 6,
 371       runTime: { median: false },
 372       waitTime: { median: false },
 373       elu: { median: false }
 374     })
 375     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 376       choiceRetries: 6,
 377       runTime: { median: false },
 378       waitTime: { median: false },
 379       elu: { median: false }
 380     })
 381     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 382       .workerChoiceStrategies) {
 383       expect(workerChoiceStrategy.opts).toStrictEqual({
 384         choiceRetries: 6,
 385         runTime: { median: false },
 386         waitTime: { median: false },
 387         elu: { median: false }
 388       })
 389     }
 390     expect(
 391       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 392     ).toStrictEqual({
 393       runTime: {
 394         aggregate: true,
 395         average: true,
 396         median: false
 397       },
 398       waitTime: {
 399         aggregate: false,
 400         average: false,
 401         median: false
 402       },
 403       elu: {
 404         aggregate: true,
 405         average: true,
 406         median: false
 407       }
 408     })
 409     pool.setWorkerChoiceStrategyOptions({
 410       runTime: { median: true },
 411       elu: { median: true }
 412     })
 413     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 414       choiceRetries: 6,
 415       runTime: { median: true },
 416       waitTime: { median: false },
 417       elu: { median: true }
 418     })
 419     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 420       choiceRetries: 6,
 421       runTime: { median: true },
 422       waitTime: { median: false },
 423       elu: { median: true }
 424     })
 425     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 426       .workerChoiceStrategies) {
 427       expect(workerChoiceStrategy.opts).toStrictEqual({
 428         choiceRetries: 6,
 429         runTime: { median: true },
 430         waitTime: { median: false },
 431         elu: { median: true }
 432       })
 433     }
 434     expect(
 435       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 436     ).toStrictEqual({
 437       runTime: {
 438         aggregate: true,
 439         average: false,
 440         median: true
 441       },
 442       waitTime: {
 443         aggregate: false,
 444         average: false,
 445         median: false
 446       },
 447       elu: {
 448         aggregate: true,
 449         average: false,
 450         median: true
 451       }
 452     })
 453     pool.setWorkerChoiceStrategyOptions({
 454       runTime: { median: false },
 455       elu: { median: false }
 456     })
 457     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 458       choiceRetries: 6,
 459       runTime: { median: false },
 460       waitTime: { median: false },
 461       elu: { median: false }
 462     })
 463     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 464       choiceRetries: 6,
 465       runTime: { median: false },
 466       waitTime: { median: false },
 467       elu: { median: false }
 468     })
 469     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 470       .workerChoiceStrategies) {
 471       expect(workerChoiceStrategy.opts).toStrictEqual({
 472         choiceRetries: 6,
 473         runTime: { median: false },
 474         waitTime: { median: false },
 475         elu: { median: false }
 476       })
 477     }
 478     expect(
 479       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 480     ).toStrictEqual({
 481       runTime: {
 482         aggregate: true,
 483         average: true,
 484         median: false
 485       },
 486       waitTime: {
 487         aggregate: false,
 488         average: false,
 489         median: false
 490       },
 491       elu: {
 492         aggregate: true,
 493         average: true,
 494         median: false
 495       }
 496     })
 497     expect(() =>
 498       pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
 499     ).toThrowError(
 500       new TypeError(
 501         'Invalid worker choice strategy options: must be a plain object'
 502       )
 503     )
 504     expect(() =>
 505       pool.setWorkerChoiceStrategyOptions({
 506         choiceRetries: 'invalidChoiceRetries'
 507       })
 508     ).toThrowError(
 509       new TypeError(
 510         'Invalid worker choice strategy options: choice retries must be an integer'
 511       )
 512     )
 513     expect(() =>
 514       pool.setWorkerChoiceStrategyOptions({ choiceRetries: -1 })
 515     ).toThrowError(
 516       new RangeError(
 517         "Invalid worker choice strategy options: choice retries '-1' must be greater or equal than zero"
 518       )
 519     )
 520     expect(() =>
 521       pool.setWorkerChoiceStrategyOptions({ weights: {} })
 522     ).toThrowError(
 523       new Error(
 524         'Invalid worker choice strategy options: must have a weight for each worker node'
 525       )
 526     )
 527     expect(() =>
 528       pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
 529     ).toThrowError(
 530       new Error(
 531         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 532       )
 533     )
 534     await pool.destroy()
 535   })
 536
 537   it('Verify that pool tasks queue can be enabled/disabled', async () => {
 538     const pool = new FixedThreadPool(
 539       numberOfWorkers,
 540       './tests/worker-files/thread/testWorker.js'
 541     )
 542     expect(pool.opts.enableTasksQueue).toBe(false)
 543     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 544     pool.enableTasksQueue(true)
 545     expect(pool.opts.enableTasksQueue).toBe(true)
 546     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 547       concurrency: 1,
 548       size: 4
 549     })
 550     pool.enableTasksQueue(true, { concurrency: 2 })
 551     expect(pool.opts.enableTasksQueue).toBe(true)
 552     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 553       concurrency: 2,
 554       size: 4
 555     })
 556     pool.enableTasksQueue(false)
 557     expect(pool.opts.enableTasksQueue).toBe(false)
 558     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 559     await pool.destroy()
 560   })
 561
 562   it('Verify that pool tasks queue options can be set', async () => {
 563     const pool = new FixedThreadPool(
 564       numberOfWorkers,
 565       './tests/worker-files/thread/testWorker.js',
 566       { enableTasksQueue: true }
 567     )
 568     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 569       concurrency: 1,
 570       size: 4
 571     })
 572     pool.setTasksQueueOptions({ concurrency: 2 })
 573     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 574       concurrency: 2,
 575       size: 4
 576     })
 577     expect(() =>
 578       pool.setTasksQueueOptions('invalidTasksQueueOptions')
 579     ).toThrowError(
 580       new TypeError('Invalid tasks queue options: must be a plain object')
 581     )
 582     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
 583       new RangeError(
 584         'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
 585       )
 586     )
 587     expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError(
 588       new RangeError(
 589         'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
 590       )
 591     )
 592     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
 593       new TypeError('Invalid worker node tasks concurrency: must be an integer')
 594     )
 595     expect(() => pool.setTasksQueueOptions({ queueMaxSize: 2 })).toThrowError(
 596       new Error(
 597         'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
 598       )
 599     )
 600     expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
 601       new RangeError(
 602         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
 603       )
 604     )
 605     expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError(
 606       new RangeError(
 607         'Invalid worker node tasks queue size: -1 is a negative integer or zero'
 608       )
 609     )
 610     expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError(
 611       new TypeError('Invalid worker node tasks queue size: must be an integer')
 612     )
 613     await pool.destroy()
 614   })
 615
 616   it('Verify that pool info is set', async () => {
 617     let pool = new FixedThreadPool(
 618       numberOfWorkers,
 619       './tests/worker-files/thread/testWorker.js'
 620     )
 621     expect(pool.info).toStrictEqual({
 622       version,
 623       type: PoolTypes.fixed,
 624       worker: WorkerTypes.thread,
 625       ready: true,
 626       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 627       minSize: numberOfWorkers,
 628       maxSize: numberOfWorkers,
 629       workerNodes: numberOfWorkers,
 630       idleWorkerNodes: numberOfWorkers,
 631       busyWorkerNodes: 0,
 632       executedTasks: 0,
 633       executingTasks: 0,
 634       failedTasks: 0
 635     })
 636     await pool.destroy()
 637     pool = new DynamicClusterPool(
 638       Math.floor(numberOfWorkers / 2),
 639       numberOfWorkers,
 640       './tests/worker-files/cluster/testWorker.js'
 641     )
 642     expect(pool.info).toStrictEqual({
 643       version,
 644       type: PoolTypes.dynamic,
 645       worker: WorkerTypes.cluster,
 646       ready: true,
 647       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 648       minSize: Math.floor(numberOfWorkers / 2),
 649       maxSize: numberOfWorkers,
 650       workerNodes: Math.floor(numberOfWorkers / 2),
 651       idleWorkerNodes: Math.floor(numberOfWorkers / 2),
 652       busyWorkerNodes: 0,
 653       executedTasks: 0,
 654       executingTasks: 0,
 655       failedTasks: 0
 656     })
 657     await pool.destroy()
 658   })
 659
 660   it('Verify that pool worker tasks usage are initialized', async () => {
 661     const pool = new FixedClusterPool(
 662       numberOfWorkers,
 663       './tests/worker-files/cluster/testWorker.js'
 664     )
 665     for (const workerNode of pool.workerNodes) {
 666       expect(workerNode.usage).toStrictEqual({
 667         tasks: {
 668           executed: 0,
 669           executing: 0,
 670           queued: 0,
 671           maxQueued: 0,
 672           stolen: 0,
 673           failed: 0
 674         },
 675         runTime: {
 676           history: expect.any(CircularArray)
 677         },
 678         waitTime: {
 679           history: expect.any(CircularArray)
 680         },
 681         elu: {
 682           idle: {
 683             history: expect.any(CircularArray)
 684           },
 685           active: {
 686             history: expect.any(CircularArray)
 687           }
 688         }
 689       })
 690     }
 691     await pool.destroy()
 692   })
 693
 694   it('Verify that pool worker tasks queue are initialized', async () => {
 695     let pool = new FixedClusterPool(
 696       numberOfWorkers,
 697       './tests/worker-files/cluster/testWorker.js'
 698     )
 699     for (const workerNode of pool.workerNodes) {
 700       expect(workerNode.tasksQueue).toBeDefined()
 701       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 702       expect(workerNode.tasksQueue.size).toBe(0)
 703       expect(workerNode.tasksQueue.maxSize).toBe(0)
 704     }
 705     await pool.destroy()
 706     pool = new DynamicThreadPool(
 707       Math.floor(numberOfWorkers / 2),
 708       numberOfWorkers,
 709       './tests/worker-files/thread/testWorker.js'
 710     )
 711     for (const workerNode of pool.workerNodes) {
 712       expect(workerNode.tasksQueue).toBeDefined()
 713       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 714       expect(workerNode.tasksQueue.size).toBe(0)
 715       expect(workerNode.tasksQueue.maxSize).toBe(0)
 716     }
 717   })
 718
 719   it('Verify that pool worker info are initialized', async () => {
 720     let pool = new FixedClusterPool(
 721       numberOfWorkers,
 722       './tests/worker-files/cluster/testWorker.js'
 723     )
 724     for (const workerNode of pool.workerNodes) {
 725       expect(workerNode.info).toStrictEqual({
 726         id: expect.any(Number),
 727         type: WorkerTypes.cluster,
 728         dynamic: false,
 729         ready: true
 730       })
 731     }
 732     await pool.destroy()
 733     pool = new DynamicThreadPool(
 734       Math.floor(numberOfWorkers / 2),
 735       numberOfWorkers,
 736       './tests/worker-files/thread/testWorker.js'
 737     )
 738     for (const workerNode of pool.workerNodes) {
 739       expect(workerNode.info).toStrictEqual({
 740         id: expect.any(Number),
 741         type: WorkerTypes.thread,
 742         dynamic: false,
 743         ready: true
 744       })
 745     }
 746   })
 747
 748   it('Verify that pool execute() arguments are checked', async () => {
 749     const pool = new FixedClusterPool(
 750       numberOfWorkers,
 751       './tests/worker-files/cluster/testWorker.js'
 752     )
 753     await expect(pool.execute(undefined, 0)).rejects.toThrowError(
 754       new TypeError('name argument must be a string')
 755     )
 756     await expect(pool.execute(undefined, '')).rejects.toThrowError(
 757       new TypeError('name argument must not be an empty string')
 758     )
 759     await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
 760       new TypeError('transferList argument must be an array')
 761     )
 762     await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
 763       "Task function 'unknown' not found"
 764     )
 765     await pool.destroy()
 766     await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
 767       new Error('Cannot execute a task on destroyed pool')
 768     )
 769   })
 770
 771   it('Verify that pool worker tasks usage are computed', async () => {
 772     const pool = new FixedClusterPool(
 773       numberOfWorkers,
 774       './tests/worker-files/cluster/testWorker.js'
 775     )
 776     const promises = new Set()
 777     const maxMultiplier = 2
 778     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 779       promises.add(pool.execute())
 780     }
 781     for (const workerNode of pool.workerNodes) {
 782       expect(workerNode.usage).toStrictEqual({
 783         tasks: {
 784           executed: 0,
 785           executing: maxMultiplier,
 786           queued: 0,
 787           maxQueued: 0,
 788           stolen: 0,
 789           failed: 0
 790         },
 791         runTime: {
 792           history: expect.any(CircularArray)
 793         },
 794         waitTime: {
 795           history: expect.any(CircularArray)
 796         },
 797         elu: {
 798           idle: {
 799             history: expect.any(CircularArray)
 800           },
 801           active: {
 802             history: expect.any(CircularArray)
 803           }
 804         }
 805       })
 806     }
 807     await Promise.all(promises)
 808     for (const workerNode of pool.workerNodes) {
 809       expect(workerNode.usage).toStrictEqual({
 810         tasks: {
 811           executed: maxMultiplier,
 812           executing: 0,
 813           queued: 0,
 814           maxQueued: 0,
 815           stolen: 0,
 816           failed: 0
 817         },
 818         runTime: {
 819           history: expect.any(CircularArray)
 820         },
 821         waitTime: {
 822           history: expect.any(CircularArray)
 823         },
 824         elu: {
 825           idle: {
 826             history: expect.any(CircularArray)
 827           },
 828           active: {
 829             history: expect.any(CircularArray)
 830           }
 831         }
 832       })
 833     }
 834     await pool.destroy()
 835   })
 836
 837   it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
 838     const pool = new DynamicThreadPool(
 839       Math.floor(numberOfWorkers / 2),
 840       numberOfWorkers,
 841       './tests/worker-files/thread/testWorker.js'
 842     )
 843     const promises = new Set()
 844     const maxMultiplier = 2
 845     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 846       promises.add(pool.execute())
 847     }
 848     await Promise.all(promises)
 849     for (const workerNode of pool.workerNodes) {
 850       expect(workerNode.usage).toStrictEqual({
 851         tasks: {
 852           executed: expect.any(Number),
 853           executing: 0,
 854           queued: 0,
 855           maxQueued: 0,
 856           stolen: 0,
 857           failed: 0
 858         },
 859         runTime: {
 860           history: expect.any(CircularArray)
 861         },
 862         waitTime: {
 863           history: expect.any(CircularArray)
 864         },
 865         elu: {
 866           idle: {
 867             history: expect.any(CircularArray)
 868           },
 869           active: {
 870             history: expect.any(CircularArray)
 871           }
 872         }
 873       })
 874       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
 875       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
 876         numberOfWorkers * maxMultiplier
 877       )
 878       expect(workerNode.usage.runTime.history.length).toBe(0)
 879       expect(workerNode.usage.waitTime.history.length).toBe(0)
 880       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 881       expect(workerNode.usage.elu.active.history.length).toBe(0)
 882     }
 883     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 884     for (const workerNode of pool.workerNodes) {
 885       expect(workerNode.usage).toStrictEqual({
 886         tasks: {
 887           executed: 0,
 888           executing: 0,
 889           queued: 0,
 890           maxQueued: 0,
 891           stolen: 0,
 892           failed: 0
 893         },
 894         runTime: {
 895           history: expect.any(CircularArray)
 896         },
 897         waitTime: {
 898           history: expect.any(CircularArray)
 899         },
 900         elu: {
 901           idle: {
 902             history: expect.any(CircularArray)
 903           },
 904           active: {
 905             history: expect.any(CircularArray)
 906           }
 907         }
 908       })
 909       expect(workerNode.usage.runTime.history.length).toBe(0)
 910       expect(workerNode.usage.waitTime.history.length).toBe(0)
 911       expect(workerNode.usage.elu.idle.history.length).toBe(0)
 912       expect(workerNode.usage.elu.active.history.length).toBe(0)
 913     }
 914     await pool.destroy()
 915   })
 916
 917   it("Verify that pool event emitter 'ready' event can register a callback", async () => {
 918     const pool = new DynamicClusterPool(
 919       Math.floor(numberOfWorkers / 2),
 920       numberOfWorkers,
 921       './tests/worker-files/cluster/testWorker.js'
 922     )
 923     let poolInfo
 924     let poolReady = 0
 925     pool.emitter.on(PoolEvents.ready, (info) => {
 926       ++poolReady
 927       poolInfo = info
 928     })
 929     await waitPoolEvents(pool, PoolEvents.ready, 1)
 930     expect(poolReady).toBe(1)
 931     expect(poolInfo).toStrictEqual({
 932       version,
 933       type: PoolTypes.dynamic,
 934       worker: WorkerTypes.cluster,
 935       ready: true,
 936       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 937       minSize: expect.any(Number),
 938       maxSize: expect.any(Number),
 939       workerNodes: expect.any(Number),
 940       idleWorkerNodes: expect.any(Number),
 941       busyWorkerNodes: expect.any(Number),
 942       executedTasks: expect.any(Number),
 943       executingTasks: expect.any(Number),
 944       failedTasks: expect.any(Number)
 945     })
 946     await pool.destroy()
 947   })
 948
 949   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 950     const pool = new FixedThreadPool(
 951       numberOfWorkers,
 952       './tests/worker-files/thread/testWorker.js'
 953     )
 954     const promises = new Set()
 955     let poolBusy = 0
 956     let poolInfo
 957     pool.emitter.on(PoolEvents.busy, (info) => {
 958       ++poolBusy
 959       poolInfo = info
 960     })
 961     for (let i = 0; i < numberOfWorkers * 2; i++) {
 962       promises.add(pool.execute())
 963     }
 964     await Promise.all(promises)
 965     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 966     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 967     expect(poolBusy).toBe(numberOfWorkers + 1)
 968     expect(poolInfo).toStrictEqual({
 969       version,
 970       type: PoolTypes.fixed,
 971       worker: WorkerTypes.thread,
 972       ready: expect.any(Boolean),
 973       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 974       minSize: expect.any(Number),
 975       maxSize: expect.any(Number),
 976       workerNodes: expect.any(Number),
 977       idleWorkerNodes: expect.any(Number),
 978       busyWorkerNodes: expect.any(Number),
 979       executedTasks: expect.any(Number),
 980       executingTasks: expect.any(Number),
 981       failedTasks: expect.any(Number)
 982     })
 983     await pool.destroy()
 984   })
 985
 986   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 987     const pool = new DynamicThreadPool(
 988       Math.floor(numberOfWorkers / 2),
 989       numberOfWorkers,
 990       './tests/worker-files/thread/testWorker.js'
 991     )
 992     const promises = new Set()
 993     let poolFull = 0
 994     let poolInfo
 995     pool.emitter.on(PoolEvents.full, (info) => {
 996       ++poolFull
 997       poolInfo = info
 998     })
 999     for (let i = 0; i < numberOfWorkers * 2; i++) {
1000       promises.add(pool.execute())
1001     }
1002     await Promise.all(promises)
1003     expect(poolFull).toBe(1)
1004     expect(poolInfo).toStrictEqual({
1005       version,
1006       type: PoolTypes.dynamic,
1007       worker: WorkerTypes.thread,
1008       ready: expect.any(Boolean),
1009       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1010       minSize: expect.any(Number),
1011       maxSize: expect.any(Number),
1012       workerNodes: expect.any(Number),
1013       idleWorkerNodes: expect.any(Number),
1014       busyWorkerNodes: expect.any(Number),
1015       executedTasks: expect.any(Number),
1016       executingTasks: expect.any(Number),
1017       failedTasks: expect.any(Number)
1018     })
1019     await pool.destroy()
1020   })
1021
1022   it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
1023     const pool = new FixedThreadPool(
1024       numberOfWorkers,
1025       './tests/worker-files/thread/testWorker.js',
1026       {
1027         enableTasksQueue: true
1028       }
1029     )
1030     sinon.stub(pool, 'hasBackPressure').returns(true)
1031     const promises = new Set()
1032     let poolBackPressure = 0
1033     let poolInfo
1034     pool.emitter.on(PoolEvents.backPressure, (info) => {
1035       ++poolBackPressure
1036       poolInfo = info
1037     })
1038     for (let i = 0; i < numberOfWorkers + 1; i++) {
1039       promises.add(pool.execute())
1040     }
1041     await Promise.all(promises)
1042     expect(poolBackPressure).toBe(1)
1043     expect(poolInfo).toStrictEqual({
1044       version,
1045       type: PoolTypes.fixed,
1046       worker: WorkerTypes.thread,
1047       ready: expect.any(Boolean),
1048       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1049       minSize: expect.any(Number),
1050       maxSize: expect.any(Number),
1051       workerNodes: expect.any(Number),
1052       idleWorkerNodes: expect.any(Number),
1053       busyWorkerNodes: expect.any(Number),
1054       executedTasks: expect.any(Number),
1055       executingTasks: expect.any(Number),
1056       maxQueuedTasks: expect.any(Number),
1057       queuedTasks: expect.any(Number),
1058       backPressure: true,
1059       stolenTasks: expect.any(Number),
1060       failedTasks: expect.any(Number)
1061     })
1062     expect(pool.hasBackPressure.called).toBe(true)
1063     await pool.destroy()
1064   })
1065
1066   it('Verify that listTaskFunctions() is working', async () => {
1067     const dynamicThreadPool = new DynamicThreadPool(
1068       Math.floor(numberOfWorkers / 2),
1069       numberOfWorkers,
1070       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.js'
1071     )
1072     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1073     expect(dynamicThreadPool.listTaskFunctions()).toStrictEqual([
1074       'default',
1075       'jsonIntegerSerialization',
1076       'factorial',
1077       'fibonacci'
1078     ])
1079     const fixedClusterPool = new FixedClusterPool(
1080       numberOfWorkers,
1081       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1082     )
1083     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1084     expect(fixedClusterPool.listTaskFunctions()).toStrictEqual([
1085       'default',
1086       'jsonIntegerSerialization',
1087       'factorial',
1088       'fibonacci'
1089     ])
1090   })
1091
1092   it('Verify that multiple task functions worker is working', async () => {
1093     const pool = new DynamicClusterPool(
1094       Math.floor(numberOfWorkers / 2),
1095       numberOfWorkers,
1096       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1097     )
1098     const data = { n: 10 }
1099     const result0 = await pool.execute(data)
1100     expect(result0).toStrictEqual({ ok: 1 })
1101     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1102     expect(result1).toStrictEqual({ ok: 1 })
1103     const result2 = await pool.execute(data, 'factorial')
1104     expect(result2).toBe(3628800)
1105     const result3 = await pool.execute(data, 'fibonacci')
1106     expect(result3).toBe(55)
1107     expect(pool.info.executingTasks).toBe(0)
1108     expect(pool.info.executedTasks).toBe(4)
1109     for (const workerNode of pool.workerNodes) {
1110       expect(workerNode.info.taskFunctions).toStrictEqual([
1111         'default',
1112         'jsonIntegerSerialization',
1113         'factorial',
1114         'fibonacci'
1115       ])
1116       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1117       for (const name of pool.listTaskFunctions()) {
1118         expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1119           tasks: {
1120             executed: expect.any(Number),
1121             executing: expect.any(Number),
1122             failed: 0,
1123             queued: 0,
1124             stolen: 0
1125           },
1126           runTime: {
1127             history: expect.any(CircularArray)
1128           },
1129           waitTime: {
1130             history: expect.any(CircularArray)
1131           },
1132           elu: {
1133             idle: {
1134               history: expect.any(CircularArray)
1135             },
1136             active: {
1137               history: expect.any(CircularArray)
1138             }
1139           }
1140         })
1141         expect(
1142           workerNode.getTaskFunctionWorkerUsage(name).tasks.executing
1143         ).toBeGreaterThanOrEqual(0)
1144       }
1145     }
1146   })
1147 })