tests/pools/cluster/dynamic.test.js

   1 const { expect } = require('expect')
   2 const { DynamicClusterPool, PoolEvents } = 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     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.workers.length).toBeLessThanOrEqual(max)
  36     expect(pool.workers.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.waitExits(pool, max - min)
  41     expect(numberOfExitEvents).toBe(max - min)
  42   })
  43
  44   it('Verify scale worker up and down is working', async () => {
  45     expect(pool.workers.length).toBe(min)
  46     for (let i = 0; i < max * 10; i++) {
  47       pool.execute()
  48     }
  49     expect(pool.workers.length).toBeGreaterThan(min)
  50     await TestUtils.waitExits(pool, max - min)
  51     expect(pool.workers.length).toBe(min)
  52     for (let i = 0; i < max * 10; i++) {
  53       pool.execute()
  54     }
  55     expect(pool.workers.length).toBeGreaterThan(min)
  56     await TestUtils.waitExits(pool, max - min)
  57     expect(pool.workers.length).toBe(min)
  58   })
  59
  60   it('Shutdown test', async () => {
  61     const exitPromise = TestUtils.waitExits(pool, min)
  62     await pool.destroy()
  63     const numberOfExitEvents = await exitPromise
  64     expect(numberOfExitEvents).toBe(min)
  65   })
  66
  67   it('Validation of inputs test', () => {
  68     expect(() => new DynamicClusterPool(min)).toThrowError(
  69       new Error('Please specify a file with a worker implementation')
  70     )
  71   })
  72
  73   it('Should work even without opts in input', async () => {
  74     const pool1 = new DynamicClusterPool(
  75       min,
  76       max,
  77       './tests/worker-files/cluster/testWorker.js'
  78     )
  79     const result = await pool1.execute()
  80     expect(result).toBe(false)
  81     // We need to clean up the resources after our test
  82     await pool1.destroy()
  83   })
  84
  85   it('Verify scale processes up and down is working when long running task is used:hard', async () => {
  86     const longRunningPool = new DynamicClusterPool(
  87       min,
  88       max,
  89       './tests/worker-files/cluster/longRunningWorkerHardBehavior.js',
  90       {
  91         errorHandler: e => console.error(e),
  92         onlineHandler: () => console.log('long running worker is online'),
  93         exitHandler: () => console.log('long running worker exited')
  94       }
  95     )
  96     expect(longRunningPool.workers.length).toBe(min)
  97     for (let i = 0; i < max * 10; i++) {
  98       longRunningPool.execute()
  99     }
 100     expect(longRunningPool.workers.length).toBe(max)
 101     await TestUtils.waitExits(longRunningPool, max - min)
 102     expect(longRunningPool.workers.length).toBe(min)
 103     // We need to clean up the resources after our test
 104     await longRunningPool.destroy()
 105   })
 106
 107   it('Verify scale processes up and down is working when long running task is used:soft', async () => {
 108     const longRunningPool = new DynamicClusterPool(
 109       min,
 110       max,
 111       './tests/worker-files/cluster/longRunningWorkerSoftBehavior.js',
 112       {
 113         errorHandler: e => console.error(e),
 114         onlineHandler: () => console.log('long running worker is online'),
 115         exitHandler: () => console.log('long running worker exited')
 116       }
 117     )
 118     expect(longRunningPool.workers.length).toBe(min)
 119     for (let i = 0; i < max * 10; i++) {
 120       longRunningPool.execute()
 121     }
 122     expect(longRunningPool.workers.length).toBe(max)
 123     await TestUtils.sleep(1500)
 124     // Here we expect the workers to be at the max size since the task is still running
 125     expect(longRunningPool.workers.length).toBe(max)
 126     // We need to clean up the resources after our test
 127     await longRunningPool.destroy()
 128   })
 129
 130   it('Verify that a pool with zero worker can be instantiated', async () => {
 131     const pool = new DynamicClusterPool(
 132       0,
 133       max,
 134       './tests/worker-files/cluster/testWorker.js'
 135     )
 136     expect(pool).toBeInstanceOf(DynamicClusterPool)
 137     // We need to clean up the resources after our test
 138     await pool.destroy()
 139   })
 140 })