tests/pools/abstract-pool.test.mjs

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