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

   1 const { expect } = require('expect')
   2 const {
   3   DynamicClusterPool,
   4   DynamicThreadPool,
   5   FixedClusterPool,
   6   FixedThreadPool,
   7   PoolEvents,
   8   WorkerChoiceStrategies,
   9   PoolTypes,
  10   WorkerTypes
  11 } = require('../../../lib')
  12 const { CircularArray } = require('../../../lib/circular-array')
  13 const { Queue } = require('../../../lib/queue')
  14
  15 describe('Abstract pool test suite', () => {
  16   const numberOfWorkers = 2
  17   class StubPoolWithRemoveAllWorker extends FixedThreadPool {
  18     removeAllWorker () {
  19       this.workerNodes = []
  20       this.promiseResponseMap.clear()
  21     }
  22   }
  23   class StubPoolWithIsMain extends FixedThreadPool {
  24     isMain () {
  25       return false
  26     }
  27   }
  28
  29   it('Simulate pool creation from a non main thread/process', () => {
  30     expect(
  31       () =>
  32         new StubPoolWithIsMain(
  33           numberOfWorkers,
  34           './tests/worker-files/thread/testWorker.js',
  35           {
  36             errorHandler: e => console.error(e)
  37           }
  38         )
  39     ).toThrowError('Cannot start a pool from a worker!')
  40   })
  41
  42   it('Verify that filePath is checked', () => {
  43     const expectedError = new Error(
  44       'Please specify a file with a worker implementation'
  45     )
  46     expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError(
  47       expectedError
  48     )
  49     expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError(
  50       expectedError
  51     )
  52   })
  53
  54   it('Verify that numberOfWorkers is checked', () => {
  55     expect(() => new FixedThreadPool()).toThrowError(
  56       'Cannot instantiate a pool without specifying the number of workers'
  57     )
  58   })
  59
  60   it('Verify that a negative number of workers is checked', () => {
  61     expect(
  62       () =>
  63         new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
  64     ).toThrowError(
  65       new RangeError(
  66         'Cannot instantiate a pool with a negative number of workers'
  67       )
  68     )
  69   })
  70
  71   it('Verify that a non integer number of workers is checked', () => {
  72     expect(
  73       () =>
  74         new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.js')
  75     ).toThrowError(
  76       new TypeError(
  77         'Cannot instantiate a pool with a non safe integer number of workers'
  78       )
  79     )
  80   })
  81
  82   it('Verify that pool options are checked', async () => {
  83     let pool = new FixedThreadPool(
  84       numberOfWorkers,
  85       './tests/worker-files/thread/testWorker.js'
  86     )
  87     expect(pool.emitter).toBeDefined()
  88     expect(pool.opts.enableEvents).toBe(true)
  89     expect(pool.opts.restartWorkerOnError).toBe(true)
  90     expect(pool.opts.enableTasksQueue).toBe(false)
  91     expect(pool.opts.tasksQueueOptions).toBeUndefined()
  92     expect(pool.opts.workerChoiceStrategy).toBe(
  93       WorkerChoiceStrategies.ROUND_ROBIN
  94     )
  95     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
  96       runTime: { median: false },
  97       waitTime: { median: false },
  98       elu: { median: false }
  99     })
 100     expect(pool.opts.messageHandler).toBeUndefined()
 101     expect(pool.opts.errorHandler).toBeUndefined()
 102     expect(pool.opts.onlineHandler).toBeUndefined()
 103     expect(pool.opts.exitHandler).toBeUndefined()
 104     await pool.destroy()
 105     const testHandler = () => console.log('test handler executed')
 106     pool = new FixedThreadPool(
 107       numberOfWorkers,
 108       './tests/worker-files/thread/testWorker.js',
 109       {
 110         workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
 111         workerChoiceStrategyOptions: {
 112           runTime: { median: true },
 113           weights: { 0: 300, 1: 200 }
 114         },
 115         enableEvents: false,
 116         restartWorkerOnError: false,
 117         enableTasksQueue: true,
 118         tasksQueueOptions: { concurrency: 2 },
 119         messageHandler: testHandler,
 120         errorHandler: testHandler,
 121         onlineHandler: testHandler,
 122         exitHandler: testHandler
 123       }
 124     )
 125     expect(pool.emitter).toBeUndefined()
 126     expect(pool.opts.enableEvents).toBe(false)
 127     expect(pool.opts.restartWorkerOnError).toBe(false)
 128     expect(pool.opts.enableTasksQueue).toBe(true)
 129     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 130     expect(pool.opts.workerChoiceStrategy).toBe(
 131       WorkerChoiceStrategies.LEAST_USED
 132     )
 133     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 134       runTime: { median: true },
 135       weights: { 0: 300, 1: 200 }
 136     })
 137     expect(pool.opts.messageHandler).toStrictEqual(testHandler)
 138     expect(pool.opts.errorHandler).toStrictEqual(testHandler)
 139     expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
 140     expect(pool.opts.exitHandler).toStrictEqual(testHandler)
 141     await pool.destroy()
 142   })
 143
 144   it('Verify that pool options are validated', async () => {
 145     expect(
 146       () =>
 147         new FixedThreadPool(
 148           numberOfWorkers,
 149           './tests/worker-files/thread/testWorker.js',
 150           {
 151             enableTasksQueue: true,
 152             tasksQueueOptions: { concurrency: 0 }
 153           }
 154         )
 155     ).toThrowError("Invalid worker tasks concurrency '0'")
 156     expect(
 157       () =>
 158         new FixedThreadPool(
 159           numberOfWorkers,
 160           './tests/worker-files/thread/testWorker.js',
 161           {
 162             workerChoiceStrategy: 'invalidStrategy'
 163           }
 164         )
 165     ).toThrowError("Invalid worker choice strategy 'invalidStrategy'")
 166     expect(
 167       () =>
 168         new FixedThreadPool(
 169           numberOfWorkers,
 170           './tests/worker-files/thread/testWorker.js',
 171           {
 172             workerChoiceStrategyOptions: { weights: {} }
 173           }
 174         )
 175     ).toThrowError(
 176       'Invalid worker choice strategy options: must have a weight for each worker node'
 177     )
 178   })
 179
 180   it('Verify that worker choice strategy options can be set', async () => {
 181     const pool = new FixedThreadPool(
 182       numberOfWorkers,
 183       './tests/worker-files/thread/testWorker.js',
 184       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 185     )
 186     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 187       runTime: { median: false },
 188       waitTime: { median: false },
 189       elu: { median: false }
 190     })
 191     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 192       .workerChoiceStrategies) {
 193       expect(workerChoiceStrategy.opts).toStrictEqual({
 194         runTime: { median: false },
 195         waitTime: { median: false },
 196         elu: { median: false }
 197       })
 198     }
 199     expect(
 200       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 201     ).toStrictEqual({
 202       runTime: {
 203         aggregate: true,
 204         average: true,
 205         median: false
 206       },
 207       waitTime: {
 208         aggregate: false,
 209         average: false,
 210         median: false
 211       },
 212       elu: {
 213         aggregate: true,
 214         average: true,
 215         median: false
 216       }
 217     })
 218     pool.setWorkerChoiceStrategyOptions({
 219       runTime: { median: true },
 220       elu: { median: true }
 221     })
 222     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 223       runTime: { median: true },
 224       elu: { median: true }
 225     })
 226     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 227       .workerChoiceStrategies) {
 228       expect(workerChoiceStrategy.opts).toStrictEqual({
 229         runTime: { median: true },
 230         elu: { median: true }
 231       })
 232     }
 233     expect(
 234       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 235     ).toStrictEqual({
 236       runTime: {
 237         aggregate: true,
 238         average: false,
 239         median: true
 240       },
 241       waitTime: {
 242         aggregate: false,
 243         average: false,
 244         median: false
 245       },
 246       elu: {
 247         aggregate: true,
 248         average: false,
 249         median: true
 250       }
 251     })
 252     pool.setWorkerChoiceStrategyOptions({
 253       runTime: { median: false },
 254       elu: { median: false }
 255     })
 256     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 257       runTime: { median: false },
 258       elu: { median: false }
 259     })
 260     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 261       .workerChoiceStrategies) {
 262       expect(workerChoiceStrategy.opts).toStrictEqual({
 263         runTime: { median: false },
 264         elu: { median: false }
 265       })
 266     }
 267     expect(
 268       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 269     ).toStrictEqual({
 270       runTime: {
 271         aggregate: true,
 272         average: true,
 273         median: false
 274       },
 275       waitTime: {
 276         aggregate: false,
 277         average: false,
 278         median: false
 279       },
 280       elu: {
 281         aggregate: true,
 282         average: true,
 283         median: false
 284       }
 285     })
 286     await pool.destroy()
 287   })
 288
 289   it('Verify that tasks queue can be enabled/disabled', async () => {
 290     const pool = new FixedThreadPool(
 291       numberOfWorkers,
 292       './tests/worker-files/thread/testWorker.js'
 293     )
 294     expect(pool.opts.enableTasksQueue).toBe(false)
 295     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 296     pool.enableTasksQueue(true)
 297     expect(pool.opts.enableTasksQueue).toBe(true)
 298     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 299     pool.enableTasksQueue(true, { concurrency: 2 })
 300     expect(pool.opts.enableTasksQueue).toBe(true)
 301     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 302     pool.enableTasksQueue(false)
 303     expect(pool.opts.enableTasksQueue).toBe(false)
 304     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 305     await pool.destroy()
 306   })
 307
 308   it('Verify that tasks queue options can be set', async () => {
 309     const pool = new FixedThreadPool(
 310       numberOfWorkers,
 311       './tests/worker-files/thread/testWorker.js',
 312       { enableTasksQueue: true }
 313     )
 314     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 315     pool.setTasksQueueOptions({ concurrency: 2 })
 316     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 317     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
 318       "Invalid worker tasks concurrency '0'"
 319     )
 320     await pool.destroy()
 321   })
 322
 323   it('Verify that pool info is set', async () => {
 324     let pool = new FixedThreadPool(
 325       numberOfWorkers,
 326       './tests/worker-files/thread/testWorker.js'
 327     )
 328     expect(pool.info).toStrictEqual({
 329       type: PoolTypes.fixed,
 330       worker: WorkerTypes.thread,
 331       minSize: numberOfWorkers,
 332       maxSize: numberOfWorkers,
 333       workerNodes: numberOfWorkers,
 334       idleWorkerNodes: numberOfWorkers,
 335       busyWorkerNodes: 0,
 336       executedTasks: 0,
 337       executingTasks: 0,
 338       queuedTasks: 0,
 339       maxQueuedTasks: 0,
 340       failedTasks: 0
 341     })
 342     await pool.destroy()
 343     pool = new DynamicClusterPool(
 344       numberOfWorkers,
 345       numberOfWorkers * 2,
 346       './tests/worker-files/thread/testWorker.js'
 347     )
 348     expect(pool.info).toStrictEqual({
 349       type: PoolTypes.dynamic,
 350       worker: WorkerTypes.cluster,
 351       minSize: numberOfWorkers,
 352       maxSize: numberOfWorkers * 2,
 353       workerNodes: numberOfWorkers,
 354       idleWorkerNodes: numberOfWorkers,
 355       busyWorkerNodes: 0,
 356       executedTasks: 0,
 357       executingTasks: 0,
 358       queuedTasks: 0,
 359       maxQueuedTasks: 0,
 360       failedTasks: 0
 361     })
 362     await pool.destroy()
 363   })
 364
 365   it('Simulate worker not found', async () => {
 366     const pool = new StubPoolWithRemoveAllWorker(
 367       numberOfWorkers,
 368       './tests/worker-files/cluster/testWorker.js',
 369       {
 370         errorHandler: e => console.error(e)
 371       }
 372     )
 373     expect(pool.workerNodes.length).toBe(numberOfWorkers)
 374     // Simulate worker not found.
 375     pool.removeAllWorker()
 376     expect(pool.workerNodes.length).toBe(0)
 377     await pool.destroy()
 378   })
 379
 380   it('Verify that worker pool tasks usage are initialized', async () => {
 381     const pool = new FixedClusterPool(
 382       numberOfWorkers,
 383       './tests/worker-files/cluster/testWorker.js'
 384     )
 385     for (const workerNode of pool.workerNodes) {
 386       expect(workerNode.workerUsage).toStrictEqual({
 387         tasks: {
 388           executed: 0,
 389           executing: 0,
 390           queued: 0,
 391           failed: 0
 392         },
 393         runTime: {
 394           aggregate: 0,
 395           average: 0,
 396           median: 0,
 397           history: expect.any(CircularArray)
 398         },
 399         waitTime: {
 400           aggregate: 0,
 401           average: 0,
 402           median: 0,
 403           history: expect.any(CircularArray)
 404         },
 405         elu: {
 406           idle: {
 407             aggregate: 0,
 408             average: 0,
 409             median: 0,
 410             history: expect.any(CircularArray)
 411           },
 412           active: {
 413             aggregate: 0,
 414             average: 0,
 415             median: 0,
 416             history: expect.any(CircularArray)
 417           },
 418           utilization: 0
 419         }
 420       })
 421     }
 422     await pool.destroy()
 423   })
 424
 425   it('Verify that worker pool tasks queue are initialized', async () => {
 426     const pool = new FixedClusterPool(
 427       numberOfWorkers,
 428       './tests/worker-files/cluster/testWorker.js'
 429     )
 430     for (const workerNode of pool.workerNodes) {
 431       expect(workerNode.tasksQueue).toBeDefined()
 432       expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
 433       expect(workerNode.tasksQueue.size).toBe(0)
 434     }
 435     await pool.destroy()
 436   })
 437
 438   it('Verify that worker pool tasks usage are computed', async () => {
 439     const pool = new FixedClusterPool(
 440       numberOfWorkers,
 441       './tests/worker-files/cluster/testWorker.js'
 442     )
 443     const promises = new Set()
 444     const maxMultiplier = 2
 445     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 446       promises.add(pool.execute())
 447     }
 448     for (const workerNode of pool.workerNodes) {
 449       expect(workerNode.workerUsage).toStrictEqual({
 450         tasks: {
 451           executed: 0,
 452           executing: maxMultiplier,
 453           queued: 0,
 454           failed: 0
 455         },
 456         runTime: {
 457           aggregate: 0,
 458           average: 0,
 459           median: 0,
 460           history: expect.any(CircularArray)
 461         },
 462         waitTime: {
 463           aggregate: 0,
 464           average: 0,
 465           median: 0,
 466           history: expect.any(CircularArray)
 467         },
 468         elu: {
 469           idle: {
 470             aggregate: 0,
 471             average: 0,
 472             median: 0,
 473             history: expect.any(CircularArray)
 474           },
 475           active: {
 476             aggregate: 0,
 477             average: 0,
 478             median: 0,
 479             history: expect.any(CircularArray)
 480           },
 481           utilization: 0
 482         }
 483       })
 484     }
 485     await Promise.all(promises)
 486     for (const workerNode of pool.workerNodes) {
 487       expect(workerNode.workerUsage).toStrictEqual({
 488         tasks: {
 489           executed: maxMultiplier,
 490           executing: 0,
 491           queued: 0,
 492           failed: 0
 493         },
 494         runTime: {
 495           aggregate: 0,
 496           average: 0,
 497           median: 0,
 498           history: expect.any(CircularArray)
 499         },
 500         waitTime: {
 501           aggregate: 0,
 502           average: 0,
 503           median: 0,
 504           history: expect.any(CircularArray)
 505         },
 506         elu: {
 507           idle: {
 508             aggregate: 0,
 509             average: 0,
 510             median: 0,
 511             history: expect.any(CircularArray)
 512           },
 513           active: {
 514             aggregate: 0,
 515             average: 0,
 516             median: 0,
 517             history: expect.any(CircularArray)
 518           },
 519           utilization: 0
 520         }
 521       })
 522     }
 523     await pool.destroy()
 524   })
 525
 526   it('Verify that worker pool tasks usage are reset at worker choice strategy change', async () => {
 527     const pool = new DynamicThreadPool(
 528       numberOfWorkers,
 529       numberOfWorkers,
 530       './tests/worker-files/thread/testWorker.js'
 531     )
 532     const promises = new Set()
 533     const maxMultiplier = 2
 534     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 535       promises.add(pool.execute())
 536     }
 537     await Promise.all(promises)
 538     for (const workerNode of pool.workerNodes) {
 539       expect(workerNode.workerUsage).toStrictEqual({
 540         tasks: {
 541           executed: expect.any(Number),
 542           executing: 0,
 543           queued: 0,
 544           failed: 0
 545         },
 546         runTime: {
 547           aggregate: 0,
 548           average: 0,
 549           median: 0,
 550           history: expect.any(CircularArray)
 551         },
 552         waitTime: {
 553           aggregate: 0,
 554           average: 0,
 555           median: 0,
 556           history: expect.any(CircularArray)
 557         },
 558         elu: {
 559           idle: {
 560             aggregate: 0,
 561             average: 0,
 562             median: 0,
 563             history: expect.any(CircularArray)
 564           },
 565           active: {
 566             aggregate: 0,
 567             average: 0,
 568             median: 0,
 569             history: expect.any(CircularArray)
 570           },
 571           utilization: 0
 572         }
 573       })
 574       expect(workerNode.workerUsage.tasks.executed).toBeGreaterThan(0)
 575       expect(workerNode.workerUsage.tasks.executed).toBeLessThanOrEqual(
 576         maxMultiplier
 577       )
 578     }
 579     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 580     for (const workerNode of pool.workerNodes) {
 581       expect(workerNode.workerUsage).toStrictEqual({
 582         tasks: {
 583           executed: 0,
 584           executing: 0,
 585           queued: 0,
 586           failed: 0
 587         },
 588         runTime: {
 589           aggregate: 0,
 590           average: 0,
 591           median: 0,
 592           history: expect.any(CircularArray)
 593         },
 594         waitTime: {
 595           aggregate: 0,
 596           average: 0,
 597           median: 0,
 598           history: expect.any(CircularArray)
 599         },
 600         elu: {
 601           idle: {
 602             aggregate: 0,
 603             average: 0,
 604             median: 0,
 605             history: expect.any(CircularArray)
 606           },
 607           active: {
 608             aggregate: 0,
 609             average: 0,
 610             median: 0,
 611             history: expect.any(CircularArray)
 612           },
 613           utilization: 0
 614         }
 615       })
 616       expect(workerNode.workerUsage.runTime.history.length).toBe(0)
 617       expect(workerNode.workerUsage.waitTime.history.length).toBe(0)
 618     }
 619     await pool.destroy()
 620   })
 621
 622   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 623     const pool = new DynamicThreadPool(
 624       numberOfWorkers,
 625       numberOfWorkers,
 626       './tests/worker-files/thread/testWorker.js'
 627     )
 628     const promises = new Set()
 629     let poolFull = 0
 630     let poolInfo
 631     pool.emitter.on(PoolEvents.full, info => {
 632       ++poolFull
 633       poolInfo = info
 634     })
 635     for (let i = 0; i < numberOfWorkers * 2; i++) {
 636       promises.add(pool.execute())
 637     }
 638     await Promise.all(promises)
 639     // The `full` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool.
 640     // So in total numberOfWorkers * 2 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool with min = max = numberOfWorkers.
 641     expect(poolFull).toBe(numberOfWorkers * 2)
 642     expect(poolInfo).toStrictEqual({
 643       type: PoolTypes.dynamic,
 644       worker: WorkerTypes.thread,
 645       minSize: expect.any(Number),
 646       maxSize: expect.any(Number),
 647       workerNodes: expect.any(Number),
 648       idleWorkerNodes: expect.any(Number),
 649       busyWorkerNodes: expect.any(Number),
 650       executedTasks: expect.any(Number),
 651       executingTasks: expect.any(Number),
 652       queuedTasks: expect.any(Number),
 653       maxQueuedTasks: expect.any(Number),
 654       failedTasks: expect.any(Number)
 655     })
 656     await pool.destroy()
 657   })
 658
 659   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 660     const pool = new FixedThreadPool(
 661       numberOfWorkers,
 662       './tests/worker-files/thread/testWorker.js'
 663     )
 664     const promises = new Set()
 665     let poolBusy = 0
 666     let poolInfo
 667     pool.emitter.on(PoolEvents.busy, info => {
 668       ++poolBusy
 669       poolInfo = info
 670     })
 671     for (let i = 0; i < numberOfWorkers * 2; i++) {
 672       promises.add(pool.execute())
 673     }
 674     await Promise.all(promises)
 675     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 676     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 677     expect(poolBusy).toBe(numberOfWorkers + 1)
 678     expect(poolInfo).toStrictEqual({
 679       type: PoolTypes.fixed,
 680       worker: WorkerTypes.thread,
 681       minSize: expect.any(Number),
 682       maxSize: expect.any(Number),
 683       workerNodes: expect.any(Number),
 684       idleWorkerNodes: expect.any(Number),
 685       busyWorkerNodes: expect.any(Number),
 686       executedTasks: expect.any(Number),
 687       executingTasks: expect.any(Number),
 688       queuedTasks: expect.any(Number),
 689       maxQueuedTasks: expect.any(Number),
 690       failedTasks: expect.any(Number)
 691     })
 692     await pool.destroy()
 693   })
 694
 695   it('Verify that multiple tasks worker is working', async () => {
 696     const pool = new DynamicClusterPool(
 697       numberOfWorkers,
 698       numberOfWorkers * 2,
 699       './tests/worker-files/cluster/testMultiTasksWorker.js'
 700     )
 701     const data = { n: 10 }
 702     const result0 = await pool.execute(data)
 703     expect(result0).toBe(false)
 704     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
 705     expect(result1).toBe(false)
 706     const result2 = await pool.execute(data, 'factorial')
 707     expect(result2).toBe(3628800)
 708     const result3 = await pool.execute(data, 'fibonacci')
 709     expect(result3).toBe(89)
 710   })
 711 })