tests/pools/abstract-pool.test.mjs

   1 import { EventEmitterAsyncResource } from 'node:events'
   2 import { dirname, join } from 'node:path'
   3 import { readFileSync } from 'node:fs'
   4 import { fileURLToPath } from 'node:url'
   5 import { createHook, executionAsyncId } from 'node:async_hooks'
   6 import { expect } from 'expect'
   7 import { restore, stub } from 'sinon'
   8 import {
   9   DynamicClusterPool,
  10   DynamicThreadPool,
  11   FixedClusterPool,
  12   FixedThreadPool,
  13   PoolEvents,
  14   PoolTypes,
  15   WorkerChoiceStrategies,
  16   WorkerTypes
  17 } from '../../lib/index.js'
  18 import { CircularArray } from '../../lib/circular-array.js'
  19 import { Deque } from '../../lib/deque.js'
  20 import { DEFAULT_TASK_NAME } from '../../lib/utils.js'
  21 import { waitPoolEvents } from '../test-utils.js'
  22 import { WorkerNode } from '../../lib/pools/worker-node.js'
  23
  24 describe('Abstract pool test suite', () => {
  25   const version = JSON.parse(
  26     readFileSync(
  27       join(dirname(fileURLToPath(import.meta.url)), '../..', 'package.json'),
  28       'utf8'
  29     )
  30   ).version
  31   const numberOfWorkers = 2
  32   class StubPoolWithIsMain extends FixedThreadPool {
  33     isMain () {
  34       return false
  35     }
  36   }
  37
  38   afterEach(() => {
  39     restore()
  40   })
  41
  42   it('Verify that pool can be created and destroyed', async () => {
  43     const pool = new FixedThreadPool(
  44       numberOfWorkers,
  45       './tests/worker-files/thread/testWorker.mjs'
  46     )
  47     expect(pool).toBeInstanceOf(FixedThreadPool)
  48     await pool.destroy()
  49   })
  50
  51   it('Verify that pool cannot be created from a non main thread/process', () => {
  52     expect(
  53       () =>
  54         new StubPoolWithIsMain(
  55           numberOfWorkers,
  56           './tests/worker-files/thread/testWorker.mjs',
  57           {
  58             errorHandler: e => console.error(e)
  59           }
  60         )
  61     ).toThrow(
  62       new Error(
  63         'Cannot start a pool from a worker with the same type as the pool'
  64       )
  65     )
  66   })
  67
  68   it('Verify that pool statuses properties are set', async () => {
  69     const pool = new FixedThreadPool(
  70       numberOfWorkers,
  71       './tests/worker-files/thread/testWorker.mjs'
  72     )
  73     expect(pool.started).toBe(true)
  74     expect(pool.starting).toBe(false)
  75     expect(pool.destroying).toBe(false)
  76     await pool.destroy()
  77   })
  78
  79   it('Verify that filePath is checked', () => {
  80     expect(() => new FixedThreadPool(numberOfWorkers)).toThrow(
  81       new TypeError('The worker file path must be specified')
  82     )
  83     expect(() => new FixedThreadPool(numberOfWorkers, 0)).toThrow(
  84       new TypeError('The worker file path must be a string')
  85     )
  86     expect(
  87       () => new FixedThreadPool(numberOfWorkers, './dummyWorker.ts')
  88     ).toThrow(new Error("Cannot find the worker file './dummyWorker.ts'"))
  89   })
  90
  91   it('Verify that numberOfWorkers is checked', () => {
  92     expect(
  93       () =>
  94         new FixedThreadPool(
  95           undefined,
  96           './tests/worker-files/thread/testWorker.mjs'
  97         )
  98     ).toThrow(
  99       new Error(
 100         'Cannot instantiate a pool without specifying the number of workers'
 101       )
 102     )
 103   })
 104
 105   it('Verify that a negative number of workers is checked', () => {
 106     expect(
 107       () =>
 108         new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
 109     ).toThrow(
 110       new RangeError(
 111         'Cannot instantiate a pool with a negative number of workers'
 112       )
 113     )
 114   })
 115
 116   it('Verify that a non integer number of workers is checked', () => {
 117     expect(
 118       () =>
 119         new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.mjs')
 120     ).toThrow(
 121       new TypeError(
 122         'Cannot instantiate a pool with a non safe integer number of workers'
 123       )
 124     )
 125   })
 126
 127   it('Verify that pool arguments number and pool type are checked', () => {
 128     expect(
 129       () =>
 130         new FixedThreadPool(
 131           numberOfWorkers,
 132           './tests/worker-files/thread/testWorker.mjs',
 133           undefined,
 134           numberOfWorkers * 2
 135         )
 136     ).toThrow(
 137       new Error(
 138         'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
 139       )
 140     )
 141   })
 142
 143   it('Verify that dynamic pool sizing is checked', () => {
 144     expect(
 145       () =>
 146         new DynamicClusterPool(
 147           1,
 148           undefined,
 149           './tests/worker-files/cluster/testWorker.js'
 150         )
 151     ).toThrow(
 152       new TypeError(
 153         'Cannot instantiate a dynamic pool without specifying the maximum pool size'
 154       )
 155     )
 156     expect(
 157       () =>
 158         new DynamicThreadPool(
 159           0.5,
 160           1,
 161           './tests/worker-files/thread/testWorker.mjs'
 162         )
 163     ).toThrow(
 164       new TypeError(
 165         'Cannot instantiate a pool with a non safe integer number of workers'
 166       )
 167     )
 168     expect(
 169       () =>
 170         new DynamicClusterPool(
 171           0,
 172           0.5,
 173           './tests/worker-files/cluster/testWorker.js'
 174         )
 175     ).toThrow(
 176       new TypeError(
 177         'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
 178       )
 179     )
 180     expect(
 181       () =>
 182         new DynamicThreadPool(
 183           2,
 184           1,
 185           './tests/worker-files/thread/testWorker.mjs'
 186         )
 187     ).toThrow(
 188       new RangeError(
 189         'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
 190       )
 191     )
 192     expect(
 193       () =>
 194         new DynamicThreadPool(
 195           0,
 196           0,
 197           './tests/worker-files/thread/testWorker.mjs'
 198         )
 199     ).toThrow(
 200       new RangeError(
 201         'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
 202       )
 203     )
 204     expect(
 205       () =>
 206         new DynamicClusterPool(
 207           1,
 208           1,
 209           './tests/worker-files/cluster/testWorker.js'
 210         )
 211     ).toThrow(
 212       new RangeError(
 213         'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
 214       )
 215     )
 216   })
 217
 218   it('Verify that pool options are checked', async () => {
 219     let pool = new FixedThreadPool(
 220       numberOfWorkers,
 221       './tests/worker-files/thread/testWorker.mjs'
 222     )
 223     expect(pool.emitter).toBeInstanceOf(EventEmitterAsyncResource)
 224     expect(pool.opts).toStrictEqual({
 225       startWorkers: true,
 226       enableEvents: true,
 227       restartWorkerOnError: true,
 228       enableTasksQueue: false,
 229       workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN
 230     })
 231     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 232       retries:
 233         pool.info.maxSize +
 234         Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
 235       runTime: { median: false },
 236       waitTime: { median: false },
 237       elu: { median: false },
 238       weights: expect.objectContaining({
 239         0: expect.any(Number),
 240         [pool.info.maxSize - 1]: expect.any(Number)
 241       })
 242     })
 243     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 244       .workerChoiceStrategies) {
 245       expect(workerChoiceStrategy.opts).toStrictEqual(
 246         expect.objectContaining({
 247           retries:
 248             pool.info.maxSize +
 249             Object.keys(workerChoiceStrategy.opts.weights).length,
 250           runTime: { median: false },
 251           waitTime: { median: false },
 252           elu: { median: false }
 253           // weights: expect.objectContaining({
 254           //   0: expect.any(Number),
 255           //   [pool.info.maxSize - 1]: expect.any(Number)
 256           // })
 257         })
 258       )
 259     }
 260     await pool.destroy()
 261     const testHandler = () => console.info('test handler executed')
 262     pool = new FixedThreadPool(
 263       numberOfWorkers,
 264       './tests/worker-files/thread/testWorker.mjs',
 265       {
 266         workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
 267         workerChoiceStrategyOptions: {
 268           runTime: { median: true },
 269           weights: { 0: 300, 1: 200 }
 270         },
 271         enableEvents: false,
 272         restartWorkerOnError: false,
 273         enableTasksQueue: true,
 274         tasksQueueOptions: { concurrency: 2 },
 275         messageHandler: testHandler,
 276         errorHandler: testHandler,
 277         onlineHandler: testHandler,
 278         exitHandler: testHandler
 279       }
 280     )
 281     expect(pool.emitter).toBeUndefined()
 282     expect(pool.opts).toStrictEqual({
 283       startWorkers: true,
 284       enableEvents: false,
 285       restartWorkerOnError: false,
 286       enableTasksQueue: true,
 287       tasksQueueOptions: {
 288         concurrency: 2,
 289         size: Math.pow(numberOfWorkers, 2),
 290         taskStealing: true,
 291         tasksStealingOnBackPressure: true,
 292         tasksFinishedTimeout: 2000
 293       },
 294       workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
 295       workerChoiceStrategyOptions: {
 296         runTime: { median: true },
 297         weights: { 0: 300, 1: 200 }
 298       },
 299       onlineHandler: testHandler,
 300       messageHandler: testHandler,
 301       errorHandler: testHandler,
 302       exitHandler: testHandler
 303     })
 304     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 305       retries:
 306         pool.info.maxSize +
 307         Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length,
 308       runTime: { median: true },
 309       waitTime: { median: false },
 310       elu: { median: false },
 311       weights: { 0: 300, 1: 200 }
 312     })
 313     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 314       .workerChoiceStrategies) {
 315       expect(workerChoiceStrategy.opts).toStrictEqual({
 316         retries:
 317           pool.info.maxSize +
 318           Object.keys(pool.opts.workerChoiceStrategyOptions.weights).length,
 319         runTime: { median: true },
 320         waitTime: { median: false },
 321         elu: { median: false },
 322         weights: { 0: 300, 1: 200 }
 323       })
 324     }
 325     await pool.destroy()
 326   })
 327
 328   it('Verify that pool options are validated', () => {
 329     expect(
 330       () =>
 331         new FixedThreadPool(
 332           numberOfWorkers,
 333           './tests/worker-files/thread/testWorker.mjs',
 334           {
 335             workerChoiceStrategy: 'invalidStrategy'
 336           }
 337         )
 338     ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'"))
 339     expect(
 340       () =>
 341         new FixedThreadPool(
 342           numberOfWorkers,
 343           './tests/worker-files/thread/testWorker.mjs',
 344           {
 345             workerChoiceStrategyOptions: { weights: {} }
 346           }
 347         )
 348     ).toThrow(
 349       new Error(
 350         'Invalid worker choice strategy options: must have a weight for each worker node'
 351       )
 352     )
 353     expect(
 354       () =>
 355         new FixedThreadPool(
 356           numberOfWorkers,
 357           './tests/worker-files/thread/testWorker.mjs',
 358           {
 359             workerChoiceStrategyOptions: { measurement: 'invalidMeasurement' }
 360           }
 361         )
 362     ).toThrow(
 363       new Error(
 364         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 365       )
 366     )
 367     expect(
 368       () =>
 369         new FixedThreadPool(
 370           numberOfWorkers,
 371           './tests/worker-files/thread/testWorker.mjs',
 372           {
 373             enableTasksQueue: true,
 374             tasksQueueOptions: 'invalidTasksQueueOptions'
 375           }
 376         )
 377     ).toThrow(
 378       new TypeError('Invalid tasks queue options: must be a plain object')
 379     )
 380     expect(
 381       () =>
 382         new FixedThreadPool(
 383           numberOfWorkers,
 384           './tests/worker-files/thread/testWorker.mjs',
 385           {
 386             enableTasksQueue: true,
 387             tasksQueueOptions: { concurrency: 0 }
 388           }
 389         )
 390     ).toThrow(
 391       new RangeError(
 392         'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
 393       )
 394     )
 395     expect(
 396       () =>
 397         new FixedThreadPool(
 398           numberOfWorkers,
 399           './tests/worker-files/thread/testWorker.mjs',
 400           {
 401             enableTasksQueue: true,
 402             tasksQueueOptions: { concurrency: -1 }
 403           }
 404         )
 405     ).toThrow(
 406       new RangeError(
 407         'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
 408       )
 409     )
 410     expect(
 411       () =>
 412         new FixedThreadPool(
 413           numberOfWorkers,
 414           './tests/worker-files/thread/testWorker.mjs',
 415           {
 416             enableTasksQueue: true,
 417             tasksQueueOptions: { concurrency: 0.2 }
 418           }
 419         )
 420     ).toThrow(
 421       new TypeError('Invalid worker node tasks concurrency: must be an integer')
 422     )
 423     expect(
 424       () =>
 425         new FixedThreadPool(
 426           numberOfWorkers,
 427           './tests/worker-files/thread/testWorker.mjs',
 428           {
 429             enableTasksQueue: true,
 430             tasksQueueOptions: { size: 0 }
 431           }
 432         )
 433     ).toThrow(
 434       new RangeError(
 435         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
 436       )
 437     )
 438     expect(
 439       () =>
 440         new FixedThreadPool(
 441           numberOfWorkers,
 442           './tests/worker-files/thread/testWorker.mjs',
 443           {
 444             enableTasksQueue: true,
 445             tasksQueueOptions: { size: -1 }
 446           }
 447         )
 448     ).toThrow(
 449       new RangeError(
 450         'Invalid worker node tasks queue size: -1 is a negative integer or zero'
 451       )
 452     )
 453     expect(
 454       () =>
 455         new FixedThreadPool(
 456           numberOfWorkers,
 457           './tests/worker-files/thread/testWorker.mjs',
 458           {
 459             enableTasksQueue: true,
 460             tasksQueueOptions: { size: 0.2 }
 461           }
 462         )
 463     ).toThrow(
 464       new TypeError('Invalid worker node tasks queue size: must be an integer')
 465     )
 466   })
 467
 468   it('Verify that pool worker choice strategy options can be set', async () => {
 469     const pool = new FixedThreadPool(
 470       numberOfWorkers,
 471       './tests/worker-files/thread/testWorker.mjs',
 472       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 473     )
 474     expect(pool.opts.workerChoiceStrategyOptions).toBeUndefined()
 475     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 476       retries:
 477         pool.info.maxSize +
 478         Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
 479       runTime: { median: false },
 480       waitTime: { median: false },
 481       elu: { median: false },
 482       weights: expect.objectContaining({
 483         0: expect.any(Number),
 484         [pool.info.maxSize - 1]: expect.any(Number)
 485       })
 486     })
 487     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 488       .workerChoiceStrategies) {
 489       expect(workerChoiceStrategy.opts).toStrictEqual(
 490         expect.objectContaining({
 491           retries:
 492             pool.info.maxSize +
 493             Object.keys(workerChoiceStrategy.opts.weights).length,
 494           runTime: { median: false },
 495           waitTime: { median: false },
 496           elu: { median: false }
 497           // weights: expect.objectContaining({
 498           //   0: expect.any(Number),
 499           //   [pool.info.maxSize - 1]: expect.any(Number)
 500           // })
 501         })
 502       )
 503     }
 504     expect(
 505       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 506     ).toStrictEqual({
 507       runTime: {
 508         aggregate: true,
 509         average: true,
 510         median: false
 511       },
 512       waitTime: {
 513         aggregate: false,
 514         average: false,
 515         median: false
 516       },
 517       elu: {
 518         aggregate: true,
 519         average: true,
 520         median: false
 521       }
 522     })
 523     pool.setWorkerChoiceStrategyOptions({
 524       runTime: { median: true },
 525       elu: { median: true }
 526     })
 527     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 528       runTime: { median: true },
 529       elu: { median: true }
 530     })
 531     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 532       retries:
 533         pool.info.maxSize +
 534         Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
 535       runTime: { median: true },
 536       waitTime: { median: false },
 537       elu: { median: true },
 538       weights: expect.objectContaining({
 539         0: expect.any(Number),
 540         [pool.info.maxSize - 1]: expect.any(Number)
 541       })
 542     })
 543     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 544       .workerChoiceStrategies) {
 545       expect(workerChoiceStrategy.opts).toStrictEqual(
 546         expect.objectContaining({
 547           retries:
 548             pool.info.maxSize +
 549             Object.keys(workerChoiceStrategy.opts.weights).length,
 550           runTime: { median: true },
 551           waitTime: { median: false },
 552           elu: { median: true }
 553           // weights: expect.objectContaining({
 554           //   0: expect.any(Number),
 555           //   [pool.info.maxSize - 1]: expect.any(Number)
 556           // })
 557         })
 558       )
 559     }
 560     expect(
 561       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 562     ).toStrictEqual({
 563       runTime: {
 564         aggregate: true,
 565         average: false,
 566         median: true
 567       },
 568       waitTime: {
 569         aggregate: false,
 570         average: false,
 571         median: false
 572       },
 573       elu: {
 574         aggregate: true,
 575         average: false,
 576         median: true
 577       }
 578     })
 579     pool.setWorkerChoiceStrategyOptions({
 580       runTime: { median: false },
 581       elu: { median: false }
 582     })
 583     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 584       runTime: { median: false },
 585       elu: { median: false }
 586     })
 587     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
 588       retries:
 589         pool.info.maxSize +
 590         Object.keys(pool.workerChoiceStrategyContext.opts.weights).length,
 591       runTime: { median: false },
 592       waitTime: { median: false },
 593       elu: { median: false },
 594       weights: expect.objectContaining({
 595         0: expect.any(Number),
 596         [pool.info.maxSize - 1]: expect.any(Number)
 597       })
 598     })
 599     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 600       .workerChoiceStrategies) {
 601       expect(workerChoiceStrategy.opts).toStrictEqual({
 602         retries:
 603           pool.info.maxSize +
 604           Object.keys(workerChoiceStrategy.opts.weights).length,
 605         runTime: { median: false },
 606         waitTime: { median: false },
 607         elu: { median: false },
 608         weights: expect.objectContaining({
 609           0: expect.any(Number),
 610           [pool.info.maxSize - 1]: expect.any(Number)
 611         })
 612       })
 613     }
 614     expect(
 615       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 616     ).toStrictEqual({
 617       runTime: {
 618         aggregate: true,
 619         average: true,
 620         median: false
 621       },
 622       waitTime: {
 623         aggregate: false,
 624         average: false,
 625         median: false
 626       },
 627       elu: {
 628         aggregate: true,
 629         average: true,
 630         median: false
 631       }
 632     })
 633     expect(() =>
 634       pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
 635     ).toThrow(
 636       new TypeError(
 637         'Invalid worker choice strategy options: must be a plain object'
 638       )
 639     )
 640     expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow(
 641       new Error(
 642         'Invalid worker choice strategy options: must have a weight for each worker node'
 643       )
 644     )
 645     expect(() =>
 646       pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
 647     ).toThrow(
 648       new Error(
 649         "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
 650       )
 651     )
 652     await pool.destroy()
 653   })
 654
 655   it('Verify that pool tasks queue can be enabled/disabled', async () => {
 656     const pool = new FixedThreadPool(
 657       numberOfWorkers,
 658       './tests/worker-files/thread/testWorker.mjs'
 659     )
 660     expect(pool.opts.enableTasksQueue).toBe(false)
 661     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 662     pool.enableTasksQueue(true)
 663     expect(pool.opts.enableTasksQueue).toBe(true)
 664     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 665       concurrency: 1,
 666       size: Math.pow(numberOfWorkers, 2),
 667       taskStealing: true,
 668       tasksStealingOnBackPressure: true,
 669       tasksFinishedTimeout: 2000
 670     })
 671     pool.enableTasksQueue(true, { concurrency: 2 })
 672     expect(pool.opts.enableTasksQueue).toBe(true)
 673     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 674       concurrency: 2,
 675       size: Math.pow(numberOfWorkers, 2),
 676       taskStealing: true,
 677       tasksStealingOnBackPressure: true,
 678       tasksFinishedTimeout: 2000
 679     })
 680     pool.enableTasksQueue(false)
 681     expect(pool.opts.enableTasksQueue).toBe(false)
 682     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 683     await pool.destroy()
 684   })
 685
 686   it('Verify that pool tasks queue options can be set', async () => {
 687     const pool = new FixedThreadPool(
 688       numberOfWorkers,
 689       './tests/worker-files/thread/testWorker.mjs',
 690       { enableTasksQueue: true }
 691     )
 692     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 693       concurrency: 1,
 694       size: Math.pow(numberOfWorkers, 2),
 695       taskStealing: true,
 696       tasksStealingOnBackPressure: true,
 697       tasksFinishedTimeout: 2000
 698     })
 699     for (const workerNode of pool.workerNodes) {
 700       expect(workerNode.tasksQueueBackPressureSize).toBe(
 701         pool.opts.tasksQueueOptions.size
 702       )
 703     }
 704     pool.setTasksQueueOptions({
 705       concurrency: 2,
 706       size: 2,
 707       taskStealing: false,
 708       tasksStealingOnBackPressure: false,
 709       tasksFinishedTimeout: 3000
 710     })
 711     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 712       concurrency: 2,
 713       size: 2,
 714       taskStealing: false,
 715       tasksStealingOnBackPressure: false,
 716       tasksFinishedTimeout: 3000
 717     })
 718     for (const workerNode of pool.workerNodes) {
 719       expect(workerNode.tasksQueueBackPressureSize).toBe(
 720         pool.opts.tasksQueueOptions.size
 721       )
 722     }
 723     pool.setTasksQueueOptions({
 724       concurrency: 1,
 725       taskStealing: true,
 726       tasksStealingOnBackPressure: true
 727     })
 728     expect(pool.opts.tasksQueueOptions).toStrictEqual({
 729       concurrency: 1,
 730       size: Math.pow(numberOfWorkers, 2),
 731       taskStealing: true,
 732       tasksStealingOnBackPressure: true,
 733       tasksFinishedTimeout: 2000
 734     })
 735     for (const workerNode of pool.workerNodes) {
 736       expect(workerNode.tasksQueueBackPressureSize).toBe(
 737         pool.opts.tasksQueueOptions.size
 738       )
 739     }
 740     expect(() => pool.setTasksQueueOptions('invalidTasksQueueOptions')).toThrow(
 741       new TypeError('Invalid tasks queue options: must be a plain object')
 742     )
 743     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrow(
 744       new RangeError(
 745         'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
 746       )
 747     )
 748     expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrow(
 749       new RangeError(
 750         'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
 751       )
 752     )
 753     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrow(
 754       new TypeError('Invalid worker node tasks concurrency: must be an integer')
 755     )
 756     expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrow(
 757       new RangeError(
 758         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
 759       )
 760     )
 761     expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrow(
 762       new RangeError(
 763         'Invalid worker node tasks queue size: -1 is a negative integer or zero'
 764       )
 765     )
 766     expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrow(
 767       new TypeError('Invalid worker node tasks queue size: must be an integer')
 768     )
 769     await pool.destroy()
 770   })
 771
 772   it('Verify that pool info is set', async () => {
 773     let pool = new FixedThreadPool(
 774       numberOfWorkers,
 775       './tests/worker-files/thread/testWorker.mjs'
 776     )
 777     expect(pool.info).toStrictEqual({
 778       version,
 779       type: PoolTypes.fixed,
 780       worker: WorkerTypes.thread,
 781       started: true,
 782       ready: true,
 783       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 784       minSize: numberOfWorkers,
 785       maxSize: numberOfWorkers,
 786       workerNodes: numberOfWorkers,
 787       idleWorkerNodes: numberOfWorkers,
 788       busyWorkerNodes: 0,
 789       executedTasks: 0,
 790       executingTasks: 0,
 791       failedTasks: 0
 792     })
 793     await pool.destroy()
 794     pool = new DynamicClusterPool(
 795       Math.floor(numberOfWorkers / 2),
 796       numberOfWorkers,
 797       './tests/worker-files/cluster/testWorker.js'
 798     )
 799     expect(pool.info).toStrictEqual({
 800       version,
 801       type: PoolTypes.dynamic,
 802       worker: WorkerTypes.cluster,
 803       started: true,
 804       ready: true,
 805       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
 806       minSize: Math.floor(numberOfWorkers / 2),
 807       maxSize: numberOfWorkers,
 808       workerNodes: Math.floor(numberOfWorkers / 2),
 809       idleWorkerNodes: Math.floor(numberOfWorkers / 2),
 810       busyWorkerNodes: 0,
 811       executedTasks: 0,
 812       executingTasks: 0,
 813       failedTasks: 0
 814     })
 815     await pool.destroy()
 816   })
 817
 818   it('Verify that pool worker tasks usage are initialized', async () => {
 819     const pool = new FixedClusterPool(
 820       numberOfWorkers,
 821       './tests/worker-files/cluster/testWorker.js'
 822     )
 823     for (const workerNode of pool.workerNodes) {
 824       expect(workerNode).toBeInstanceOf(WorkerNode)
 825       expect(workerNode.usage).toStrictEqual({
 826         tasks: {
 827           executed: 0,
 828           executing: 0,
 829           queued: 0,
 830           maxQueued: 0,
 831           sequentiallyStolen: 0,
 832           stolen: 0,
 833           failed: 0
 834         },
 835         runTime: {
 836           history: new CircularArray()
 837         },
 838         waitTime: {
 839           history: new CircularArray()
 840         },
 841         elu: {
 842           idle: {
 843             history: new CircularArray()
 844           },
 845           active: {
 846             history: new CircularArray()
 847           }
 848         }
 849       })
 850     }
 851     await pool.destroy()
 852   })
 853
 854   it('Verify that pool worker tasks queue are initialized', async () => {
 855     let pool = new FixedClusterPool(
 856       numberOfWorkers,
 857       './tests/worker-files/cluster/testWorker.js'
 858     )
 859     for (const workerNode of pool.workerNodes) {
 860       expect(workerNode).toBeInstanceOf(WorkerNode)
 861       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 862       expect(workerNode.tasksQueue.size).toBe(0)
 863       expect(workerNode.tasksQueue.maxSize).toBe(0)
 864     }
 865     await pool.destroy()
 866     pool = new DynamicThreadPool(
 867       Math.floor(numberOfWorkers / 2),
 868       numberOfWorkers,
 869       './tests/worker-files/thread/testWorker.mjs'
 870     )
 871     for (const workerNode of pool.workerNodes) {
 872       expect(workerNode).toBeInstanceOf(WorkerNode)
 873       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
 874       expect(workerNode.tasksQueue.size).toBe(0)
 875       expect(workerNode.tasksQueue.maxSize).toBe(0)
 876     }
 877     await pool.destroy()
 878   })
 879
 880   it('Verify that pool worker info are initialized', async () => {
 881     let pool = new FixedClusterPool(
 882       numberOfWorkers,
 883       './tests/worker-files/cluster/testWorker.js'
 884     )
 885     for (const workerNode of pool.workerNodes) {
 886       expect(workerNode).toBeInstanceOf(WorkerNode)
 887       expect(workerNode.info).toStrictEqual({
 888         id: expect.any(Number),
 889         type: WorkerTypes.cluster,
 890         dynamic: false,
 891         ready: true
 892       })
 893     }
 894     await pool.destroy()
 895     pool = new DynamicThreadPool(
 896       Math.floor(numberOfWorkers / 2),
 897       numberOfWorkers,
 898       './tests/worker-files/thread/testWorker.mjs'
 899     )
 900     for (const workerNode of pool.workerNodes) {
 901       expect(workerNode).toBeInstanceOf(WorkerNode)
 902       expect(workerNode.info).toStrictEqual({
 903         id: expect.any(Number),
 904         type: WorkerTypes.thread,
 905         dynamic: false,
 906         ready: true
 907       })
 908     }
 909     await pool.destroy()
 910   })
 911
 912   it('Verify that pool statuses are checked at start or destroy', async () => {
 913     const pool = new FixedThreadPool(
 914       numberOfWorkers,
 915       './tests/worker-files/thread/testWorker.mjs'
 916     )
 917     expect(pool.info.started).toBe(true)
 918     expect(pool.info.ready).toBe(true)
 919     expect(() => pool.start()).toThrow(
 920       new Error('Cannot start an already started pool')
 921     )
 922     await pool.destroy()
 923     expect(pool.info.started).toBe(false)
 924     expect(pool.info.ready).toBe(false)
 925     await expect(pool.destroy()).rejects.toThrow(
 926       new Error('Cannot destroy an already destroyed pool')
 927     )
 928   })
 929
 930   it('Verify that pool can be started after initialization', async () => {
 931     const pool = new FixedClusterPool(
 932       numberOfWorkers,
 933       './tests/worker-files/cluster/testWorker.js',
 934       {
 935         startWorkers: false
 936       }
 937     )
 938     expect(pool.info.started).toBe(false)
 939     expect(pool.info.ready).toBe(false)
 940     expect(pool.readyEventEmitted).toBe(false)
 941     expect(pool.workerNodes).toStrictEqual([])
 942     await expect(pool.execute()).rejects.toThrow(
 943       new Error('Cannot execute a task on not started pool')
 944     )
 945     pool.start()
 946     expect(pool.info.started).toBe(true)
 947     expect(pool.info.ready).toBe(true)
 948     await waitPoolEvents(pool, PoolEvents.ready, 1)
 949     expect(pool.readyEventEmitted).toBe(true)
 950     expect(pool.workerNodes.length).toBe(numberOfWorkers)
 951     for (const workerNode of pool.workerNodes) {
 952       expect(workerNode).toBeInstanceOf(WorkerNode)
 953     }
 954     await pool.destroy()
 955   })
 956
 957   it('Verify that pool execute() arguments are checked', async () => {
 958     const pool = new FixedClusterPool(
 959       numberOfWorkers,
 960       './tests/worker-files/cluster/testWorker.js'
 961     )
 962     await expect(pool.execute(undefined, 0)).rejects.toThrow(
 963       new TypeError('name argument must be a string')
 964     )
 965     await expect(pool.execute(undefined, '')).rejects.toThrow(
 966       new TypeError('name argument must not be an empty string')
 967     )
 968     await expect(pool.execute(undefined, undefined, {})).rejects.toThrow(
 969       new TypeError('transferList argument must be an array')
 970     )
 971     await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
 972       "Task function 'unknown' not found"
 973     )
 974     await pool.destroy()
 975     await expect(pool.execute()).rejects.toThrow(
 976       new Error('Cannot execute a task on not started pool')
 977     )
 978   })
 979
 980   it('Verify that pool worker tasks usage are computed', async () => {
 981     const pool = new FixedClusterPool(
 982       numberOfWorkers,
 983       './tests/worker-files/cluster/testWorker.js'
 984     )
 985     const promises = new Set()
 986     const maxMultiplier = 2
 987     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 988       promises.add(pool.execute())
 989     }
 990     for (const workerNode of pool.workerNodes) {
 991       expect(workerNode.usage).toStrictEqual({
 992         tasks: {
 993           executed: 0,
 994           executing: maxMultiplier,
 995           queued: 0,
 996           maxQueued: 0,
 997           sequentiallyStolen: 0,
 998           stolen: 0,
 999           failed: 0
1000         },
1001         runTime: {
1002           history: expect.any(CircularArray)
1003         },
1004         waitTime: {
1005           history: expect.any(CircularArray)
1006         },
1007         elu: {
1008           idle: {
1009             history: expect.any(CircularArray)
1010           },
1011           active: {
1012             history: expect.any(CircularArray)
1013           }
1014         }
1015       })
1016     }
1017     await Promise.all(promises)
1018     for (const workerNode of pool.workerNodes) {
1019       expect(workerNode.usage).toStrictEqual({
1020         tasks: {
1021           executed: maxMultiplier,
1022           executing: 0,
1023           queued: 0,
1024           maxQueued: 0,
1025           sequentiallyStolen: 0,
1026           stolen: 0,
1027           failed: 0
1028         },
1029         runTime: {
1030           history: expect.any(CircularArray)
1031         },
1032         waitTime: {
1033           history: expect.any(CircularArray)
1034         },
1035         elu: {
1036           idle: {
1037             history: expect.any(CircularArray)
1038           },
1039           active: {
1040             history: expect.any(CircularArray)
1041           }
1042         }
1043       })
1044     }
1045     await pool.destroy()
1046   })
1047
1048   it('Verify that pool worker tasks usage are reset at worker choice strategy change', async () => {
1049     const pool = new DynamicThreadPool(
1050       Math.floor(numberOfWorkers / 2),
1051       numberOfWorkers,
1052       './tests/worker-files/thread/testWorker.mjs'
1053     )
1054     const promises = new Set()
1055     const maxMultiplier = 2
1056     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1057       promises.add(pool.execute())
1058     }
1059     await Promise.all(promises)
1060     for (const workerNode of pool.workerNodes) {
1061       expect(workerNode.usage).toStrictEqual({
1062         tasks: {
1063           executed: expect.any(Number),
1064           executing: 0,
1065           queued: 0,
1066           maxQueued: 0,
1067           sequentiallyStolen: 0,
1068           stolen: 0,
1069           failed: 0
1070         },
1071         runTime: {
1072           history: expect.any(CircularArray)
1073         },
1074         waitTime: {
1075           history: expect.any(CircularArray)
1076         },
1077         elu: {
1078           idle: {
1079             history: expect.any(CircularArray)
1080           },
1081           active: {
1082             history: expect.any(CircularArray)
1083           }
1084         }
1085       })
1086       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1087       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1088         numberOfWorkers * maxMultiplier
1089       )
1090       expect(workerNode.usage.runTime.history.length).toBe(0)
1091       expect(workerNode.usage.waitTime.history.length).toBe(0)
1092       expect(workerNode.usage.elu.idle.history.length).toBe(0)
1093       expect(workerNode.usage.elu.active.history.length).toBe(0)
1094     }
1095     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
1096     for (const workerNode of pool.workerNodes) {
1097       expect(workerNode.usage).toStrictEqual({
1098         tasks: {
1099           executed: 0,
1100           executing: 0,
1101           queued: 0,
1102           maxQueued: 0,
1103           sequentiallyStolen: 0,
1104           stolen: 0,
1105           failed: 0
1106         },
1107         runTime: {
1108           history: expect.any(CircularArray)
1109         },
1110         waitTime: {
1111           history: expect.any(CircularArray)
1112         },
1113         elu: {
1114           idle: {
1115             history: expect.any(CircularArray)
1116           },
1117           active: {
1118             history: expect.any(CircularArray)
1119           }
1120         }
1121       })
1122       expect(workerNode.usage.runTime.history.length).toBe(0)
1123       expect(workerNode.usage.waitTime.history.length).toBe(0)
1124       expect(workerNode.usage.elu.idle.history.length).toBe(0)
1125       expect(workerNode.usage.elu.active.history.length).toBe(0)
1126     }
1127     await pool.destroy()
1128   })
1129
1130   it("Verify that pool event emitter 'ready' event can register a callback", async () => {
1131     const pool = new DynamicClusterPool(
1132       Math.floor(numberOfWorkers / 2),
1133       numberOfWorkers,
1134       './tests/worker-files/cluster/testWorker.js'
1135     )
1136     expect(pool.emitter.eventNames()).toStrictEqual([])
1137     let poolInfo
1138     let poolReady = 0
1139     pool.emitter.on(PoolEvents.ready, info => {
1140       ++poolReady
1141       poolInfo = info
1142     })
1143     await waitPoolEvents(pool, PoolEvents.ready, 1)
1144     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.ready])
1145     expect(poolReady).toBe(1)
1146     expect(poolInfo).toStrictEqual({
1147       version,
1148       type: PoolTypes.dynamic,
1149       worker: WorkerTypes.cluster,
1150       started: true,
1151       ready: true,
1152       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1153       minSize: expect.any(Number),
1154       maxSize: expect.any(Number),
1155       workerNodes: expect.any(Number),
1156       idleWorkerNodes: expect.any(Number),
1157       busyWorkerNodes: expect.any(Number),
1158       executedTasks: expect.any(Number),
1159       executingTasks: expect.any(Number),
1160       failedTasks: expect.any(Number)
1161     })
1162     await pool.destroy()
1163   })
1164
1165   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
1166     const pool = new FixedThreadPool(
1167       numberOfWorkers,
1168       './tests/worker-files/thread/testWorker.mjs'
1169     )
1170     expect(pool.emitter.eventNames()).toStrictEqual([])
1171     const promises = new Set()
1172     let poolBusy = 0
1173     let poolInfo
1174     pool.emitter.on(PoolEvents.busy, info => {
1175       ++poolBusy
1176       poolInfo = info
1177     })
1178     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
1179     for (let i = 0; i < numberOfWorkers * 2; i++) {
1180       promises.add(pool.execute())
1181     }
1182     await Promise.all(promises)
1183     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
1184     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
1185     expect(poolBusy).toBe(numberOfWorkers + 1)
1186     expect(poolInfo).toStrictEqual({
1187       version,
1188       type: PoolTypes.fixed,
1189       worker: WorkerTypes.thread,
1190       started: true,
1191       ready: true,
1192       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1193       minSize: expect.any(Number),
1194       maxSize: expect.any(Number),
1195       workerNodes: expect.any(Number),
1196       idleWorkerNodes: expect.any(Number),
1197       busyWorkerNodes: expect.any(Number),
1198       executedTasks: expect.any(Number),
1199       executingTasks: expect.any(Number),
1200       failedTasks: expect.any(Number)
1201     })
1202     await pool.destroy()
1203   })
1204
1205   it("Verify that pool event emitter 'full' event can register a callback", async () => {
1206     const pool = new DynamicThreadPool(
1207       Math.floor(numberOfWorkers / 2),
1208       numberOfWorkers,
1209       './tests/worker-files/thread/testWorker.mjs'
1210     )
1211     expect(pool.emitter.eventNames()).toStrictEqual([])
1212     const promises = new Set()
1213     let poolFull = 0
1214     let poolInfo
1215     pool.emitter.on(PoolEvents.full, info => {
1216       ++poolFull
1217       poolInfo = info
1218     })
1219     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.full])
1220     for (let i = 0; i < numberOfWorkers * 2; i++) {
1221       promises.add(pool.execute())
1222     }
1223     await Promise.all(promises)
1224     expect(poolFull).toBe(1)
1225     expect(poolInfo).toStrictEqual({
1226       version,
1227       type: PoolTypes.dynamic,
1228       worker: WorkerTypes.thread,
1229       started: true,
1230       ready: true,
1231       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1232       minSize: expect.any(Number),
1233       maxSize: expect.any(Number),
1234       workerNodes: expect.any(Number),
1235       idleWorkerNodes: expect.any(Number),
1236       busyWorkerNodes: expect.any(Number),
1237       executedTasks: expect.any(Number),
1238       executingTasks: expect.any(Number),
1239       failedTasks: expect.any(Number)
1240     })
1241     await pool.destroy()
1242   })
1243
1244   it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
1245     const pool = new FixedThreadPool(
1246       numberOfWorkers,
1247       './tests/worker-files/thread/testWorker.mjs',
1248       {
1249         enableTasksQueue: true
1250       }
1251     )
1252     stub(pool, 'hasBackPressure').returns(true)
1253     expect(pool.emitter.eventNames()).toStrictEqual([])
1254     const promises = new Set()
1255     let poolBackPressure = 0
1256     let poolInfo
1257     pool.emitter.on(PoolEvents.backPressure, info => {
1258       ++poolBackPressure
1259       poolInfo = info
1260     })
1261     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.backPressure])
1262     for (let i = 0; i < numberOfWorkers + 1; i++) {
1263       promises.add(pool.execute())
1264     }
1265     await Promise.all(promises)
1266     expect(poolBackPressure).toBe(1)
1267     expect(poolInfo).toStrictEqual({
1268       version,
1269       type: PoolTypes.fixed,
1270       worker: WorkerTypes.thread,
1271       started: true,
1272       ready: true,
1273       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
1274       minSize: expect.any(Number),
1275       maxSize: expect.any(Number),
1276       workerNodes: expect.any(Number),
1277       idleWorkerNodes: expect.any(Number),
1278       busyWorkerNodes: expect.any(Number),
1279       executedTasks: expect.any(Number),
1280       executingTasks: expect.any(Number),
1281       maxQueuedTasks: expect.any(Number),
1282       queuedTasks: expect.any(Number),
1283       backPressure: true,
1284       stolenTasks: expect.any(Number),
1285       failedTasks: expect.any(Number)
1286     })
1287     expect(pool.hasBackPressure.callCount).toBe(5)
1288     await pool.destroy()
1289   })
1290
1291   it('Verify that destroy() waits for queued tasks to finish', async () => {
1292     const tasksFinishedTimeout = 2500
1293     const pool = new FixedThreadPool(
1294       numberOfWorkers,
1295       './tests/worker-files/thread/asyncWorker.mjs',
1296       {
1297         enableTasksQueue: true,
1298         tasksQueueOptions: { tasksFinishedTimeout }
1299       }
1300     )
1301     const maxMultiplier = 4
1302     let tasksFinished = 0
1303     for (const workerNode of pool.workerNodes) {
1304       workerNode.on('taskFinished', () => {
1305         ++tasksFinished
1306       })
1307     }
1308     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1309       pool.execute()
1310     }
1311     expect(pool.info.queuedTasks).toBeGreaterThan(0)
1312     const startTime = performance.now()
1313     await pool.destroy()
1314     const elapsedTime = performance.now() - startTime
1315     expect(tasksFinished).toBe(numberOfWorkers * maxMultiplier)
1316     expect(elapsedTime).toBeGreaterThanOrEqual(2000)
1317     expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 100)
1318   })
1319
1320   it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
1321     const tasksFinishedTimeout = 1000
1322     const pool = new FixedThreadPool(
1323       numberOfWorkers,
1324       './tests/worker-files/thread/asyncWorker.mjs',
1325       {
1326         enableTasksQueue: true,
1327         tasksQueueOptions: { tasksFinishedTimeout }
1328       }
1329     )
1330     const maxMultiplier = 4
1331     let tasksFinished = 0
1332     for (const workerNode of pool.workerNodes) {
1333       workerNode.on('taskFinished', () => {
1334         ++tasksFinished
1335       })
1336     }
1337     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
1338       pool.execute()
1339     }
1340     expect(pool.info.queuedTasks).toBeGreaterThan(0)
1341     const startTime = performance.now()
1342     await pool.destroy()
1343     const elapsedTime = performance.now() - startTime
1344     expect(tasksFinished).toBe(0)
1345     expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 600)
1346   })
1347
1348   it('Verify that pool asynchronous resource track tasks execution', async () => {
1349     let taskAsyncId
1350     let initCalls = 0
1351     let beforeCalls = 0
1352     let afterCalls = 0
1353     let resolveCalls = 0
1354     const hook = createHook({
1355       init (asyncId, type) {
1356         if (type === 'poolifier:task') {
1357           initCalls++
1358           taskAsyncId = asyncId
1359         }
1360       },
1361       before (asyncId) {
1362         if (asyncId === taskAsyncId) beforeCalls++
1363       },
1364       after (asyncId) {
1365         if (asyncId === taskAsyncId) afterCalls++
1366       },
1367       promiseResolve () {
1368         if (executionAsyncId() === taskAsyncId) resolveCalls++
1369       }
1370     })
1371     const pool = new FixedThreadPool(
1372       numberOfWorkers,
1373       './tests/worker-files/thread/testWorker.mjs'
1374     )
1375     hook.enable()
1376     await pool.execute()
1377     hook.disable()
1378     expect(initCalls).toBe(1)
1379     expect(beforeCalls).toBe(1)
1380     expect(afterCalls).toBe(1)
1381     expect(resolveCalls).toBe(1)
1382     await pool.destroy()
1383   })
1384
1385   it('Verify that hasTaskFunction() is working', async () => {
1386     const dynamicThreadPool = new DynamicThreadPool(
1387       Math.floor(numberOfWorkers / 2),
1388       numberOfWorkers,
1389       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1390     )
1391     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1392     expect(dynamicThreadPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1393     expect(dynamicThreadPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1394       true
1395     )
1396     expect(dynamicThreadPool.hasTaskFunction('factorial')).toBe(true)
1397     expect(dynamicThreadPool.hasTaskFunction('fibonacci')).toBe(true)
1398     expect(dynamicThreadPool.hasTaskFunction('unknown')).toBe(false)
1399     await dynamicThreadPool.destroy()
1400     const fixedClusterPool = new FixedClusterPool(
1401       numberOfWorkers,
1402       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1403     )
1404     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1405     expect(fixedClusterPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
1406     expect(fixedClusterPool.hasTaskFunction('jsonIntegerSerialization')).toBe(
1407       true
1408     )
1409     expect(fixedClusterPool.hasTaskFunction('factorial')).toBe(true)
1410     expect(fixedClusterPool.hasTaskFunction('fibonacci')).toBe(true)
1411     expect(fixedClusterPool.hasTaskFunction('unknown')).toBe(false)
1412     await fixedClusterPool.destroy()
1413   })
1414
1415   it('Verify that addTaskFunction() is working', async () => {
1416     const dynamicThreadPool = new DynamicThreadPool(
1417       Math.floor(numberOfWorkers / 2),
1418       numberOfWorkers,
1419       './tests/worker-files/thread/testWorker.mjs'
1420     )
1421     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1422     await expect(
1423       dynamicThreadPool.addTaskFunction(0, () => {})
1424     ).rejects.toThrow(new TypeError('name argument must be a string'))
1425     await expect(
1426       dynamicThreadPool.addTaskFunction('', () => {})
1427     ).rejects.toThrow(
1428       new TypeError('name argument must not be an empty string')
1429     )
1430     await expect(dynamicThreadPool.addTaskFunction('test', 0)).rejects.toThrow(
1431       new TypeError('fn argument must be a function')
1432     )
1433     await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow(
1434       new TypeError('fn argument must be a function')
1435     )
1436     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1437       DEFAULT_TASK_NAME,
1438       'test'
1439     ])
1440     const echoTaskFunction = data => {
1441       return data
1442     }
1443     await expect(
1444       dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1445     ).resolves.toBe(true)
1446     expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1447     expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1448       echoTaskFunction
1449     )
1450     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1451       DEFAULT_TASK_NAME,
1452       'test',
1453       'echo'
1454     ])
1455     const taskFunctionData = { test: 'test' }
1456     const echoResult = await dynamicThreadPool.execute(taskFunctionData, 'echo')
1457     expect(echoResult).toStrictEqual(taskFunctionData)
1458     for (const workerNode of dynamicThreadPool.workerNodes) {
1459       expect(workerNode.getTaskFunctionWorkerUsage('echo')).toStrictEqual({
1460         tasks: {
1461           executed: expect.any(Number),
1462           executing: 0,
1463           queued: 0,
1464           sequentiallyStolen: 0,
1465           stolen: 0,
1466           failed: 0
1467         },
1468         runTime: {
1469           history: new CircularArray()
1470         },
1471         waitTime: {
1472           history: new CircularArray()
1473         },
1474         elu: {
1475           idle: {
1476             history: new CircularArray()
1477           },
1478           active: {
1479             history: new CircularArray()
1480           }
1481         }
1482       })
1483     }
1484     await dynamicThreadPool.destroy()
1485   })
1486
1487   it('Verify that removeTaskFunction() is working', async () => {
1488     const dynamicThreadPool = new DynamicThreadPool(
1489       Math.floor(numberOfWorkers / 2),
1490       numberOfWorkers,
1491       './tests/worker-files/thread/testWorker.mjs'
1492     )
1493     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1494     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1495       DEFAULT_TASK_NAME,
1496       'test'
1497     ])
1498     await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow(
1499       new Error('Cannot remove a task function not handled on the pool side')
1500     )
1501     const echoTaskFunction = data => {
1502       return data
1503     }
1504     await dynamicThreadPool.addTaskFunction('echo', echoTaskFunction)
1505     expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1506     expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual(
1507       echoTaskFunction
1508     )
1509     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1510       DEFAULT_TASK_NAME,
1511       'test',
1512       'echo'
1513     ])
1514     await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe(
1515       true
1516     )
1517     expect(dynamicThreadPool.taskFunctions.size).toBe(0)
1518     expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined()
1519     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1520       DEFAULT_TASK_NAME,
1521       'test'
1522     ])
1523     await dynamicThreadPool.destroy()
1524   })
1525
1526   it('Verify that listTaskFunctionNames() is working', async () => {
1527     const dynamicThreadPool = new DynamicThreadPool(
1528       Math.floor(numberOfWorkers / 2),
1529       numberOfWorkers,
1530       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1531     )
1532     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1533     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1534       DEFAULT_TASK_NAME,
1535       'jsonIntegerSerialization',
1536       'factorial',
1537       'fibonacci'
1538     ])
1539     await dynamicThreadPool.destroy()
1540     const fixedClusterPool = new FixedClusterPool(
1541       numberOfWorkers,
1542       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1543     )
1544     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1545     expect(fixedClusterPool.listTaskFunctionNames()).toStrictEqual([
1546       DEFAULT_TASK_NAME,
1547       'jsonIntegerSerialization',
1548       'factorial',
1549       'fibonacci'
1550     ])
1551     await fixedClusterPool.destroy()
1552   })
1553
1554   it('Verify that setDefaultTaskFunction() is working', async () => {
1555     const dynamicThreadPool = new DynamicThreadPool(
1556       Math.floor(numberOfWorkers / 2),
1557       numberOfWorkers,
1558       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1559     )
1560     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1561     const workerId = dynamicThreadPool.workerNodes[0].info.id
1562     await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow(
1563       new Error(
1564         `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'`
1565       )
1566     )
1567     await expect(
1568       dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME)
1569     ).rejects.toThrow(
1570       new Error(
1571         `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function reserved name as the default task function'`
1572       )
1573     )
1574     await expect(
1575       dynamicThreadPool.setDefaultTaskFunction('unknown')
1576     ).rejects.toThrow(
1577       new Error(
1578         `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function to a non-existing task function'`
1579       )
1580     )
1581     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1582       DEFAULT_TASK_NAME,
1583       'jsonIntegerSerialization',
1584       'factorial',
1585       'fibonacci'
1586     ])
1587     await expect(
1588       dynamicThreadPool.setDefaultTaskFunction('factorial')
1589     ).resolves.toBe(true)
1590     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1591       DEFAULT_TASK_NAME,
1592       'factorial',
1593       'jsonIntegerSerialization',
1594       'fibonacci'
1595     ])
1596     await expect(
1597       dynamicThreadPool.setDefaultTaskFunction('fibonacci')
1598     ).resolves.toBe(true)
1599     expect(dynamicThreadPool.listTaskFunctionNames()).toStrictEqual([
1600       DEFAULT_TASK_NAME,
1601       'fibonacci',
1602       'jsonIntegerSerialization',
1603       'factorial'
1604     ])
1605     await dynamicThreadPool.destroy()
1606   })
1607
1608   it('Verify that multiple task functions worker is working', async () => {
1609     const pool = new DynamicClusterPool(
1610       Math.floor(numberOfWorkers / 2),
1611       numberOfWorkers,
1612       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
1613     )
1614     const data = { n: 10 }
1615     const result0 = await pool.execute(data)
1616     expect(result0).toStrictEqual({ ok: 1 })
1617     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1618     expect(result1).toStrictEqual({ ok: 1 })
1619     const result2 = await pool.execute(data, 'factorial')
1620     expect(result2).toBe(3628800)
1621     const result3 = await pool.execute(data, 'fibonacci')
1622     expect(result3).toBe(55)
1623     expect(pool.info.executingTasks).toBe(0)
1624     expect(pool.info.executedTasks).toBe(4)
1625     for (const workerNode of pool.workerNodes) {
1626       expect(workerNode.info.taskFunctionNames).toStrictEqual([
1627         DEFAULT_TASK_NAME,
1628         'jsonIntegerSerialization',
1629         'factorial',
1630         'fibonacci'
1631       ])
1632       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1633       for (const name of pool.listTaskFunctionNames()) {
1634         expect(workerNode.getTaskFunctionWorkerUsage(name)).toStrictEqual({
1635           tasks: {
1636             executed: expect.any(Number),
1637             executing: 0,
1638             failed: 0,
1639             queued: 0,
1640             sequentiallyStolen: 0,
1641             stolen: 0
1642           },
1643           runTime: {
1644             history: expect.any(CircularArray)
1645           },
1646           waitTime: {
1647             history: expect.any(CircularArray)
1648           },
1649           elu: {
1650             idle: {
1651               history: expect.any(CircularArray)
1652             },
1653             active: {
1654               history: expect.any(CircularArray)
1655             }
1656           }
1657         })
1658         expect(
1659           workerNode.getTaskFunctionWorkerUsage(name).tasks.executed
1660         ).toBeGreaterThan(0)
1661       }
1662       expect(
1663         workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1664       ).toStrictEqual(
1665         workerNode.getTaskFunctionWorkerUsage(
1666           workerNode.info.taskFunctionNames[1]
1667         )
1668       )
1669     }
1670     await pool.destroy()
1671   })
1672
1673   it('Verify sendKillMessageToWorker()', async () => {
1674     const pool = new DynamicClusterPool(
1675       Math.floor(numberOfWorkers / 2),
1676       numberOfWorkers,
1677       './tests/worker-files/cluster/testWorker.js'
1678     )
1679     const workerNodeKey = 0
1680     await expect(
1681       pool.sendKillMessageToWorker(workerNodeKey)
1682     ).resolves.toBeUndefined()
1683     await expect(
1684       pool.sendKillMessageToWorker(numberOfWorkers)
1685     ).rejects.toStrictEqual(
1686       new Error(`Invalid worker node key '${numberOfWorkers}'`)
1687     )
1688     await pool.destroy()
1689   })
1690
1691   it('Verify sendTaskFunctionOperationToWorker()', async () => {
1692     const pool = new DynamicClusterPool(
1693       Math.floor(numberOfWorkers / 2),
1694       numberOfWorkers,
1695       './tests/worker-files/cluster/testWorker.js'
1696     )
1697     const workerNodeKey = 0
1698     await expect(
1699       pool.sendTaskFunctionOperationToWorker(workerNodeKey, {
1700         taskFunctionOperation: 'add',
1701         taskFunctionName: 'empty',
1702         taskFunction: (() => {}).toString()
1703       })
1704     ).resolves.toBe(true)
1705     expect(
1706       pool.workerNodes[workerNodeKey].info.taskFunctionNames
1707     ).toStrictEqual([DEFAULT_TASK_NAME, 'test', 'empty'])
1708     await pool.destroy()
1709   })
1710
1711   it('Verify sendTaskFunctionOperationToWorkers()', async () => {
1712     const pool = new DynamicClusterPool(
1713       Math.floor(numberOfWorkers / 2),
1714       numberOfWorkers,
1715       './tests/worker-files/cluster/testWorker.js'
1716     )
1717     await expect(
1718       pool.sendTaskFunctionOperationToWorkers({
1719         taskFunctionOperation: 'add',
1720         taskFunctionName: 'empty',
1721         taskFunction: (() => {}).toString()
1722       })
1723     ).resolves.toBe(true)
1724     for (const workerNode of pool.workerNodes) {
1725       expect(workerNode.info.taskFunctionNames).toStrictEqual([
1726         DEFAULT_TASK_NAME,
1727         'test',
1728         'empty'
1729       ])
1730     }
1731     await pool.destroy()
1732   })
1733 })