tests/pools/cluster/dynamic.test.js

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