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 { CircularBuffer } from '../../lib/circular-buffer.cjs'
  12 import {
  13   DynamicClusterPool,
  14   DynamicThreadPool,
  15   FixedClusterPool,
  16   FixedThreadPool,
  17   PoolEvents,
  18   PoolTypes,
  19   WorkerChoiceStrategies,
  20   WorkerTypes
  21 } from '../../lib/index.cjs'
  22 import { WorkerNode } from '../../lib/pools/worker-node.cjs'
  23 import { PriorityQueue } from '../../lib/priority-queue.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.workerChoiceStrategiesContext
 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: false,
 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.workerChoiceStrategiesContext
 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.workerChoiceStrategiesContext
 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.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
 464     ).toStrictEqual({
 465       runTime: {
 466         aggregate: true,
 467         average: true,
 468         median: false
 469       },
 470       waitTime: {
 471         aggregate: true,
 472         average: true,
 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.workerChoiceStrategiesContext
 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.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
 503     ).toStrictEqual({
 504       runTime: {
 505         aggregate: true,
 506         average: false,
 507         median: true
 508       },
 509       waitTime: {
 510         aggregate: true,
 511         average: true,
 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.workerChoiceStrategiesContext
 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.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
 542     ).toStrictEqual({
 543       runTime: {
 544         aggregate: true,
 545         average: true,
 546         median: false
 547       },
 548       waitTime: {
 549         aggregate: true,
 550         average: true,
 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: false,
 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: false,
 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: false,
 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       defaultStrategy: 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       defaultStrategy: 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: expect.any(CircularBuffer)
 765         },
 766         waitTime: {
 767           history: expect.any(CircularBuffer)
 768         },
 769         elu: {
 770           idle: {
 771             history: expect.any(CircularBuffer)
 772           },
 773           active: {
 774             history: expect.any(CircularBuffer)
 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(PriorityQueue)
 790       expect(workerNode.tasksQueue.size).toBe(0)
 791       expect(workerNode.tasksQueue.maxSize).toBe(0)
 792       expect(workerNode.tasksQueue.bucketSize).toBe(numberOfWorkers * 2)
 793     }
 794     await pool.destroy()
 795     pool = new DynamicThreadPool(
 796       Math.floor(numberOfWorkers / 2),
 797       numberOfWorkers,
 798       './tests/worker-files/thread/testWorker.mjs'
 799     )
 800     for (const workerNode of pool.workerNodes) {
 801       expect(workerNode).toBeInstanceOf(WorkerNode)
 802       expect(workerNode.tasksQueue).toBeInstanceOf(PriorityQueue)
 803       expect(workerNode.tasksQueue.size).toBe(0)
 804       expect(workerNode.tasksQueue.maxSize).toBe(0)
 805       expect(workerNode.tasksQueue.bucketSize).toBe(numberOfWorkers * 2)
 806     }
 807     await pool.destroy()
 808   })
 809
 810   it('Verify that pool worker info are initialized', async () => {
 811     let pool = new FixedClusterPool(
 812       numberOfWorkers,
 813       './tests/worker-files/cluster/testWorker.cjs'
 814     )
 815     for (const workerNode of pool.workerNodes) {
 816       expect(workerNode).toBeInstanceOf(WorkerNode)
 817       expect(workerNode.info).toStrictEqual({
 818         id: expect.any(Number),
 819         type: WorkerTypes.cluster,
 820         dynamic: false,
 821         ready: true,
 822         stealing: false,
 823         backPressure: false
 824       })
 825     }
 826     await pool.destroy()
 827     pool = new DynamicThreadPool(
 828       Math.floor(numberOfWorkers / 2),
 829       numberOfWorkers,
 830       './tests/worker-files/thread/testWorker.mjs'
 831     )
 832     for (const workerNode of pool.workerNodes) {
 833       expect(workerNode).toBeInstanceOf(WorkerNode)
 834       expect(workerNode.info).toStrictEqual({
 835         id: expect.any(Number),
 836         type: WorkerTypes.thread,
 837         dynamic: false,
 838         ready: true,
 839         stealing: false,
 840         backPressure: false
 841       })
 842     }
 843     await pool.destroy()
 844   })
 845
 846   it('Verify that pool statuses are checked at start or destroy', async () => {
 847     const pool = new FixedThreadPool(
 848       numberOfWorkers,
 849       './tests/worker-files/thread/testWorker.mjs'
 850     )
 851     expect(pool.info.started).toBe(true)
 852     expect(pool.info.ready).toBe(true)
 853     expect(() => pool.start()).toThrow(
 854       new Error('Cannot start an already started pool')
 855     )
 856     await pool.destroy()
 857     expect(pool.info.started).toBe(false)
 858     expect(pool.info.ready).toBe(false)
 859     await expect(pool.destroy()).rejects.toThrow(
 860       new Error('Cannot destroy an already destroyed pool')
 861     )
 862   })
 863
 864   it('Verify that pool can be started after initialization', async () => {
 865     const pool = new FixedClusterPool(
 866       numberOfWorkers,
 867       './tests/worker-files/cluster/testWorker.cjs',
 868       {
 869         startWorkers: false
 870       }
 871     )
 872     expect(pool.info.started).toBe(false)
 873     expect(pool.info.ready).toBe(false)
 874     expect(pool.workerNodes).toStrictEqual([])
 875     expect(pool.readyEventEmitted).toBe(false)
 876     await expect(pool.execute()).rejects.toThrow(
 877       new Error('Cannot execute a task on not started pool')
 878     )
 879     pool.start()
 880     expect(pool.info.started).toBe(true)
 881     expect(pool.info.ready).toBe(true)
 882     await waitPoolEvents(pool, PoolEvents.ready, 1)
 883     expect(pool.readyEventEmitted).toBe(true)
 884     expect(pool.workerNodes.length).toBe(numberOfWorkers)
 885     for (const workerNode of pool.workerNodes) {
 886       expect(workerNode).toBeInstanceOf(WorkerNode)
 887     }
 888     await pool.destroy()
 889   })
 890
 891   it('Verify that pool execute() arguments are checked', async () => {
 892     const pool = new FixedClusterPool(
 893       numberOfWorkers,
 894       './tests/worker-files/cluster/testWorker.cjs'
 895     )
 896     await expect(pool.execute(undefined, 0)).rejects.toThrow(
 897       new TypeError('name argument must be a string')
 898     )
 899     await expect(pool.execute(undefined, '')).rejects.toThrow(
 900       new TypeError('name argument must not be an empty string')
 901     )
 902     await expect(pool.execute(undefined, undefined, {})).rejects.toThrow(
 903       new TypeError('transferList argument must be an array')
 904     )
 905     await expect(pool.execute(undefined, 'unknown')).rejects.toBe(
 906       "Task function 'unknown' not found"
 907     )
 908     await pool.destroy()
 909     await expect(pool.execute()).rejects.toThrow(
 910       new Error('Cannot execute a task on not started pool')
 911     )
 912   })
 913
 914   it('Verify that pool worker tasks usage are computed', async () => {
 915     const pool = new FixedClusterPool(
 916       numberOfWorkers,
 917       './tests/worker-files/cluster/testWorker.cjs'
 918     )
 919     const promises = new Set()
 920     const maxMultiplier = 2
 921     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 922       promises.add(pool.execute())
 923     }
 924     for (const workerNode of pool.workerNodes) {
 925       expect(workerNode.usage).toStrictEqual({
 926         tasks: {
 927           executed: 0,
 928           executing: maxMultiplier,
 929           queued: 0,
 930           maxQueued: 0,
 931           sequentiallyStolen: 0,
 932           stolen: 0,
 933           failed: 0
 934         },
 935         runTime: {
 936           history: expect.any(CircularBuffer)
 937         },
 938         waitTime: {
 939           history: expect.any(CircularBuffer)
 940         },
 941         elu: {
 942           idle: {
 943             history: expect.any(CircularBuffer)
 944           },
 945           active: {
 946             history: expect.any(CircularBuffer)
 947           }
 948         }
 949       })
 950     }
 951     await Promise.all(promises)
 952     for (const workerNode of pool.workerNodes) {
 953       expect(workerNode.usage).toStrictEqual({
 954         tasks: {
 955           executed: maxMultiplier,
 956           executing: 0,
 957           queued: 0,
 958           maxQueued: 0,
 959           sequentiallyStolen: 0,
 960           stolen: 0,
 961           failed: 0
 962         },
 963         runTime: {
 964           history: expect.any(CircularBuffer)
 965         },
 966         waitTime: {
 967           history: expect.any(CircularBuffer)
 968         },
 969         elu: {
 970           idle: {
 971             history: expect.any(CircularBuffer)
 972           },
 973           active: {
 974             history: expect.any(CircularBuffer)
 975           }
 976         }
 977       })
 978     }
 979     await pool.destroy()
 980   })
 981
 982   it("Verify that pool worker tasks usage aren't reset at worker choice strategy change", async () => {
 983     const pool = new DynamicThreadPool(
 984       Math.floor(numberOfWorkers / 2),
 985       numberOfWorkers,
 986       './tests/worker-files/thread/testWorker.mjs'
 987     )
 988     const promises = new Set()
 989     const maxMultiplier = 2
 990     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 991       promises.add(pool.execute())
 992     }
 993     await Promise.all(promises)
 994     for (const workerNode of pool.workerNodes) {
 995       expect(workerNode.usage).toStrictEqual({
 996         tasks: {
 997           executed: expect.any(Number),
 998           executing: 0,
 999           queued: 0,
1000           maxQueued: 0,
1001           sequentiallyStolen: 0,
1002           stolen: 0,
1003           failed: 0
1004         },
1005         runTime: {
1006           history: expect.any(CircularBuffer)
1007         },
1008         waitTime: {
1009           history: expect.any(CircularBuffer)
1010         },
1011         elu: {
1012           idle: {
1013             history: expect.any(CircularBuffer)
1014           },
1015           active: {
1016             history: expect.any(CircularBuffer)
1017           }
1018         }
1019       })
1020       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1021       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1022         numberOfWorkers * maxMultiplier
1023       )
1024     }
1025     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
1026     for (const workerNode of pool.workerNodes) {
1027       expect(workerNode.usage).toStrictEqual({
1028         tasks: {
1029           executed: expect.any(Number),
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(CircularBuffer)
1039         },
1040         waitTime: {
1041           history: expect.any(CircularBuffer)
1042         },
1043         elu: {
1044           idle: {
1045             history: expect.any(CircularBuffer)
1046           },
1047           active: {
1048             history: expect.any(CircularBuffer)
1049           }
1050         }
1051       })
1052       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1053       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
1054         numberOfWorkers * maxMultiplier
1055       )
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       defaultStrategy: 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       defaultStrategy: 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       defaultStrategy: 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       defaultStrategy: 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('taskFunction property must be a function')
1367     )
1368     await expect(dynamicThreadPool.addTaskFunction('test', '')).rejects.toThrow(
1369       new TypeError('taskFunction property must be a function')
1370     )
1371     await expect(
1372       dynamicThreadPool.addTaskFunction('test', { taskFunction: 0 })
1373     ).rejects.toThrow(new TypeError('taskFunction property must be a function'))
1374     await expect(
1375       dynamicThreadPool.addTaskFunction('test', { taskFunction: '' })
1376     ).rejects.toThrow(new TypeError('taskFunction property must be a function'))
1377     await expect(
1378       dynamicThreadPool.addTaskFunction('test', {
1379         taskFunction: () => {},
1380         priority: -21
1381       })
1382     ).rejects.toThrow(
1383       new RangeError("Property 'priority' must be between -20 and 19")
1384     )
1385     await expect(
1386       dynamicThreadPool.addTaskFunction('test', {
1387         taskFunction: () => {},
1388         priority: 20
1389       })
1390     ).rejects.toThrow(
1391       new RangeError("Property 'priority' must be between -20 and 19")
1392     )
1393     await expect(
1394       dynamicThreadPool.addTaskFunction('test', {
1395         taskFunction: () => {},
1396         strategy: 'invalidStrategy'
1397       })
1398     ).rejects.toThrow(
1399       new Error("Invalid worker choice strategy 'invalidStrategy'")
1400     )
1401     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1402       { name: DEFAULT_TASK_NAME },
1403       { name: 'test' }
1404     ])
1405     expect([
1406       ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys()
1407     ]).toStrictEqual([WorkerChoiceStrategies.ROUND_ROBIN])
1408     const echoTaskFunction = data => {
1409       return data
1410     }
1411     await expect(
1412       dynamicThreadPool.addTaskFunction('echo', {
1413         taskFunction: echoTaskFunction,
1414         strategy: WorkerChoiceStrategies.LEAST_ELU
1415       })
1416     ).resolves.toBe(true)
1417     expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1418     expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual({
1419       taskFunction: echoTaskFunction,
1420       strategy: WorkerChoiceStrategies.LEAST_ELU
1421     })
1422     expect([
1423       ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys()
1424     ]).toStrictEqual([
1425       WorkerChoiceStrategies.ROUND_ROBIN,
1426       WorkerChoiceStrategies.LEAST_ELU
1427     ])
1428     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1429       { name: DEFAULT_TASK_NAME },
1430       { name: 'test' },
1431       { name: 'echo', strategy: WorkerChoiceStrategies.LEAST_ELU }
1432     ])
1433     const taskFunctionData = { test: 'test' }
1434     const echoResult = await dynamicThreadPool.execute(taskFunctionData, 'echo')
1435     expect(echoResult).toStrictEqual(taskFunctionData)
1436     for (const workerNode of dynamicThreadPool.workerNodes) {
1437       expect(workerNode.getTaskFunctionWorkerUsage('echo')).toStrictEqual({
1438         tasks: {
1439           executed: expect.any(Number),
1440           executing: 0,
1441           queued: 0,
1442           sequentiallyStolen: 0,
1443           stolen: 0,
1444           failed: 0
1445         },
1446         runTime: {
1447           history: expect.any(CircularBuffer)
1448         },
1449         waitTime: {
1450           history: expect.any(CircularBuffer)
1451         },
1452         elu: expect.objectContaining({
1453           idle: expect.objectContaining({
1454             history: expect.any(CircularBuffer)
1455           }),
1456           active: expect.objectContaining({
1457             history: expect.any(CircularBuffer)
1458           })
1459         })
1460       })
1461       expect(
1462         workerNode.getTaskFunctionWorkerUsage('echo').tasks.executed
1463       ).toBeGreaterThan(0)
1464       if (
1465         workerNode.getTaskFunctionWorkerUsage('echo').elu.active.aggregate ==
1466         null
1467       ) {
1468         expect(
1469           workerNode.getTaskFunctionWorkerUsage('echo').elu.active.aggregate
1470         ).toBeUndefined()
1471       } else {
1472         expect(
1473           workerNode.getTaskFunctionWorkerUsage('echo').elu.active.aggregate
1474         ).toBeGreaterThan(0)
1475       }
1476       if (
1477         workerNode.getTaskFunctionWorkerUsage('echo').elu.idle.aggregate == null
1478       ) {
1479         expect(
1480           workerNode.getTaskFunctionWorkerUsage('echo').elu.idle.aggregate
1481         ).toBeUndefined()
1482       } else {
1483         expect(
1484           workerNode.getTaskFunctionWorkerUsage('echo').elu.idle.aggregate
1485         ).toBeGreaterThanOrEqual(0)
1486       }
1487       if (
1488         workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization == null
1489       ) {
1490         expect(
1491           workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization
1492         ).toBeUndefined()
1493       } else {
1494         expect(
1495           workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization
1496         ).toBeGreaterThanOrEqual(0)
1497         expect(
1498           workerNode.getTaskFunctionWorkerUsage('echo').elu.utilization
1499         ).toBeLessThanOrEqual(1)
1500       }
1501     }
1502     await dynamicThreadPool.destroy()
1503   })
1504
1505   it('Verify that removeTaskFunction() is working', async () => {
1506     const dynamicThreadPool = new DynamicThreadPool(
1507       Math.floor(numberOfWorkers / 2),
1508       numberOfWorkers,
1509       './tests/worker-files/thread/testWorker.mjs'
1510     )
1511     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1512     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1513       { name: DEFAULT_TASK_NAME },
1514       { name: 'test' }
1515     ])
1516     await expect(dynamicThreadPool.removeTaskFunction('test')).rejects.toThrow(
1517       new Error('Cannot remove a task function not handled on the pool side')
1518     )
1519     const echoTaskFunction = data => {
1520       return data
1521     }
1522     await dynamicThreadPool.addTaskFunction('echo', {
1523       taskFunction: echoTaskFunction,
1524       strategy: WorkerChoiceStrategies.LEAST_ELU
1525     })
1526     expect(dynamicThreadPool.taskFunctions.size).toBe(1)
1527     expect(dynamicThreadPool.taskFunctions.get('echo')).toStrictEqual({
1528       taskFunction: echoTaskFunction,
1529       strategy: WorkerChoiceStrategies.LEAST_ELU
1530     })
1531     expect([
1532       ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys()
1533     ]).toStrictEqual([
1534       WorkerChoiceStrategies.ROUND_ROBIN,
1535       WorkerChoiceStrategies.LEAST_ELU
1536     ])
1537     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1538       { name: DEFAULT_TASK_NAME },
1539       { name: 'test' },
1540       { name: 'echo', strategy: WorkerChoiceStrategies.LEAST_ELU }
1541     ])
1542     await expect(dynamicThreadPool.removeTaskFunction('echo')).resolves.toBe(
1543       true
1544     )
1545     expect(dynamicThreadPool.taskFunctions.size).toBe(0)
1546     expect(dynamicThreadPool.taskFunctions.get('echo')).toBeUndefined()
1547     expect([
1548       ...dynamicThreadPool.workerChoiceStrategiesContext.workerChoiceStrategies.keys()
1549     ]).toStrictEqual([WorkerChoiceStrategies.ROUND_ROBIN])
1550     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1551       { name: DEFAULT_TASK_NAME },
1552       { name: 'test' }
1553     ])
1554     await dynamicThreadPool.destroy()
1555   })
1556
1557   it('Verify that listTaskFunctionsProperties() is working', async () => {
1558     const dynamicThreadPool = new DynamicThreadPool(
1559       Math.floor(numberOfWorkers / 2),
1560       numberOfWorkers,
1561       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1562     )
1563     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1564     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1565       { name: DEFAULT_TASK_NAME },
1566       { name: 'jsonIntegerSerialization' },
1567       { name: 'factorial' },
1568       { name: 'fibonacci' }
1569     ])
1570     await dynamicThreadPool.destroy()
1571     const fixedClusterPool = new FixedClusterPool(
1572       numberOfWorkers,
1573       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
1574     )
1575     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
1576     expect(fixedClusterPool.listTaskFunctionsProperties()).toStrictEqual([
1577       { name: DEFAULT_TASK_NAME },
1578       { name: 'jsonIntegerSerialization' },
1579       { name: 'factorial' },
1580       { name: 'fibonacci' }
1581     ])
1582     await fixedClusterPool.destroy()
1583   })
1584
1585   it('Verify that setDefaultTaskFunction() is working', async () => {
1586     const dynamicThreadPool = new DynamicThreadPool(
1587       Math.floor(numberOfWorkers / 2),
1588       numberOfWorkers,
1589       './tests/worker-files/thread/testMultipleTaskFunctionsWorker.mjs'
1590     )
1591     await waitPoolEvents(dynamicThreadPool, PoolEvents.ready, 1)
1592     const workerId = dynamicThreadPool.workerNodes[0].info.id
1593     await expect(dynamicThreadPool.setDefaultTaskFunction(0)).rejects.toThrow(
1594       new Error(
1595         `Task function operation 'default' failed on worker ${workerId} with error: 'TypeError: name parameter is not a string'`
1596       )
1597     )
1598     await expect(
1599       dynamicThreadPool.setDefaultTaskFunction(DEFAULT_TASK_NAME)
1600     ).rejects.toThrow(
1601       new Error(
1602         `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function reserved name as the default task function'`
1603       )
1604     )
1605     await expect(
1606       dynamicThreadPool.setDefaultTaskFunction('unknown')
1607     ).rejects.toThrow(
1608       new Error(
1609         `Task function operation 'default' failed on worker ${workerId} with error: 'Error: Cannot set the default task function to a non-existing task function'`
1610       )
1611     )
1612     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1613       { name: DEFAULT_TASK_NAME },
1614       { name: 'jsonIntegerSerialization' },
1615       { name: 'factorial' },
1616       { name: 'fibonacci' }
1617     ])
1618     await expect(
1619       dynamicThreadPool.setDefaultTaskFunction('factorial')
1620     ).resolves.toBe(true)
1621     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1622       { name: DEFAULT_TASK_NAME },
1623       { name: 'factorial' },
1624       { name: 'jsonIntegerSerialization' },
1625       { name: 'fibonacci' }
1626     ])
1627     await expect(
1628       dynamicThreadPool.setDefaultTaskFunction('fibonacci')
1629     ).resolves.toBe(true)
1630     expect(dynamicThreadPool.listTaskFunctionsProperties()).toStrictEqual([
1631       { name: DEFAULT_TASK_NAME },
1632       { name: 'fibonacci' },
1633       { name: 'jsonIntegerSerialization' },
1634       { name: 'factorial' }
1635     ])
1636     await dynamicThreadPool.destroy()
1637   })
1638
1639   it('Verify that multiple task functions worker is working', async () => {
1640     const pool = new DynamicClusterPool(
1641       Math.floor(numberOfWorkers / 2),
1642       numberOfWorkers,
1643       './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
1644     )
1645     const data = { n: 10 }
1646     const result0 = await pool.execute(data)
1647     expect(result0).toStrictEqual({ ok: 1 })
1648     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1649     expect(result1).toStrictEqual({ ok: 1 })
1650     const result2 = await pool.execute(data, 'factorial')
1651     expect(result2).toBe(3628800)
1652     const result3 = await pool.execute(data, 'fibonacci')
1653     expect(result3).toBe(55)
1654     expect(pool.info.executingTasks).toBe(0)
1655     expect(pool.info.executedTasks).toBe(4)
1656     for (const workerNode of pool.workerNodes) {
1657       expect(workerNode.info.taskFunctionsProperties).toStrictEqual([
1658         { name: DEFAULT_TASK_NAME },
1659         { name: 'jsonIntegerSerialization' },
1660         { name: 'factorial' },
1661         { name: 'fibonacci' }
1662       ])
1663       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1664       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1665       for (const taskFunctionProperties of pool.listTaskFunctionsProperties()) {
1666         expect(
1667           workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1668         ).toStrictEqual({
1669           tasks: {
1670             executed: expect.any(Number),
1671             executing: 0,
1672             failed: 0,
1673             queued: 0,
1674             sequentiallyStolen: 0,
1675             stolen: 0
1676           },
1677           runTime: {
1678             history: expect.any(CircularBuffer)
1679           },
1680           waitTime: {
1681             history: expect.any(CircularBuffer)
1682           },
1683           elu: {
1684             idle: {
1685               history: expect.any(CircularBuffer)
1686             },
1687             active: {
1688               history: expect.any(CircularBuffer)
1689             }
1690           }
1691         })
1692         expect(
1693           workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1694             .tasks.executed
1695         ).toBeGreaterThan(0)
1696       }
1697       expect(
1698         workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1699       ).toStrictEqual(
1700         workerNode.getTaskFunctionWorkerUsage(
1701           workerNode.info.taskFunctionsProperties[1].name
1702         )
1703       )
1704     }
1705     await pool.destroy()
1706   })
1707
1708   it('Verify that task function objects worker is working', async () => {
1709     const pool = new DynamicThreadPool(
1710       Math.floor(numberOfWorkers / 2),
1711       numberOfWorkers,
1712       './tests/worker-files/thread/testTaskFunctionObjectsWorker.mjs'
1713     )
1714     const data = { n: 10 }
1715     const result0 = await pool.execute(data)
1716     expect(result0).toStrictEqual({ ok: 1 })
1717     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
1718     expect(result1).toStrictEqual({ ok: 1 })
1719     const result2 = await pool.execute(data, 'factorial')
1720     expect(result2).toBe(3628800)
1721     const result3 = await pool.execute(data, 'fibonacci')
1722     expect(result3).toBe(55)
1723     expect(pool.info.executingTasks).toBe(0)
1724     expect(pool.info.executedTasks).toBe(4)
1725     for (const workerNode of pool.workerNodes) {
1726       expect(workerNode.info.taskFunctionsProperties).toStrictEqual([
1727         { name: DEFAULT_TASK_NAME },
1728         { name: 'jsonIntegerSerialization' },
1729         { name: 'factorial' },
1730         { name: 'fibonacci' }
1731       ])
1732       expect(workerNode.taskFunctionsUsage.size).toBe(3)
1733       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
1734       for (const taskFunctionProperties of pool.listTaskFunctionsProperties()) {
1735         expect(
1736           workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1737         ).toStrictEqual({
1738           tasks: {
1739             executed: expect.any(Number),
1740             executing: 0,
1741             failed: 0,
1742             queued: 0,
1743             sequentiallyStolen: 0,
1744             stolen: 0
1745           },
1746           runTime: {
1747             history: expect.any(CircularBuffer)
1748           },
1749           waitTime: {
1750             history: expect.any(CircularBuffer)
1751           },
1752           elu: {
1753             idle: {
1754               history: expect.any(CircularBuffer)
1755             },
1756             active: {
1757               history: expect.any(CircularBuffer)
1758             }
1759           }
1760         })
1761         expect(
1762           workerNode.getTaskFunctionWorkerUsage(taskFunctionProperties.name)
1763             .tasks.executed
1764         ).toBeGreaterThan(0)
1765       }
1766       expect(
1767         workerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME)
1768       ).toStrictEqual(
1769         workerNode.getTaskFunctionWorkerUsage(
1770           workerNode.info.taskFunctionsProperties[1].name
1771         )
1772       )
1773     }
1774     await pool.destroy()
1775   })
1776
1777   it('Verify sendKillMessageToWorker()', async () => {
1778     const pool = new DynamicClusterPool(
1779       Math.floor(numberOfWorkers / 2),
1780       numberOfWorkers,
1781       './tests/worker-files/cluster/testWorker.cjs'
1782     )
1783     const workerNodeKey = 0
1784     await expect(
1785       pool.sendKillMessageToWorker(workerNodeKey)
1786     ).resolves.toBeUndefined()
1787     await pool.destroy()
1788   })
1789
1790   it('Verify sendTaskFunctionOperationToWorker()', async () => {
1791     const pool = new DynamicClusterPool(
1792       Math.floor(numberOfWorkers / 2),
1793       numberOfWorkers,
1794       './tests/worker-files/cluster/testWorker.cjs'
1795     )
1796     const workerNodeKey = 0
1797     await expect(
1798       pool.sendTaskFunctionOperationToWorker(workerNodeKey, {
1799         taskFunctionOperation: 'add',
1800         taskFunctionProperties: { name: 'empty' },
1801         taskFunction: (() => {}).toString()
1802       })
1803     ).resolves.toBe(true)
1804     expect(
1805       pool.workerNodes[workerNodeKey].info.taskFunctionsProperties
1806     ).toStrictEqual([
1807       { name: DEFAULT_TASK_NAME },
1808       { name: 'test' },
1809       { name: 'empty' }
1810     ])
1811     await pool.destroy()
1812   })
1813
1814   it('Verify sendTaskFunctionOperationToWorkers()', async () => {
1815     const pool = new DynamicClusterPool(
1816       Math.floor(numberOfWorkers / 2),
1817       numberOfWorkers,
1818       './tests/worker-files/cluster/testWorker.cjs'
1819     )
1820     await expect(
1821       pool.sendTaskFunctionOperationToWorkers({
1822         taskFunctionOperation: 'add',
1823         taskFunctionProperties: { name: 'empty' },
1824         taskFunction: (() => {}).toString()
1825       })
1826     ).resolves.toBe(true)
1827     for (const workerNode of pool.workerNodes) {
1828       expect(workerNode.info.taskFunctionsProperties).toStrictEqual([
1829         { name: DEFAULT_TASK_NAME },
1830         { name: 'test' },
1831         { name: 'empty' }
1832       ])
1833     }
1834     await pool.destroy()
1835   })
1836 })