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

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