tests/pools/cluster/dynamic.test.mjs

   1 import { expect } from 'expect'
   2
   3 import {
   4   DynamicClusterPool,
   5   PoolEvents,
   6   WorkerChoiceStrategies
   7 } from '../../../lib/index.cjs'
   8 import { TaskFunctions } from '../../test-types.cjs'
   9 import { sleep, waitPoolEvents, waitWorkerEvents } from '../../test-utils.cjs'
  10
  11 describe('Dynamic cluster pool test suite', () => {
  12   const min = 1
  13   const max = 3
  14   const pool = new DynamicClusterPool(
  15     min,
  16     max,
  17     './tests/worker-files/cluster/testWorker.cjs',
  18     {
  19       errorHandler: e => console.error(e)
  20     }
  21   )
  22
  23   it('Verify that the function is executed in a worker cluster', async () => {
  24     let result = await pool.execute({
  25       function: TaskFunctions.fibonacci
  26     })
  27     expect(result).toBe(354224848179262000000)
  28     result = await pool.execute({
  29       function: TaskFunctions.factorial
  30     })
  31     expect(result).toBe(9.33262154439441e157)
  32   })
  33
  34   it('Verify that new workers are created when required, max size is not exceeded and that after a while new workers will die', async () => {
  35     let poolBusy = 0
  36     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
  37     for (let i = 0; i < max * 2; i++) {
  38       pool.execute()
  39     }
  40     expect(pool.workerNodes.length).toBeLessThanOrEqual(max)
  41     expect(pool.workerNodes.length).toBeGreaterThan(min)
  42     expect(poolBusy).toBe(1)
  43     const numberOfExitEvents = await waitWorkerEvents(pool, 'exit', max - min)
  44     expect(numberOfExitEvents).toBe(max - min)
  45     expect(pool.workerNodes.length).toBe(min)
  46   })
  47
  48   it('Verify scale worker up and down is working', async () => {
  49     for (let i = 0; i < max * 2; i++) {
  50       pool.execute()
  51     }
  52     expect(pool.workerNodes.length).toBeGreaterThan(min)
  53     await waitWorkerEvents(pool, 'exit', max - min)
  54     expect(pool.workerNodes.length).toBe(min)
  55     for (let i = 0; i < max * 2; i++) {
  56       pool.execute()
  57     }
  58     expect(pool.workerNodes.length).toBeGreaterThan(min)
  59     await waitWorkerEvents(pool, 'exit', max - min)
  60     expect(pool.workerNodes.length).toBe(min)
  61   })
  62
  63   it('Shutdown test', async () => {
  64     const exitPromise = waitWorkerEvents(pool, 'exit', min)
  65     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
  66     let poolDestroy = 0
  67     pool.emitter.on(PoolEvents.destroy, () => ++poolDestroy)
  68     expect(pool.emitter.eventNames()).toStrictEqual([
  69       PoolEvents.busy,
  70       PoolEvents.destroy
  71     ])
  72     await pool.destroy()
  73     const numberOfExitEvents = await exitPromise
  74     expect(pool.started).toBe(false)
  75     expect(pool.emitter.eventNames()).toStrictEqual([
  76       PoolEvents.busy,
  77       PoolEvents.destroy
  78     ])
  79     expect(pool.readyEventEmitted).toBe(false)
  80     expect(pool.workerNodes.length).toBe(0)
  81     expect(numberOfExitEvents).toBe(min)
  82     expect(poolDestroy).toBe(1)
  83   })
  84
  85   it('Validation of inputs test', () => {
  86     expect(() => new DynamicClusterPool(min)).toThrow(
  87       'The worker file path must be specified'
  88     )
  89   })
  90
  91   it('Should work even without opts in input', async () => {
  92     const pool = new DynamicClusterPool(
  93       min,
  94       max,
  95       './tests/worker-files/cluster/testWorker.cjs'
  96     )
  97     const result = await pool.execute()
  98     expect(result).toStrictEqual({ ok: 1 })
  99     // We need to clean up the resources after our test
 100     await pool.destroy()
 101   })
 102
 103   it('Verify scale processes up and down is working when long executing task is used:hard', async () => {
 104     const longRunningPool = new DynamicClusterPool(
 105       min,
 106       max,
 107       './tests/worker-files/cluster/longRunningWorkerHardBehavior.cjs',
 108       {
 109         errorHandler: e => console.error(e),
 110         onlineHandler: () => console.info('long executing worker is online'),
 111         exitHandler: () => console.info('long executing worker exited')
 112       }
 113     )
 114     expect(longRunningPool.workerNodes.length).toBe(min)
 115     for (let i = 0; i < max * 2; i++) {
 116       longRunningPool.execute()
 117     }
 118     expect(longRunningPool.workerNodes.length).toBe(max)
 119     await waitWorkerEvents(longRunningPool, 'exit', max - min)
 120     expect(longRunningPool.workerNodes.length).toBe(min)
 121     expect(
 122       longRunningPool.workerChoiceStrategiesContext.workerChoiceStrategies.get(
 123         longRunningPool.workerChoiceStrategiesContext
 124           .defaultWorkerChoiceStrategy
 125       ).nextWorkerNodeKey
 126     ).toBeLessThan(longRunningPool.workerNodes.length)
 127     // We need to clean up the resources after our test
 128     await longRunningPool.destroy()
 129   })
 130
 131   it('Verify scale processes up and down is working when long executing task is used:soft', async () => {
 132     const longRunningPool = new DynamicClusterPool(
 133       min,
 134       max,
 135       './tests/worker-files/cluster/longRunningWorkerSoftBehavior.cjs',
 136       {
 137         errorHandler: e => console.error(e),
 138         onlineHandler: () => console.info('long executing worker is online'),
 139         exitHandler: () => console.info('long executing worker exited')
 140       }
 141     )
 142     expect(longRunningPool.workerNodes.length).toBe(min)
 143     for (let i = 0; i < max * 2; i++) {
 144       longRunningPool.execute()
 145     }
 146     expect(longRunningPool.workerNodes.length).toBe(max)
 147     await sleep(1000)
 148     // Here we expect the workerNodes to be at the max size since the task is still executing
 149     expect(longRunningPool.workerNodes.length).toBe(max)
 150     // We need to clean up the resources after our test
 151     await longRunningPool.destroy()
 152   })
 153
 154   it('Verify that a pool with zero worker can be instantiated', async () => {
 155     const pool = new DynamicClusterPool(
 156       0,
 157       max,
 158       './tests/worker-files/cluster/testWorker.cjs'
 159     )
 160     expect(pool).toBeInstanceOf(DynamicClusterPool)
 161     // We need to clean up the resources after our test
 162     await pool.destroy()
 163   })
 164
 165   it('Verify that a pool with zero worker works', async () => {
 166     for (const workerChoiceStrategy of Object.values(WorkerChoiceStrategies)) {
 167       const pool = new DynamicClusterPool(
 168         0,
 169         max,
 170         './tests/worker-files/cluster/testWorker.cjs',
 171         {
 172           workerChoiceStrategy
 173         }
 174       )
 175       expect(pool.starting).toBe(false)
 176       expect(pool.readyEventEmitted).toBe(false)
 177       for (let run = 0; run < 2; run++) {
 178         run % 2 !== 0 && pool.enableTasksQueue(true)
 179         const maxMultiplier = 4
 180         const promises = new Set()
 181         expect(pool.workerNodes.length).toBe(pool.info.minSize)
 182         for (let i = 0; i < max * maxMultiplier; i++) {
 183           promises.add(pool.execute())
 184         }
 185         await Promise.all(promises)
 186         expect(pool.readyEventEmitted).toBe(true)
 187         expect(pool.workerNodes.length).toBeGreaterThan(pool.info.minSize)
 188         expect(pool.workerNodes.length).toBeLessThanOrEqual(pool.info.maxSize)
 189         await waitPoolEvents(pool, PoolEvents.empty, 1)
 190         expect(pool.readyEventEmitted).toBe(false)
 191         expect(pool.workerNodes.length).toBe(pool.info.minSize)
 192       }
 193       // We need to clean up the resources after our test
 194       await pool.destroy()
 195     }
 196   })
 197 })