tests/pools/thread/fixed.test.js

   1 const { expect } = require('expect')
   2 const { FixedThreadPool, PoolEvents } = require('../../../lib')
   3 const { WorkerFunctions } = require('../../test-types')
   4 const TestUtils = require('../../test-utils')
   5
   6 describe('Fixed thread pool test suite', () => {
   7   const numberOfThreads = 6
   8   const pool = new FixedThreadPool(
   9     numberOfThreads,
  10     './tests/worker-files/thread/testWorker.js',
  11     {
  12       errorHandler: e => console.error(e)
  13     }
  14   )
  15   const queuePool = new FixedThreadPool(
  16     numberOfThreads,
  17     './tests/worker-files/thread/testWorker.js',
  18     {
  19       enableTasksQueue: true,
  20       tasksQueueOptions: {
  21         concurrency: 2
  22       },
  23       errorHandler: e => console.error(e)
  24     }
  25   )
  26   const emptyPool = new FixedThreadPool(
  27     numberOfThreads,
  28     './tests/worker-files/thread/emptyWorker.js',
  29     { exitHandler: () => console.log('empty pool worker exited') }
  30   )
  31   const echoPool = new FixedThreadPool(
  32     numberOfThreads,
  33     './tests/worker-files/thread/echoWorker.js'
  34   )
  35   const errorPool = new FixedThreadPool(
  36     numberOfThreads,
  37     './tests/worker-files/thread/errorWorker.js',
  38     {
  39       errorHandler: e => console.error(e)
  40     }
  41   )
  42   const asyncErrorPool = new FixedThreadPool(
  43     numberOfThreads,
  44     './tests/worker-files/thread/asyncErrorWorker.js',
  45     {
  46       errorHandler: e => console.error(e)
  47     }
  48   )
  49   const asyncPool = new FixedThreadPool(
  50     numberOfThreads,
  51     './tests/worker-files/thread/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 thread', async () => {
  65     let result = await pool.execute({
  66       function: WorkerFunctions.fibonacci
  67     })
  68     expect(result).toBe(121393)
  69     result = await pool.execute({
  70       function: WorkerFunctions.factorial
  71     })
  72     expect(result).toBe(9.33262154439441e157)
  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 < numberOfThreads * 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 numberOfThreads + 1 times for a loop submitting up to numberOfThreads * 2 tasks to the fixed pool.
  88     expect(poolBusy).toBe(numberOfThreads + 1)
  89   })
  90
  91   it('Verify that tasks queuing is working', async () => {
  92     const promises = new Set()
  93     const maxMultiplier = 2
  94     for (let i = 0; i < numberOfThreads * maxMultiplier; i++) {
  95       promises.add(queuePool.execute())
  96     }
  97     expect(promises.size).toBe(numberOfThreads * maxMultiplier)
  98     for (const workerNode of queuePool.workerNodes) {
  99       expect(workerNode.workerUsage.tasks.executing).toBeLessThanOrEqual(
 100         queuePool.opts.tasksQueueOptions.concurrency
 101       )
 102       expect(workerNode.workerUsage.tasks.executed).toBe(0)
 103       expect(workerNode.tasksQueue.size).toBeGreaterThan(0)
 104     }
 105     expect(queuePool.info.executingTasks).toBe(numberOfThreads)
 106     expect(queuePool.info.queuedTasks).toBe(
 107       numberOfThreads * maxMultiplier - numberOfThreads
 108     )
 109     expect(queuePool.info.maxQueuedTasks).toBe(
 110       numberOfThreads * maxMultiplier - numberOfThreads
 111     )
 112     await Promise.all(promises)
 113     for (const workerNode of queuePool.workerNodes) {
 114       expect(workerNode.workerUsage.tasks.executing).toBe(0)
 115       expect(workerNode.workerUsage.tasks.executed).toBeGreaterThan(0)
 116       expect(workerNode.workerUsage.tasks.executed).toBeLessThanOrEqual(
 117         maxMultiplier
 118       )
 119       expect(workerNode.tasksQueue.size).toBe(0)
 120     }
 121   })
 122
 123   it('Verify that is possible to have a worker that return undefined', async () => {
 124     const result = await emptyPool.execute()
 125     expect(result).toBeUndefined()
 126   })
 127
 128   it('Verify that data are sent to the worker correctly', async () => {
 129     const data = { f: 10 }
 130     const result = await echoPool.execute(data)
 131     expect(result).toStrictEqual(data)
 132   })
 133
 134   it('Verify that error handling is working properly:sync', async () => {
 135     const data = { f: 10 }
 136     let taskError
 137     errorPool.emitter.on(PoolEvents.taskError, e => {
 138       taskError = e
 139     })
 140     let inError
 141     try {
 142       await errorPool.execute(data)
 143     } catch (e) {
 144       inError = e
 145     }
 146     expect(inError).toBeDefined()
 147     expect(inError).toBeInstanceOf(Error)
 148     expect(inError.message).toBeDefined()
 149     expect(typeof inError.message === 'string').toBe(true)
 150     expect(inError.message).toBe('Error Message from ThreadWorker')
 151     expect(taskError).toStrictEqual({
 152       message: new Error('Error Message from ThreadWorker'),
 153       data
 154     })
 155     expect(
 156       errorPool.workerNodes.some(
 157         workerNode => workerNode.workerUsage.tasks.failed === 1
 158       )
 159     ).toBe(true)
 160   })
 161
 162   it('Verify that error handling is working properly:async', async () => {
 163     const data = { f: 10 }
 164     let taskError
 165     asyncErrorPool.emitter.on(PoolEvents.taskError, e => {
 166       taskError = e
 167     })
 168     let inError
 169     try {
 170       await asyncErrorPool.execute(data)
 171     } catch (e) {
 172       inError = e
 173     }
 174     expect(inError).toBeDefined()
 175     expect(inError).toBeInstanceOf(Error)
 176     expect(inError.message).toBeDefined()
 177     expect(typeof inError.message === 'string').toBe(true)
 178     expect(inError.message).toBe('Error Message from ThreadWorker:async')
 179     expect(taskError).toStrictEqual({
 180       message: new Error('Error Message from ThreadWorker:async'),
 181       data
 182     })
 183     expect(
 184       asyncErrorPool.workerNodes.some(
 185         workerNode => workerNode.workerUsage.tasks.failed === 1
 186       )
 187     ).toBe(true)
 188   })
 189
 190   it('Verify that async function is working properly', async () => {
 191     const data = { f: 10 }
 192     const startTime = performance.now()
 193     const result = await asyncPool.execute(data)
 194     const usedTime = performance.now() - startTime
 195     expect(result).toStrictEqual(data)
 196     expect(usedTime).toBeGreaterThanOrEqual(2000)
 197   })
 198
 199   it('Shutdown test', async () => {
 200     const exitPromise = TestUtils.waitWorkerEvents(
 201       pool,
 202       'exit',
 203       numberOfThreads
 204     )
 205     await pool.destroy()
 206     const numberOfExitEvents = await exitPromise
 207     expect(numberOfExitEvents).toBe(numberOfThreads)
 208   })
 209
 210   it('Verify that thread pool options are checked', async () => {
 211     const workerFilePath = './tests/worker-files/cluster/testWorker.js'
 212     let pool1 = new FixedThreadPool(numberOfThreads, workerFilePath)
 213     expect(pool1.opts.workerOptions).toBeUndefined()
 214     await pool1.destroy()
 215     pool1 = new FixedThreadPool(numberOfThreads, workerFilePath, {
 216       workerOptions: {
 217         env: { TEST: 'test' },
 218         name: 'test'
 219       }
 220     })
 221     expect(pool1.opts.workerOptions).toStrictEqual({
 222       env: { TEST: 'test' },
 223       name: 'test'
 224     })
 225     await pool1.destroy()
 226   })
 227
 228   it('Should work even without opts in input', async () => {
 229     const pool1 = new FixedThreadPool(
 230       numberOfThreads,
 231       './tests/worker-files/thread/testWorker.js'
 232     )
 233     const res = await pool1.execute()
 234     expect(res).toBe(false)
 235     // We need to clean up the resources after our test
 236     await pool1.destroy()
 237   })
 238
 239   it('Verify that a pool with zero worker fails', async () => {
 240     expect(
 241       () => new FixedThreadPool(0, './tests/worker-files/thread/testWorker.js')
 242     ).toThrowError('Cannot instantiate a fixed pool with no worker')
 243   })
 244 })