tests/pools/thread/dynamic.test.mjs

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