tests/pools/abstract-pool.test.mjs

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