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 } = require('../../../lib')
  11 const { CircularArray } = require('../../../lib/circular-array')
  12 const { Queue } = require('../../../lib/queue')
  13
  14 describe('Abstract pool test suite', () => {
  15   const numberOfWorkers = 2
  16   class StubPoolWithRemoveAllWorker extends FixedThreadPool {
  17     removeAllWorker () {
  18       this.workerNodes = []
  19       this.promiseResponseMap.clear()
  20     }
  21   }
  22   class StubPoolWithIsMain extends FixedThreadPool {
  23     isMain () {
  24       return false
  25     }
  26   }
  27
  28   it('Simulate pool creation from a non main thread/process', () => {
  29     expect(
  30       () =>
  31         new StubPoolWithIsMain(
  32           numberOfWorkers,
  33           './tests/worker-files/thread/testWorker.js',
  34           {
  35             errorHandler: e => console.error(e)
  36           }
  37         )
  38     ).toThrowError('Cannot start a pool from a worker!')
  39   })
  40
  41   it('Verify that filePath is checked', () => {
  42     const expectedError = new Error(
  43       'Please specify a file with a worker implementation'
  44     )
  45     expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError(
  46       expectedError
  47     )
  48     expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError(
  49       expectedError
  50     )
  51   })
  52
  53   it('Verify that numberOfWorkers is checked', () => {
  54     expect(() => new FixedThreadPool()).toThrowError(
  55       'Cannot instantiate a pool without specifying the number of workers'
  56     )
  57   })
  58
  59   it('Verify that a negative number of workers is checked', () => {
  60     expect(
  61       () =>
  62         new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
  63     ).toThrowError(
  64       new RangeError(
  65         'Cannot instantiate a pool with a negative number of workers'
  66       )
  67     )
  68   })
  69
  70   it('Verify that a non integer number of workers is checked', () => {
  71     expect(
  72       () =>
  73         new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.js')
  74     ).toThrowError(
  75       new TypeError(
  76         'Cannot instantiate a pool with a non safe integer number of workers'
  77       )
  78     )
  79   })
  80
  81   it('Verify that pool options are checked', async () => {
  82     let pool = new FixedThreadPool(
  83       numberOfWorkers,
  84       './tests/worker-files/thread/testWorker.js'
  85     )
  86     expect(pool.emitter).toBeDefined()
  87     expect(pool.opts.enableEvents).toBe(true)
  88     expect(pool.opts.restartWorkerOnError).toBe(true)
  89     expect(pool.opts.enableTasksQueue).toBe(false)
  90     expect(pool.opts.tasksQueueOptions).toBeUndefined()
  91     expect(pool.opts.workerChoiceStrategy).toBe(
  92       WorkerChoiceStrategies.ROUND_ROBIN
  93     )
  94     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
  95       medRunTime: false,
  96       medWaitTime: false
  97     })
  98     expect(pool.opts.messageHandler).toBeUndefined()
  99     expect(pool.opts.errorHandler).toBeUndefined()
 100     expect(pool.opts.onlineHandler).toBeUndefined()
 101     expect(pool.opts.exitHandler).toBeUndefined()
 102     await pool.destroy()
 103     const testHandler = () => console.log('test handler executed')
 104     pool = new FixedThreadPool(
 105       numberOfWorkers,
 106       './tests/worker-files/thread/testWorker.js',
 107       {
 108         workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
 109         workerChoiceStrategyOptions: {
 110           medRunTime: true,
 111           weights: { 0: 300, 1: 200 }
 112         },
 113         enableEvents: false,
 114         restartWorkerOnError: false,
 115         enableTasksQueue: true,
 116         tasksQueueOptions: { concurrency: 2 },
 117         messageHandler: testHandler,
 118         errorHandler: testHandler,
 119         onlineHandler: testHandler,
 120         exitHandler: testHandler
 121       }
 122     )
 123     expect(pool.emitter).toBeUndefined()
 124     expect(pool.opts.enableEvents).toBe(false)
 125     expect(pool.opts.restartWorkerOnError).toBe(false)
 126     expect(pool.opts.enableTasksQueue).toBe(true)
 127     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 128     expect(pool.opts.workerChoiceStrategy).toBe(
 129       WorkerChoiceStrategies.LEAST_USED
 130     )
 131     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 132       medRunTime: true,
 133       weights: { 0: 300, 1: 200 }
 134     })
 135     expect(pool.opts.messageHandler).toStrictEqual(testHandler)
 136     expect(pool.opts.errorHandler).toStrictEqual(testHandler)
 137     expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
 138     expect(pool.opts.exitHandler).toStrictEqual(testHandler)
 139     await pool.destroy()
 140   })
 141
 142   it('Verify that pool options are validated', async () => {
 143     expect(
 144       () =>
 145         new FixedThreadPool(
 146           numberOfWorkers,
 147           './tests/worker-files/thread/testWorker.js',
 148           {
 149             enableTasksQueue: true,
 150             tasksQueueOptions: { concurrency: 0 }
 151           }
 152         )
 153     ).toThrowError("Invalid worker tasks concurrency '0'")
 154     expect(
 155       () =>
 156         new FixedThreadPool(
 157           numberOfWorkers,
 158           './tests/worker-files/thread/testWorker.js',
 159           {
 160             workerChoiceStrategy: 'invalidStrategy'
 161           }
 162         )
 163     ).toThrowError("Invalid worker choice strategy 'invalidStrategy'")
 164     expect(
 165       () =>
 166         new FixedThreadPool(
 167           numberOfWorkers,
 168           './tests/worker-files/thread/testWorker.js',
 169           {
 170             workerChoiceStrategyOptions: { weights: {} }
 171           }
 172         )
 173     ).toThrowError(
 174       'Invalid worker choice strategy options: must have a weight for each worker node'
 175     )
 176   })
 177
 178   it('Verify that worker choice strategy options can be set', async () => {
 179     const pool = new FixedThreadPool(
 180       numberOfWorkers,
 181       './tests/worker-files/thread/testWorker.js',
 182       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
 183     )
 184     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 185       medRunTime: false,
 186       medWaitTime: false
 187     })
 188     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 189       .workerChoiceStrategies) {
 190       expect(workerChoiceStrategy.opts).toStrictEqual({
 191         medRunTime: false,
 192         medWaitTime: false
 193       })
 194     }
 195     expect(
 196       pool.workerChoiceStrategyContext.getRequiredStatistics()
 197     ).toStrictEqual({
 198       runTime: true,
 199       avgRunTime: true,
 200       medRunTime: false,
 201       waitTime: false,
 202       avgWaitTime: false,
 203       medWaitTime: false
 204     })
 205     pool.setWorkerChoiceStrategyOptions({ medRunTime: true })
 206     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 207       medRunTime: true
 208     })
 209     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 210       .workerChoiceStrategies) {
 211       expect(workerChoiceStrategy.opts).toStrictEqual({ medRunTime: true })
 212     }
 213     expect(
 214       pool.workerChoiceStrategyContext.getRequiredStatistics()
 215     ).toStrictEqual({
 216       runTime: true,
 217       avgRunTime: false,
 218       medRunTime: true,
 219       waitTime: false,
 220       avgWaitTime: false,
 221       medWaitTime: false
 222     })
 223     pool.setWorkerChoiceStrategyOptions({ medRunTime: false })
 224     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
 225       medRunTime: false
 226     })
 227     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
 228       .workerChoiceStrategies) {
 229       expect(workerChoiceStrategy.opts).toStrictEqual({ medRunTime: false })
 230     }
 231     expect(
 232       pool.workerChoiceStrategyContext.getRequiredStatistics()
 233     ).toStrictEqual({
 234       runTime: true,
 235       avgRunTime: true,
 236       medRunTime: false,
 237       waitTime: false,
 238       avgWaitTime: false,
 239       medWaitTime: false
 240     })
 241     await pool.destroy()
 242   })
 243
 244   it('Verify that tasks queue can be enabled/disabled', async () => {
 245     const pool = new FixedThreadPool(
 246       numberOfWorkers,
 247       './tests/worker-files/thread/testWorker.js'
 248     )
 249     expect(pool.opts.enableTasksQueue).toBe(false)
 250     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 251     pool.enableTasksQueue(true)
 252     expect(pool.opts.enableTasksQueue).toBe(true)
 253     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 254     pool.enableTasksQueue(true, { concurrency: 2 })
 255     expect(pool.opts.enableTasksQueue).toBe(true)
 256     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 257     pool.enableTasksQueue(false)
 258     expect(pool.opts.enableTasksQueue).toBe(false)
 259     expect(pool.opts.tasksQueueOptions).toBeUndefined()
 260     await pool.destroy()
 261   })
 262
 263   it('Verify that tasks queue options can be set', async () => {
 264     const pool = new FixedThreadPool(
 265       numberOfWorkers,
 266       './tests/worker-files/thread/testWorker.js',
 267       { enableTasksQueue: true }
 268     )
 269     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
 270     pool.setTasksQueueOptions({ concurrency: 2 })
 271     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
 272     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
 273       "Invalid worker tasks concurrency '0'"
 274     )
 275     await pool.destroy()
 276   })
 277
 278   it('Verify that pool info is set', async () => {
 279     let pool = new FixedThreadPool(
 280       numberOfWorkers,
 281       './tests/worker-files/thread/testWorker.js'
 282     )
 283     expect(pool.info).toStrictEqual({
 284       type: PoolTypes.fixed,
 285       minSize: numberOfWorkers,
 286       maxSize: numberOfWorkers,
 287       workerNodes: numberOfWorkers,
 288       idleWorkerNodes: numberOfWorkers,
 289       busyWorkerNodes: 0,
 290       runningTasks: 0,
 291       queuedTasks: 0,
 292       maxQueuedTasks: 0
 293     })
 294     await pool.destroy()
 295     pool = new DynamicClusterPool(
 296       numberOfWorkers,
 297       numberOfWorkers * 2,
 298       './tests/worker-files/thread/testWorker.js'
 299     )
 300     expect(pool.info).toStrictEqual({
 301       type: PoolTypes.dynamic,
 302       minSize: numberOfWorkers,
 303       maxSize: numberOfWorkers * 2,
 304       workerNodes: numberOfWorkers,
 305       idleWorkerNodes: numberOfWorkers,
 306       busyWorkerNodes: 0,
 307       runningTasks: 0,
 308       queuedTasks: 0,
 309       maxQueuedTasks: 0
 310     })
 311     await pool.destroy()
 312   })
 313
 314   it('Simulate worker not found', async () => {
 315     const pool = new StubPoolWithRemoveAllWorker(
 316       numberOfWorkers,
 317       './tests/worker-files/cluster/testWorker.js',
 318       {
 319         errorHandler: e => console.error(e)
 320       }
 321     )
 322     expect(pool.workerNodes.length).toBe(numberOfWorkers)
 323     // Simulate worker not found.
 324     pool.removeAllWorker()
 325     expect(pool.workerNodes.length).toBe(0)
 326     await pool.destroy()
 327   })
 328
 329   it('Verify that worker pool tasks usage are initialized', async () => {
 330     const pool = new FixedClusterPool(
 331       numberOfWorkers,
 332       './tests/worker-files/cluster/testWorker.js'
 333     )
 334     for (const workerNode of pool.workerNodes) {
 335       expect(workerNode.tasksUsage).toStrictEqual({
 336         run: 0,
 337         running: 0,
 338         runTime: 0,
 339         runTimeHistory: expect.any(CircularArray),
 340         avgRunTime: 0,
 341         medRunTime: 0,
 342         waitTime: 0,
 343         waitTimeHistory: expect.any(CircularArray),
 344         avgWaitTime: 0,
 345         medWaitTime: 0,
 346         error: 0
 347       })
 348     }
 349     await pool.destroy()
 350   })
 351
 352   it('Verify that worker pool tasks queue are initialized', async () => {
 353     const pool = new FixedClusterPool(
 354       numberOfWorkers,
 355       './tests/worker-files/cluster/testWorker.js'
 356     )
 357     for (const workerNode of pool.workerNodes) {
 358       expect(workerNode.tasksQueue).toBeDefined()
 359       expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
 360       expect(workerNode.tasksQueue.size).toBe(0)
 361     }
 362     await pool.destroy()
 363   })
 364
 365   it('Verify that worker pool tasks usage are computed', async () => {
 366     const pool = new FixedClusterPool(
 367       numberOfWorkers,
 368       './tests/worker-files/cluster/testWorker.js'
 369     )
 370     const promises = new Set()
 371     const maxMultiplier = 2
 372     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 373       promises.add(pool.execute())
 374     }
 375     for (const workerNode of pool.workerNodes) {
 376       expect(workerNode.tasksUsage).toStrictEqual({
 377         run: 0,
 378         running: maxMultiplier,
 379         runTime: 0,
 380         runTimeHistory: expect.any(CircularArray),
 381         avgRunTime: 0,
 382         medRunTime: 0,
 383         waitTime: 0,
 384         waitTimeHistory: expect.any(CircularArray),
 385         avgWaitTime: 0,
 386         medWaitTime: 0,
 387         error: 0
 388       })
 389     }
 390     await Promise.all(promises)
 391     for (const workerNode of pool.workerNodes) {
 392       expect(workerNode.tasksUsage).toStrictEqual({
 393         run: maxMultiplier,
 394         running: 0,
 395         runTime: 0,
 396         runTimeHistory: expect.any(CircularArray),
 397         avgRunTime: 0,
 398         medRunTime: 0,
 399         waitTime: 0,
 400         waitTimeHistory: expect.any(CircularArray),
 401         avgWaitTime: 0,
 402         medWaitTime: 0,
 403         error: 0
 404       })
 405     }
 406     await pool.destroy()
 407   })
 408
 409   it('Verify that worker pool tasks usage are reset at worker choice strategy change', async () => {
 410     const pool = new DynamicThreadPool(
 411       numberOfWorkers,
 412       numberOfWorkers,
 413       './tests/worker-files/thread/testWorker.js'
 414     )
 415     const promises = new Set()
 416     const maxMultiplier = 2
 417     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 418       promises.add(pool.execute())
 419     }
 420     await Promise.all(promises)
 421     for (const workerNode of pool.workerNodes) {
 422       expect(workerNode.tasksUsage).toStrictEqual({
 423         run: expect.any(Number),
 424         running: 0,
 425         runTime: 0,
 426         runTimeHistory: expect.any(CircularArray),
 427         avgRunTime: 0,
 428         medRunTime: 0,
 429         waitTime: 0,
 430         waitTimeHistory: expect.any(CircularArray),
 431         avgWaitTime: 0,
 432         medWaitTime: 0,
 433         error: 0
 434       })
 435       expect(workerNode.tasksUsage.run).toBeGreaterThan(0)
 436       expect(workerNode.tasksUsage.run).toBeLessThanOrEqual(maxMultiplier)
 437     }
 438     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 439     for (const workerNode of pool.workerNodes) {
 440       expect(workerNode.tasksUsage).toStrictEqual({
 441         run: 0,
 442         running: 0,
 443         runTime: 0,
 444         runTimeHistory: expect.any(CircularArray),
 445         avgRunTime: 0,
 446         medRunTime: 0,
 447         waitTime: 0,
 448         waitTimeHistory: expect.any(CircularArray),
 449         avgWaitTime: 0,
 450         medWaitTime: 0,
 451         error: 0
 452       })
 453       expect(workerNode.tasksUsage.runTimeHistory.length).toBe(0)
 454       expect(workerNode.tasksUsage.waitTimeHistory.length).toBe(0)
 455     }
 456     await pool.destroy()
 457   })
 458
 459   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 460     const pool = new DynamicThreadPool(
 461       numberOfWorkers,
 462       numberOfWorkers,
 463       './tests/worker-files/thread/testWorker.js'
 464     )
 465     const promises = new Set()
 466     let poolFull = 0
 467     pool.emitter.on(PoolEvents.full, () => ++poolFull)
 468     for (let i = 0; i < numberOfWorkers * 2; i++) {
 469       promises.add(pool.execute())
 470     }
 471     await Promise.all(promises)
 472     // The `full` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool.
 473     // So in total numberOfWorkers * 2 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool with min = max = numberOfWorkers.
 474     expect(poolFull).toBe(numberOfWorkers * 2)
 475     await pool.destroy()
 476   })
 477
 478   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 479     const pool = new FixedThreadPool(
 480       numberOfWorkers,
 481       './tests/worker-files/thread/testWorker.js'
 482     )
 483     const promises = new Set()
 484     let poolBusy = 0
 485     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
 486     for (let i = 0; i < numberOfWorkers * 2; i++) {
 487       promises.add(pool.execute())
 488     }
 489     await Promise.all(promises)
 490     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 491     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 492     expect(poolBusy).toBe(numberOfWorkers + 1)
 493     await pool.destroy()
 494   })
 495
 496   it('Verify that multiple tasks worker is working', async () => {
 497     const pool = new DynamicClusterPool(
 498       numberOfWorkers,
 499       numberOfWorkers * 2,
 500       './tests/worker-files/cluster/testMultiTasksWorker.js'
 501     )
 502     const data = { n: 10 }
 503     const result0 = await pool.execute(data)
 504     expect(result0).toBe(false)
 505     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
 506     expect(result1).toBe(false)
 507     const result2 = await pool.execute(data, 'factorial')
 508     expect(result2).toBe(3628800)
 509     const result3 = await pool.execute(data, 'fibonacci')
 510     expect(result3).toBe(89)
 511   })
 512 })