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 { waitPoolEvents, waitWorkerEvents } = 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(75025)
  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).toStrictEqual({ ok: 1 })
  78   })
  79
  80   it("Verify that 'ready' event is emitted", async () => {
  81     const pool1 = new FixedThreadPool(
  82       numberOfThreads,
  83       './tests/worker-files/thread/testWorker.js',
  84       {
  85         errorHandler: e => console.error(e)
  86       }
  87     )
  88     let poolReady = 0
  89     pool1.emitter.on(PoolEvents.ready, () => ++poolReady)
  90     if (!pool1.info.ready) {
  91       await waitPoolEvents(pool1, PoolEvents.ready, 1)
  92     }
  93     expect(poolReady).toBe(1)
  94   })
  95
  96   it("Verify that 'busy' event is emitted", async () => {
  97     let poolBusy = 0
  98     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
  99     for (let i = 0; i < numberOfThreads * 2; i++) {
 100       pool.execute()
 101     }
 102     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 103     // So in total numberOfThreads + 1 times for a loop submitting up to numberOfThreads * 2 tasks to the fixed pool.
 104     expect(poolBusy).toBe(numberOfThreads + 1)
 105   })
 106
 107   it('Verify that tasks queuing is working', async () => {
 108     const promises = new Set()
 109     const maxMultiplier = 2
 110     for (let i = 0; i < numberOfThreads * maxMultiplier; i++) {
 111       promises.add(queuePool.execute())
 112     }
 113     expect(promises.size).toBe(numberOfThreads * maxMultiplier)
 114     for (const workerNode of queuePool.workerNodes) {
 115       expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
 116         queuePool.opts.tasksQueueOptions.concurrency
 117       )
 118       expect(workerNode.usage.tasks.executed).toBe(0)
 119       expect(workerNode.usage.tasks.queued).toBeGreaterThan(0)
 120       expect(workerNode.usage.tasks.maxQueued).toBeGreaterThan(0)
 121     }
 122     expect(queuePool.info.executingTasks).toBe(numberOfThreads)
 123     expect(queuePool.info.queuedTasks).toBe(
 124       numberOfThreads * maxMultiplier - numberOfThreads
 125     )
 126     expect(queuePool.info.maxQueuedTasks).toBe(
 127       numberOfThreads * maxMultiplier - numberOfThreads
 128     )
 129     await Promise.all(promises)
 130     for (const workerNode of queuePool.workerNodes) {
 131       expect(workerNode.usage.tasks.executing).toBe(0)
 132       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
 133       expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(maxMultiplier)
 134       expect(workerNode.usage.tasks.queued).toBe(0)
 135       expect(workerNode.usage.tasks.maxQueued).toBe(1)
 136     }
 137   })
 138
 139   it('Verify that is possible to have a worker that return undefined', async () => {
 140     const result = await emptyPool.execute()
 141     expect(result).toBeUndefined()
 142   })
 143
 144   it('Verify that data are sent to the worker correctly', async () => {
 145     const data = { f: 10 }
 146     const result = await echoPool.execute(data)
 147     expect(result).toStrictEqual(data)
 148   })
 149
 150   it('Verify that error handling is working properly:sync', async () => {
 151     const data = { f: 10 }
 152     let taskError
 153     errorPool.emitter.on(PoolEvents.taskError, e => {
 154       taskError = e
 155     })
 156     let inError
 157     try {
 158       await errorPool.execute(data)
 159     } catch (e) {
 160       inError = e
 161     }
 162     expect(inError).toBeDefined()
 163     expect(inError).toBeInstanceOf(Error)
 164     expect(inError.message).toBeDefined()
 165     expect(typeof inError.message === 'string').toBe(true)
 166     expect(inError.message).toBe('Error Message from ThreadWorker')
 167     expect(taskError).toStrictEqual({
 168       name: 'default',
 169       message: new Error('Error Message from ThreadWorker'),
 170       data
 171     })
 172     expect(
 173       errorPool.workerNodes.some(
 174         workerNode => workerNode.usage.tasks.failed === 1
 175       )
 176     ).toBe(true)
 177   })
 178
 179   it('Verify that error handling is working properly:async', async () => {
 180     const data = { f: 10 }
 181     let taskError
 182     asyncErrorPool.emitter.on(PoolEvents.taskError, e => {
 183       taskError = e
 184     })
 185     let inError
 186     try {
 187       await asyncErrorPool.execute(data)
 188     } catch (e) {
 189       inError = e
 190     }
 191     expect(inError).toBeDefined()
 192     expect(inError).toBeInstanceOf(Error)
 193     expect(inError.message).toBeDefined()
 194     expect(typeof inError.message === 'string').toBe(true)
 195     expect(inError.message).toBe('Error Message from ThreadWorker:async')
 196     expect(taskError).toStrictEqual({
 197       name: 'default',
 198       message: new Error('Error Message from ThreadWorker:async'),
 199       data
 200     })
 201     expect(
 202       asyncErrorPool.workerNodes.some(
 203         workerNode => workerNode.usage.tasks.failed === 1
 204       )
 205     ).toBe(true)
 206   })
 207
 208   it('Verify that async function is working properly', async () => {
 209     const data = { f: 10 }
 210     const startTime = performance.now()
 211     const result = await asyncPool.execute(data)
 212     const usedTime = performance.now() - startTime
 213     expect(result).toStrictEqual(data)
 214     expect(usedTime).toBeGreaterThanOrEqual(2000)
 215   })
 216
 217   it('Shutdown test', async () => {
 218     const exitPromise = waitWorkerEvents(pool, 'exit', numberOfThreads)
 219     await pool.destroy()
 220     const numberOfExitEvents = await exitPromise
 221     expect(numberOfExitEvents).toBe(numberOfThreads)
 222   })
 223
 224   it('Verify that thread pool options are checked', async () => {
 225     const workerFilePath = './tests/worker-files/thread/testWorker.js'
 226     let pool1 = new FixedThreadPool(numberOfThreads, workerFilePath)
 227     expect(pool1.opts.workerOptions).toBeUndefined()
 228     await pool1.destroy()
 229     pool1 = new FixedThreadPool(numberOfThreads, workerFilePath, {
 230       workerOptions: {
 231         env: { TEST: 'test' },
 232         name: 'test'
 233       }
 234     })
 235     expect(pool1.opts.workerOptions).toStrictEqual({
 236       env: { TEST: 'test' },
 237       name: 'test'
 238     })
 239     await pool1.destroy()
 240   })
 241
 242   it('Should work even without opts in input', async () => {
 243     const pool1 = new FixedThreadPool(
 244       numberOfThreads,
 245       './tests/worker-files/thread/testWorker.js'
 246     )
 247     const res = await pool1.execute()
 248     expect(res).toStrictEqual({ ok: 1 })
 249     // We need to clean up the resources after our test
 250     await pool1.destroy()
 251   })
 252
 253   it('Verify that a pool with zero worker fails', async () => {
 254     expect(
 255       () => new FixedThreadPool(0, './tests/worker-files/thread/testWorker.js')
 256     ).toThrowError('Cannot instantiate a fixed pool with zero worker')
 257   })
 258 })