tests/pools/cluster/dynamic.test.js

   1 const { expect } = require('expect')
   2 const { DynamicClusterPool, PoolEvents } = require('../../../lib')
   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(121393)
  23     result = await pool.execute({
  24       function: WorkerFunctions.factorial
  25     })
  26     expect(result).toBe(9.33262154439441e157)
  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     let poolBusy = 0
  31     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
  32     for (let i = 0; i < max * 2; i++) {
  33       pool.execute()
  34     }
  35     expect(pool.workerNodes.length).toBeLessThanOrEqual(max)
  36     expect(pool.workerNodes.length).toBeGreaterThan(min)
  37     // The `busy` event is triggered when the number of submitted tasks at once reach the max number of workers in the dynamic pool.
  38     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool.
  39     expect(poolBusy).toBe(max + 1)
  40     const numberOfExitEvents = await TestUtils.waitWorkerEvents(
  41       pool,
  42       'exit',
  43       max - min
  44     )
  45     expect(numberOfExitEvents).toBe(max - min)
  46   })
  47
  48   it('Verify scale worker up and down is working', async () => {
  49     expect(pool.workerNodes.length).toBe(min)
  50     for (let i = 0; i < max * 2; i++) {
  51       pool.execute()
  52     }
  53     expect(pool.workerNodes.length).toBeGreaterThan(min)
  54     await TestUtils.waitWorkerEvents(pool, 'exit', max - min)
  55     expect(pool.workerNodes.length).toBe(min)
  56     for (let i = 0; i < max * 2; i++) {
  57       pool.execute()
  58     }
  59     expect(pool.workerNodes.length).toBeGreaterThan(min)
  60     await TestUtils.waitWorkerEvents(pool, 'exit', max - min)
  61     expect(pool.workerNodes.length).toBe(min)
  62   })
  63
  64   it('Shutdown test', async () => {
  65     const exitPromise = TestUtils.waitWorkerEvents(pool, 'exit', min)
  66     await pool.destroy()
  67     const numberOfExitEvents = await exitPromise
  68     expect(numberOfExitEvents).toBe(min)
  69   })
  70
  71   it('Validation of inputs test', () => {
  72     expect(() => new DynamicClusterPool(min)).toThrowError(
  73       'Please specify a file with a worker implementation'
  74     )
  75   })
  76
  77   it('Should work even without opts in input', async () => {
  78     const pool1 = new DynamicClusterPool(
  79       min,
  80       max,
  81       './tests/worker-files/cluster/testWorker.js'
  82     )
  83     const result = await pool1.execute()
  84     expect(result).toBe(false)
  85     // We need to clean up the resources after our test
  86     await pool1.destroy()
  87   })
  88
  89   it('Verify scale processes up and down is working when long executing task is used:hard', async () => {
  90     const longRunningPool = new DynamicClusterPool(
  91       min,
  92       max,
  93       './tests/worker-files/cluster/longRunningWorkerHardBehavior.js',
  94       {
  95         errorHandler: e => console.error(e),
  96         onlineHandler: () => console.log('long executing worker is online'),
  97         exitHandler: () => console.log('long executing worker exited')
  98       }
  99     )
 100     expect(longRunningPool.workerNodes.length).toBe(min)
 101     for (let i = 0; i < max * 2; i++) {
 102       longRunningPool.execute()
 103     }
 104     expect(longRunningPool.workerNodes.length).toBe(max)
 105     await TestUtils.waitWorkerEvents(longRunningPool, 'exit', max - min)
 106     expect(longRunningPool.workerNodes.length).toBe(min)
 107     expect(
 108       longRunningPool.workerChoiceStrategyContext.workerChoiceStrategies.get(
 109         longRunningPool.workerChoiceStrategyContext.workerChoiceStrategy
 110       ).nextWorkerNodeId
 111     ).toBeLessThan(longRunningPool.workerNodes.length)
 112     // We need to clean up the resources after our test
 113     await longRunningPool.destroy()
 114   })
 115
 116   it('Verify scale processes up and down is working when long executing task is used:soft', async () => {
 117     const longRunningPool = new DynamicClusterPool(
 118       min,
 119       max,
 120       './tests/worker-files/cluster/longRunningWorkerSoftBehavior.js',
 121       {
 122         errorHandler: e => console.error(e),
 123         onlineHandler: () => console.log('long executing worker is online'),
 124         exitHandler: () => console.log('long executing worker exited')
 125       }
 126     )
 127     expect(longRunningPool.workerNodes.length).toBe(min)
 128     for (let i = 0; i < max * 2; i++) {
 129       longRunningPool.execute()
 130     }
 131     expect(longRunningPool.workerNodes.length).toBe(max)
 132     await TestUtils.sleep(1500)
 133     // Here we expect the workerNodes to be at the max size since the task is still executing
 134     expect(longRunningPool.workerNodes.length).toBe(max)
 135     // We need to clean up the resources after our test
 136     await longRunningPool.destroy()
 137   })
 138
 139   it('Verify that a pool with zero worker can be instantiated', async () => {
 140     const pool = new DynamicClusterPool(
 141       0,
 142       max,
 143       './tests/worker-files/cluster/testWorker.js'
 144     )
 145     expect(pool).toBeInstanceOf(DynamicClusterPool)
 146     // We need to clean up the resources after our test
 147     await pool.destroy()
 148   })
 149 })