tests/pools/cluster/fixed.test.js

   1 const { expect } = require('expect')
   2 const { FixedClusterPool, PoolEvents } = require('../../../lib/index')
   3 const { WorkerFunctions } = require('../../test-types')
   4 const TestUtils = require('../../test-utils')
   5
   6 describe('Fixed cluster pool test suite', () => {
   7   const numberOfWorkers = 6
   8   const pool = new FixedClusterPool(
   9     numberOfWorkers,
  10     './tests/worker-files/cluster/testWorker.js',
  11     {
  12       errorHandler: e => console.error(e)
  13     }
  14   )
  15   const queuePool = new FixedClusterPool(
  16     numberOfWorkers,
  17     './tests/worker-files/cluster/testWorker.js',
  18     {
  19       enableTasksQueue: true,
  20       tasksQueueOptions: {
  21         concurrency: 2
  22       },
  23       errorHandler: e => console.error(e)
  24     }
  25   )
  26   const emptyPool = new FixedClusterPool(
  27     numberOfWorkers,
  28     './tests/worker-files/cluster/emptyWorker.js',
  29     { exitHandler: () => console.log('empty pool worker exited') }
  30   )
  31   const echoPool = new FixedClusterPool(
  32     numberOfWorkers,
  33     './tests/worker-files/cluster/echoWorker.js'
  34   )
  35   const errorPool = new FixedClusterPool(
  36     numberOfWorkers,
  37     './tests/worker-files/cluster/errorWorker.js',
  38     {
  39       errorHandler: e => console.error(e)
  40     }
  41   )
  42   const asyncErrorPool = new FixedClusterPool(
  43     numberOfWorkers,
  44     './tests/worker-files/cluster/asyncErrorWorker.js',
  45     {
  46       errorHandler: e => console.error(e)
  47     }
  48   )
  49   const asyncPool = new FixedClusterPool(
  50     numberOfWorkers,
  51     './tests/worker-files/cluster/asyncWorker.js'
  52   )
  53
  54   after('Destroy all pools', async () => {
  55     // We need to clean up the resources after our test
  56     await echoPool.destroy()
  57     await asyncPool.destroy()
  58     await errorPool.destroy()
  59     await asyncErrorPool.destroy()
  60     await emptyPool.destroy()
  61     await queuePool.destroy()
  62   })
  63
  64   it('Verify that the function is executed in a worker cluster', async () => {
  65     let result = await pool.execute({
  66       function: WorkerFunctions.fibonacci
  67     })
  68     expect(result).toBe(false)
  69     result = await pool.execute({
  70       function: WorkerFunctions.factorial
  71     })
  72     expect(result).toBe(false)
  73   })
  74
  75   it('Verify that is possible to invoke the execute method without input', async () => {
  76     const result = await pool.execute()
  77     expect(result).toBe(false)
  78   })
  79
  80   it("Verify that 'busy' event is emitted", async () => {
  81     let poolBusy = 0
  82     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
  83     for (let i = 0; i < numberOfWorkers * 2; i++) {
  84       pool.execute()
  85     }
  86     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
  87     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
  88     expect(poolBusy).toBe(numberOfWorkers + 1)
  89   })
  90
  91   it('Verify that tasks queuing is working', async () => {
  92     const maxMultiplier = 10
  93     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
  94       queuePool.execute()
  95     }
  96     for (const workerNode of queuePool.workerNodes) {
  97       expect(workerNode.tasksUsage.running).toBeLessThanOrEqual(
  98         queuePool.opts.tasksQueueOptions.concurrency
  99       )
 100       expect(workerNode.tasksUsage.run).toBe(0)
 101       expect(workerNode.tasksQueue.length).toBeGreaterThan(0)
 102     }
 103     // FIXME: wait for ongoing tasks to be executed
 104     const promises = []
 105     for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
 106       promises.push(queuePool.execute())
 107     }
 108     await Promise.all(promises)
 109     for (const workerNode of queuePool.workerNodes) {
 110       expect(workerNode.tasksUsage.running).toBe(0)
 111       expect(workerNode.tasksUsage.run).toBeGreaterThan(0)
 112       expect(workerNode.tasksQueue.length).toBe(0)
 113     }
 114   })
 115
 116   it('Verify that is possible to have a worker that return undefined', async () => {
 117     const result = await emptyPool.execute()
 118     expect(result).toBeUndefined()
 119   })
 120
 121   it('Verify that data are sent to the worker correctly', async () => {
 122     const data = { f: 10 }
 123     const result = await echoPool.execute(data)
 124     expect(result).toStrictEqual(data)
 125   })
 126
 127   it('Verify that error handling is working properly:sync', async () => {
 128     const data = { f: 10 }
 129     let inError
 130     try {
 131       await errorPool.execute(data)
 132     } catch (e) {
 133       inError = e
 134     }
 135     expect(inError).toBeDefined()
 136     expect(typeof inError === 'string').toBe(true)
 137     expect(inError).toBe('Error Message from ClusterWorker')
 138   })
 139
 140   it('Verify that error handling is working properly:async', async () => {
 141     const data = { f: 10 }
 142     let inError
 143     try {
 144       await asyncErrorPool.execute(data)
 145     } catch (e) {
 146       inError = e
 147     }
 148     expect(inError).toBeDefined()
 149     expect(typeof inError === 'string').toBe(true)
 150     expect(inError).toBe('Error Message from ClusterWorker:async')
 151   })
 152
 153   it('Verify that async function is working properly', async () => {
 154     const data = { f: 10 }
 155     const startTime = performance.now()
 156     const result = await asyncPool.execute(data)
 157     const usedTime = performance.now() - startTime
 158     expect(result).toStrictEqual(data)
 159     expect(usedTime).toBeGreaterThanOrEqual(2000)
 160   })
 161
 162   it('Shutdown test', async () => {
 163     const exitPromise = TestUtils.waitExits(pool, numberOfWorkers)
 164     await pool.destroy()
 165     const numberOfExitEvents = await exitPromise
 166     expect(numberOfExitEvents).toBe(numberOfWorkers)
 167   })
 168
 169   it('Verify that cluster pool options are checked', async () => {
 170     const workerFilePath = './tests/worker-files/cluster/testWorker.js'
 171     let pool1 = new FixedClusterPool(numberOfWorkers, workerFilePath)
 172     expect(pool1.opts.env).toBeUndefined()
 173     expect(pool1.opts.settings).toBeUndefined()
 174     await pool1.destroy()
 175     pool1 = new FixedClusterPool(numberOfWorkers, workerFilePath, {
 176       env: { TEST: 'test' },
 177       settings: { args: ['--use', 'http'], silent: true }
 178     })
 179     expect(pool1.opts.env).toStrictEqual({ TEST: 'test' })
 180     expect(pool1.opts.settings).toStrictEqual({
 181       args: ['--use', 'http'],
 182       silent: true
 183     })
 184     expect({ ...pool1.opts.settings, exec: workerFilePath }).toStrictEqual({
 185       args: ['--use', 'http'],
 186       silent: true,
 187       exec: workerFilePath
 188     })
 189     await pool1.destroy()
 190   })
 191
 192   it('Should work even without opts in input', async () => {
 193     const pool1 = new FixedClusterPool(
 194       numberOfWorkers,
 195       './tests/worker-files/cluster/testWorker.js'
 196     )
 197     const res = await pool1.execute()
 198     expect(res).toBe(false)
 199     // We need to clean up the resources after our test
 200     await pool1.destroy()
 201   })
 202
 203   it('Verify that a pool with zero worker fails', async () => {
 204     expect(
 205       () =>
 206         new FixedClusterPool(0, './tests/worker-files/cluster/testWorker.js')
 207     ).toThrowError(new Error('Cannot instantiate a fixed pool with no worker'))
 208   })
 209 })