tests/pools/abstract/abstract-pool.test.js

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