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