tests/pools/thread/dynamic.test.mjs

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