tests/pools/cluster/dynamic.test.js

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