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