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     })
  99     expect(pool.opts.messageHandler).toBeUndefined()
 100     expect(pool.opts.errorHandler).toBeUndefined()
 101     expect(pool.opts.onlineHandler).toBeUndefined()
 102     expect(pool.opts.exitHandler).toBeUndefined()
 103     await pool.destroy()
 104     const testHandler = () => console.log('test handler executed')
 105     pool = new FixedThreadPool(
 106       numberOfWorkers,
 107       './tests/worker-files/thread/testWorker.js',
 108       {
 109         workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
 110         workerChoiceStrategyOptions: {
 111           runTime: { median: true },
 112           weights: { 0: 300, 1: 200 }
 113         },
 114         enableEvents: false,
 115         restartWorkerOnError: false,
 116         enableTasksQueue: true,
 117         tasksQueueOptions: { concurrency: 2 },
 118         messageHandler: testHandler,
 119         errorHandler: testHandler,
 120         onlineHandler: testHandler,
 121         exitHandler: testHandler
 122       }
 123     )
 124     expect(pool.emitter).toBeUndefined()
 125     expect(pool.opts.enableEvents).toBe(false)
 126     expect(pool.opts.restartWorkerOnError).toBe(false)
 127     expect(pool.opts.enableTasksQueue).toBe(true)
 128     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 129     expect(pool.opts.workerChoiceStrategy).toBe(
 130       WorkerChoiceStrategies.LEAST_USED
 131     )
 132     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 133       runTime: { median: true },
 134       weights: { 0: 300, 1: 200 }
 135     })
 136     expect(pool.opts.messageHandler).toStrictEqual(testHandler)
 137     expect(pool.opts.errorHandler).toStrictEqual(testHandler)
 138     expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
 139     expect(pool.opts.exitHandler).toStrictEqual(testHandler)
 140     await pool.destroy()
 141   })
 142
 143   it('Verify that pool options are validated', async () => {
 144     expect(
 145       () =>
 146         new FixedThreadPool(
 147           numberOfWorkers,
 148           './tests/worker-files/thread/testWorker.js',
 149           {
 150             enableTasksQueue: true,
 151             tasksQueueOptions: { concurrency: 0 }
 152           }
 153         )
 154     ).toThrowError("Invalid worker tasks concurrency '0'")
 155     expect(
 156       () =>
 157         new FixedThreadPool(
 158           numberOfWorkers,
 159           './tests/worker-files/thread/testWorker.js',
 160           {
 161             workerChoiceStrategy: 'invalidStrategy'
 162           }
 163         )
 164     ).toThrowError("Invalid worker choice strategy 'invalidStrategy'")
 165     expect(
 166       () =>
 167         new FixedThreadPool(
 168           numberOfWorkers,
 169           './tests/worker-files/thread/testWorker.js',
 170           {
 171             workerChoiceStrategyOptions: { weights: {} }
 172           }
 173         )
 174     ).toThrowError(
 175       'Invalid worker choice strategy options: must have a weight for each worker node'
 176     )
 177   })
 178
 179   it('Verify that worker choice strategy options can be set', async () => {
 180     const pool = new FixedThreadPool(
 181       numberOfWorkers,
 182       './tests/worker-files/thread/testWorker.js',
 183       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 184     )
 185     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 186       runTime: { median: false },
 187       waitTime: { median: false }
 188     })
 189     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 190       .workerChoiceStrategies) {
 191       expect(workerChoiceStrategy.opts).toStrictEqual({
 192         runTime: { median: false },
 193         waitTime: { median: false }
 194       })
 195     }
 196     expect(
 197       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 198     ).toStrictEqual({
 199       runTime: {
 200         aggregate: true,
 201         average: true,
 202         median: false
 203       },
 204       waitTime: {
 205         aggregate: false,
 206         average: false,
 207         median: false
 208       },
 209       elu: false
 210     })
 211     pool.setWorkerChoiceStrategyOptions({ runTime: { median: true } })
 212     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 213       runTime: { median: true }
 214     })
 215     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 216       .workerChoiceStrategies) {
 217       expect(workerChoiceStrategy.opts).toStrictEqual({
 218         runTime: { median: true }
 219       })
 220     }
 221     expect(
 222       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 223     ).toStrictEqual({
 224       runTime: {
 225         aggregate: true,
 226         average: false,
 227         median: true
 228       },
 229       waitTime: {
 230         aggregate: false,
 231         average: false,
 232         median: false
 233       },
 234       elu: false
 235     })
 236     pool.setWorkerChoiceStrategyOptions({ runTime: { median: false } })
 237     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 238       runTime: { median: false }
 239     })
 240     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 241       .workerChoiceStrategies) {
 242       expect(workerChoiceStrategy.opts).toStrictEqual({
 243         runTime: { median: false }
 244       })
 245     }
 246     expect(
 247       pool.workerChoiceStrategyContext.getTaskStatisticsRequirements()
 248     ).toStrictEqual({
 249       runTime: {
 250         aggregate: true,
 251         average: true,
 252         median: false
 253       },
 254       waitTime: {
 255         aggregate: false,
 256         average: false,
 257         median: false
 258       },
 259       elu: false
 260     })
 261     await pool.destroy()
 262   })
 263
 264   it('Verify that tasks queue can be enabled/disabled', async () => {
 265     const pool = new FixedThreadPool(
 266       numberOfWorkers,
 267       './tests/worker-files/thread/testWorker.js'
 268     )
 269     expect(pool.opts.enableTasksQueue).toBe(false)
 270     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 271     pool.enableTasksQueue(true)
 272     expect(pool.opts.enableTasksQueue).toBe(true)
 273     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 274     pool.enableTasksQueue(true, { concurrency: 2 })
 275     expect(pool.opts.enableTasksQueue).toBe(true)
 276     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 277     pool.enableTasksQueue(false)
 278     expect(pool.opts.enableTasksQueue).toBe(false)
 279     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 280     await pool.destroy()
 281   })
 282
 283   it('Verify that tasks queue options can be set', async () => {
 284     const pool = new FixedThreadPool(
 285       numberOfWorkers,
 286       './tests/worker-files/thread/testWorker.js',
 287       { enableTasksQueue: true }
 288     )
 289     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 290     pool.setTasksQueueOptions({ concurrency: 2 })
 291     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 292     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
 293       "Invalid worker tasks concurrency '0'"
 294     )
 295     await pool.destroy()
 296   })
 297
 298   it('Verify that pool info is set', async () => {
 299     let pool = new FixedThreadPool(
 300       numberOfWorkers,
 301       './tests/worker-files/thread/testWorker.js'
 302     )
 303     expect(pool.info).toStrictEqual({
 304       type: PoolTypes.fixed,
 305       worker: WorkerTypes.thread,
 306       minSize: numberOfWorkers,
 307       maxSize: numberOfWorkers,
 308       workerNodes: numberOfWorkers,
 309       idleWorkerNodes: numberOfWorkers,
 310       busyWorkerNodes: 0,
 311       executedTasks: 0,
 312       executingTasks: 0,
 313       queuedTasks: 0,
 314       maxQueuedTasks: 0,
 315       failedTasks: 0
 316     })
 317     await pool.destroy()
 318     pool = new DynamicClusterPool(
 319       numberOfWorkers,
 320       numberOfWorkers * 2,
 321       './tests/worker-files/thread/testWorker.js'
 322     )
 323     expect(pool.info).toStrictEqual({
 324       type: PoolTypes.dynamic,
 325       worker: WorkerTypes.cluster,
 326       minSize: numberOfWorkers,
 327       maxSize: numberOfWorkers * 2,
 328       workerNodes: numberOfWorkers,
 329       idleWorkerNodes: numberOfWorkers,
 330       busyWorkerNodes: 0,
 331       executedTasks: 0,
 332       executingTasks: 0,
 333       queuedTasks: 0,
 334       maxQueuedTasks: 0,
 335       failedTasks: 0
 336     })
 337     await pool.destroy()
 338   })
 339
 340   it('Simulate worker not found', async () => {
 341     const pool = new StubPoolWithRemoveAllWorker(
 342       numberOfWorkers,
 343       './tests/worker-files/cluster/testWorker.js',
 344       {
 345         errorHandler: e => console.error(e)
 346       }
 347     )
 348     expect(pool.workerNodes.length).toBe(numberOfWorkers)
 349     // Simulate worker not found.
 350     pool.removeAllWorker()
 351     expect(pool.workerNodes.length).toBe(0)
 352     await pool.destroy()
 353   })
 354
 355   it('Verify that worker pool tasks usage are initialized', async () => {
 356     const pool = new FixedClusterPool(
 357       numberOfWorkers,
 358       './tests/worker-files/cluster/testWorker.js'
 359     )
 360     for (const workerNode of pool.workerNodes) {
 361       expect(workerNode.workerUsage).toStrictEqual({
 362         tasks: {
 363           executed: 0,
 364           executing: 0,
 365           queued: 0,
 366           failed: 0
 367         },
 368         runTime: {
 369           aggregate: 0,
 370           average: 0,
 371           median: 0,
 372           history: expect.any(CircularArray)
 373         },
 374         waitTime: {
 375           aggregate: 0,
 376           average: 0,
 377           median: 0,
 378           history: expect.any(CircularArray)
 379         },
 380         elu: undefined
 381       })
 382     }
 383     await pool.destroy()
 384   })
 385
 386   it('Verify that worker pool tasks queue are initialized', async () => {
 387     const pool = new FixedClusterPool(
 388       numberOfWorkers,
 389       './tests/worker-files/cluster/testWorker.js'
 390     )
 391     for (const workerNode of pool.workerNodes) {
 392       expect(workerNode.tasksQueue).toBeDefined()
 393       expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
 394       expect(workerNode.tasksQueue.size).toBe(0)
 395     }
 396     await pool.destroy()
 397   })
 398
 399   it('Verify that worker pool tasks usage are computed', async () => {
 400     const pool = new FixedClusterPool(
 401       numberOfWorkers,
 402       './tests/worker-files/cluster/testWorker.js'
 403     )
 404     const promises = new Set()
 405     const maxMultiplier = 2
 406     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 407       promises.add(pool.execute())
 408     }
 409     for (const workerNode of pool.workerNodes) {
 410       expect(workerNode.workerUsage).toStrictEqual({
 411         tasks: {
 412           executed: 0,
 413           executing: maxMultiplier,
 414           queued: 0,
 415           failed: 0
 416         },
 417         runTime: {
 418           aggregate: 0,
 419           average: 0,
 420           median: 0,
 421           history: expect.any(CircularArray)
 422         },
 423         waitTime: {
 424           aggregate: 0,
 425           average: 0,
 426           median: 0,
 427           history: expect.any(CircularArray)
 428         },
 429         elu: undefined
 430       })
 431     }
 432     await Promise.all(promises)
 433     for (const workerNode of pool.workerNodes) {
 434       expect(workerNode.workerUsage).toStrictEqual({
 435         tasks: {
 436           executed: maxMultiplier,
 437           executing: 0,
 438           queued: 0,
 439           failed: 0
 440         },
 441         runTime: {
 442           aggregate: 0,
 443           average: 0,
 444           median: 0,
 445           history: expect.any(CircularArray)
 446         },
 447         waitTime: {
 448           aggregate: 0,
 449           average: 0,
 450           median: 0,
 451           history: expect.any(CircularArray)
 452         },
 453         elu: undefined
 454       })
 455     }
 456     await pool.destroy()
 457   })
 458
 459   it('Verify that worker pool tasks usage are reset at worker choice strategy change', async () => {
 460     const pool = new DynamicThreadPool(
 461       numberOfWorkers,
 462       numberOfWorkers,
 463       './tests/worker-files/thread/testWorker.js'
 464     )
 465     const promises = new Set()
 466     const maxMultiplier = 2
 467     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 468       promises.add(pool.execute())
 469     }
 470     await Promise.all(promises)
 471     for (const workerNode of pool.workerNodes) {
 472       expect(workerNode.workerUsage).toStrictEqual({
 473         tasks: {
 474           executed: expect.any(Number),
 475           executing: 0,
 476           queued: 0,
 477           failed: 0
 478         },
 479         runTime: {
 480           aggregate: 0,
 481           average: 0,
 482           median: 0,
 483           history: expect.any(CircularArray)
 484         },
 485         waitTime: {
 486           aggregate: 0,
 487           average: 0,
 488           median: 0,
 489           history: expect.any(CircularArray)
 490         },
 491         elu: undefined
 492       })
 493       expect(workerNode.workerUsage.tasks.executed).toBeGreaterThan(0)
 494       expect(workerNode.workerUsage.tasks.executed).toBeLessThanOrEqual(
 495         maxMultiplier
 496       )
 497     }
 498     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 499     for (const workerNode of pool.workerNodes) {
 500       expect(workerNode.workerUsage).toStrictEqual({
 501         tasks: {
 502           executed: 0,
 503           executing: 0,
 504           queued: 0,
 505           failed: 0
 506         },
 507         runTime: {
 508           aggregate: 0,
 509           average: 0,
 510           median: 0,
 511           history: expect.any(CircularArray)
 512         },
 513         waitTime: {
 514           aggregate: 0,
 515           average: 0,
 516           median: 0,
 517           history: expect.any(CircularArray)
 518         },
 519         elu: undefined
 520       })
 521       expect(workerNode.workerUsage.runTime.history.length).toBe(0)
 522       expect(workerNode.workerUsage.waitTime.history.length).toBe(0)
 523     }
 524     await pool.destroy()
 525   })
 526
 527   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 528     const pool = new DynamicThreadPool(
 529       numberOfWorkers,
 530       numberOfWorkers,
 531       './tests/worker-files/thread/testWorker.js'
 532     )
 533     const promises = new Set()
 534     let poolFull = 0
 535     let poolInfo
 536     pool.emitter.on(PoolEvents.full, info => {
 537       ++poolFull
 538       poolInfo = info
 539     })
 540     for (let i = 0; i < numberOfWorkers * 2; i++) {
 541       promises.add(pool.execute())
 542     }
 543     await Promise.all(promises)
 544     // The `full` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool.
 545     // So in total numberOfWorkers * 2 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool with min = max = numberOfWorkers.
 546     expect(poolFull).toBe(numberOfWorkers * 2)
 547     expect(poolInfo).toStrictEqual({
 548       type: PoolTypes.dynamic,
 549       worker: WorkerTypes.thread,
 550       minSize: expect.any(Number),
 551       maxSize: expect.any(Number),
 552       workerNodes: expect.any(Number),
 553       idleWorkerNodes: expect.any(Number),
 554       busyWorkerNodes: expect.any(Number),
 555       executedTasks: expect.any(Number),
 556       executingTasks: expect.any(Number),
 557       queuedTasks: expect.any(Number),
 558       maxQueuedTasks: expect.any(Number),
 559       failedTasks: expect.any(Number)
 560     })
 561     await pool.destroy()
 562   })
 563
 564   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 565     const pool = new FixedThreadPool(
 566       numberOfWorkers,
 567       './tests/worker-files/thread/testWorker.js'
 568     )
 569     const promises = new Set()
 570     let poolBusy = 0
 571     let poolInfo
 572     pool.emitter.on(PoolEvents.busy, info => {
 573       ++poolBusy
 574       poolInfo = info
 575     })
 576     for (let i = 0; i < numberOfWorkers * 2; i++) {
 577       promises.add(pool.execute())
 578     }
 579     await Promise.all(promises)
 580     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 581     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 582     expect(poolBusy).toBe(numberOfWorkers + 1)
 583     expect(poolInfo).toStrictEqual({
 584       type: PoolTypes.fixed,
 585       worker: WorkerTypes.thread,
 586       minSize: expect.any(Number),
 587       maxSize: expect.any(Number),
 588       workerNodes: expect.any(Number),
 589       idleWorkerNodes: expect.any(Number),
 590       busyWorkerNodes: expect.any(Number),
 591       executedTasks: expect.any(Number),
 592       executingTasks: expect.any(Number),
 593       queuedTasks: expect.any(Number),
 594       maxQueuedTasks: expect.any(Number),
 595       failedTasks: expect.any(Number)
 596     })
 597     await pool.destroy()
 598   })
 599
 600   it('Verify that multiple tasks worker is working', async () => {
 601     const pool = new DynamicClusterPool(
 602       numberOfWorkers,
 603       numberOfWorkers * 2,
 604       './tests/worker-files/cluster/testMultiTasksWorker.js'
 605     )
 606     const data = { n: 10 }
 607     const result0 = await pool.execute(data)
 608     expect(result0).toBe(false)
 609     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
 610     expect(result1).toBe(false)
 611     const result2 = await pool.execute(data, 'factorial')
 612     expect(result2).toBe(3628800)
 613     const result3 = await pool.execute(data, 'fibonacci')
 614     expect(result3).toBe(89)
 615   })
 616 })