tests/pools/cluster/dynamic.test.mjs

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