tests/pools/cluster/dynamic.test.js

   1 const { expect } = require('expect')
   2 const { DynamicClusterPool } = require('../../../lib/index')
   3 const { WorkerFunctions } = require('../../test-types')
   4 const TestUtils = require('../../test-utils')
   5
   6 describe('Dynamic cluster pool test suite', () => {
   7   const min = 1
   8   const max = 3
   9   const pool = new DynamicClusterPool(
  10     min,
  11     max,
  12     './tests/worker-files/cluster/testWorker.js',
  13     {
  14       errorHandler: e => console.error(e)
  15     }
  16   )
  17
  18   it('Verify that the function is executed in a worker cluster', async () => {
  19     let result = await pool.execute({
  20       function: WorkerFunctions.fibonacci
  21     })
  22     expect(result).toBe(false)
  23     result = await pool.execute({
  24       function: WorkerFunctions.factorial
  25     })
  26     expect(result).toBe(false)
  27   })
  28
  29   it('Verify that new workers are created when required, max size is not exceeded and that after a while new workers will die', async () => {
  30     const promises = []
  31     let poolBusy = 0
  32     pool.emitter.on('busy', () => poolBusy++)
  33     for (let i = 0; i < max * 2; i++) {
  34       promises.push(pool.execute())
  35     }
  36     expect(pool.workers.length).toBeLessThanOrEqual(max)
  37     expect(pool.workers.length).toBeGreaterThan(min)
  38     // The `busy` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool.
  39     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool.
  40     expect(poolBusy).toBe(max + 1)
  41     const numberOfExitEvents = await TestUtils.waitExits(pool, max - min)
  42     expect(numberOfExitEvents).toBe(max - min)
  43   })
  44
  45   it('Verify scale worker up and down is working', async () => {
  46     expect(pool.workers.length).toBe(min)
  47     for (let i = 0; i < max * 10; i++) {
  48       pool.execute()
  49     }
  50     expect(pool.workers.length).toBeGreaterThan(min)
  51     await TestUtils.waitExits(pool, max - min)
  52     expect(pool.workers.length).toBe(min)
  53     for (let i = 0; i < max * 10; i++) {
  54       pool.execute()
  55     }
  56     expect(pool.workers.length).toBeGreaterThan(min)
  57     await TestUtils.waitExits(pool, max - min)
  58     expect(pool.workers.length).toBe(min)
  59   })
  60
  61   it('Shutdown test', async () => {
  62     const exitPromise = TestUtils.waitExits(pool, min)
  63     await pool.destroy()
  64     const numberOfExitEvents = await exitPromise
  65     expect(numberOfExitEvents).toBe(min)
  66   })
  67
  68   it('Validation of inputs test', () => {
  69     expect(() => new DynamicClusterPool(min)).toThrowError(
  70       new Error('Please specify a file with a worker implementation')
  71     )
  72   })
  73
  74   it('Should work even without opts in input', async () => {
  75     const pool1 = new DynamicClusterPool(
  76       min,
  77       max,
  78       './tests/worker-files/cluster/testWorker.js'
  79     )
  80     const result = await pool1.execute()
  81     expect(result).toBe(false)
  82     // We need to clean up the resources after our test
  83     await pool1.destroy()
  84   })
  85
  86   it('Verify scale processes up and down is working when long running task is used:hard', async () => {
  87     const longRunningPool = new DynamicClusterPool(
  88       min,
  89       max,
  90       './tests/worker-files/cluster/longRunningWorkerHardBehavior.js',
  91       {
  92         errorHandler: e => console.error(e),
  93         onlineHandler: () => console.log('long running worker is online'),
  94         exitHandler: () => console.log('long running worker exited')
  95       }
  96     )
  97     expect(longRunningPool.workers.length).toBe(min)
  98     for (let i = 0; i < max * 10; i++) {
  99       longRunningPool.execute()
 100     }
 101     expect(longRunningPool.workers.length).toBe(max)
 102     await TestUtils.waitExits(longRunningPool, max - min)
 103     // Here we expect the workers to be at the max size since that the task is still running
 104     expect(longRunningPool.workers.length).toBe(min)
 105     // We need to clean up the resources after our test
 106     await longRunningPool.destroy()
 107   })
 108
 109   it('Verify scale processes up and down is working when long running task is used:soft', async () => {
 110     const longRunningPool = new DynamicClusterPool(
 111       min,
 112       max,
 113       './tests/worker-files/cluster/longRunningWorkerSoftBehavior.js',
 114       {
 115         errorHandler: e => console.error(e),
 116         onlineHandler: () => console.log('long running worker is online'),
 117         exitHandler: () => console.log('long running worker exited')
 118       }
 119     )
 120     expect(longRunningPool.workers.length).toBe(min)
 121     for (let i = 0; i < max * 10; i++) {
 122       longRunningPool.execute()
 123     }
 124     expect(longRunningPool.workers.length).toBe(max)
 125     await TestUtils.sleep(1500)
 126     // Here we expect the workers to be at the max size since that the task is still running
 127     expect(longRunningPool.workers.length).toBe(max)
 128     // We need to clean up the resources after our test
 129     await longRunningPool.destroy()
 130   })
 131
 132   it('Verify that a pool with zero worker can be instantiated', async () => {
 133     const pool = new DynamicClusterPool(
 134       0,
 135       max,
 136       './tests/worker-files/cluster/testWorker.js'
 137     )
 138     expect(pool).toBeInstanceOf(DynamicClusterPool)
 139     // We need to clean up the resources after our test
 140     await pool.destroy()
 141   })
 142 })