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 = 1
  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 }
 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 }
 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     for (let i = 0; i < numberOfWorkers * 2; i++) {
 332       promises.push(pool.execute())
 333     }
 334     for (const workerNode of pool.workerNodes) {
 335       expect(workerNode.tasksUsage).toStrictEqual({
 336         run: 0,
 337         running: numberOfWorkers * 2,
 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 Promise.all(promises)
 350     for (const workerNode of pool.workerNodes) {
 351       expect(workerNode.tasksUsage).toStrictEqual({
 352         run: numberOfWorkers * 2,
 353         running: 0,
 354         runTime: 0,
 355         runTimeHistory: expect.any(CircularArray),
 356         avgRunTime: 0,
 357         medRunTime: 0,
 358         waitTime: 0,
 359         waitTimeHistory: expect.any(CircularArray),
 360         avgWaitTime: 0,
 361         medWaitTime: 0,
 362         error: 0
 363       })
 364     }
 365     await pool.destroy()
 366   })
 367
 368   it('Verify that worker pool tasks usage are reset at worker choice strategy change', async () => {
 369     const pool = new DynamicThreadPool(
 370       numberOfWorkers,
 371       numberOfWorkers,
 372       './tests/worker-files/thread/testWorker.js'
 373     )
 374     const promises = []
 375     const maxMultiplier = 2
 376     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 377       promises.push(pool.execute())
 378     }
 379     await Promise.all(promises)
 380     for (const workerNode of pool.workerNodes) {
 381       expect(workerNode.tasksUsage).toStrictEqual({
 382         run: numberOfWorkers * maxMultiplier,
 383         running: 0,
 384         runTime: 0,
 385         runTimeHistory: expect.any(CircularArray),
 386         avgRunTime: 0,
 387         medRunTime: 0,
 388         waitTime: 0,
 389         waitTimeHistory: expect.any(CircularArray),
 390         avgWaitTime: 0,
 391         medWaitTime: 0,
 392         error: 0
 393       })
 394     }
 395     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 396     for (const workerNode of pool.workerNodes) {
 397       expect(workerNode.tasksUsage).toStrictEqual({
 398         run: 0,
 399         running: 0,
 400         runTime: 0,
 401         runTimeHistory: expect.any(CircularArray),
 402         avgRunTime: 0,
 403         medRunTime: 0,
 404         waitTime: 0,
 405         waitTimeHistory: expect.any(CircularArray),
 406         avgWaitTime: 0,
 407         medWaitTime: 0,
 408         error: 0
 409       })
 410       expect(workerNode.tasksUsage.runTimeHistory.length).toBe(0)
 411       expect(workerNode.tasksUsage.waitTimeHistory.length).toBe(0)
 412     }
 413     await pool.destroy()
 414   })
 415
 416   it("Verify that pool event emitter 'full' event can register a callback", async () => {
 417     const pool = new DynamicThreadPool(
 418       numberOfWorkers,
 419       numberOfWorkers,
 420       './tests/worker-files/thread/testWorker.js'
 421     )
 422     const promises = []
 423     let poolFull = 0
 424     pool.emitter.on(PoolEvents.full, () => ++poolFull)
 425     for (let i = 0; i < numberOfWorkers * 2; i++) {
 426       promises.push(pool.execute())
 427     }
 428     await Promise.all(promises)
 429     // The `full` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool.
 430     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool.
 431     expect(poolFull).toBe(numberOfWorkers + 1)
 432     await pool.destroy()
 433   })
 434
 435   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 436     const pool = new FixedThreadPool(
 437       numberOfWorkers,
 438       './tests/worker-files/thread/testWorker.js'
 439     )
 440     const promises = []
 441     let poolBusy = 0
 442     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
 443     for (let i = 0; i < numberOfWorkers * 2; i++) {
 444       promises.push(pool.execute())
 445     }
 446     await Promise.all(promises)
 447     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 448     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 449     expect(poolBusy).toBe(numberOfWorkers + 1)
 450     await pool.destroy()
 451   })
 452
 453   it('Verify that multiple tasks worker is working', async () => {
 454     const pool = new DynamicClusterPool(
 455       numberOfWorkers,
 456       numberOfWorkers * 2,
 457       './tests/worker-files/cluster/testMultiTasksWorker.js'
 458     )
 459     const data = { n: 10 }
 460     const result0 = await pool.execute(data)
 461     expect(result0).toBe(false)
 462     const result1 = await pool.execute(data, 'jsonIntegerSerialization')
 463     expect(result1).toBe(false)
 464     const result2 = await pool.execute(data, 'factorial')
 465     expect(result2).toBe(3628800)
 466     const result3 = await pool.execute(data, 'fibonacci')
 467     expect(result3).toBe(89)
 468   })
 469 })