tests/pools/thread/fixed.test.js

   1 const { expect } = require('expect')
   2 const { FixedThreadPool, PoolEvents } = require('../../../lib')
   3 const { TaskFunctions } = require('../../test-types')
   4 const { waitPoolEvents, waitWorkerEvents } = require('../../test-utils')
   5
   6 describe('Fixed thread pool test suite', () => {
   7   const numberOfThreads = 6
   8   const tasksConcurrency = 2
   9   const pool = new FixedThreadPool(
  10     numberOfThreads,
  11     './tests/worker-files/thread/testWorker.js',
  12     {
  13       errorHandler: (e) => console.error(e)
  14     }
  15   )
  16   const queuePool = new FixedThreadPool(
  17     numberOfThreads,
  18     './tests/worker-files/thread/testWorker.js',
  19     {
  20       enableTasksQueue: true,
  21       tasksQueueOptions: {
  22         concurrency: tasksConcurrency
  23       },
  24       errorHandler: (e) => console.error(e)
  25     }
  26   )
  27   const emptyPool = new FixedThreadPool(
  28     numberOfThreads,
  29     './tests/worker-files/thread/emptyWorker.js',
  30     { exitHandler: () => console.info('empty pool worker exited') }
  31   )
  32   const echoPool = new FixedThreadPool(
  33     numberOfThreads,
  34     './tests/worker-files/thread/echoWorker.js'
  35   )
  36   const errorPool = new FixedThreadPool(
  37     numberOfThreads,
  38     './tests/worker-files/thread/errorWorker.js',
  39     {
  40       errorHandler: (e) => console.error(e)
  41     }
  42   )
  43   const asyncErrorPool = new FixedThreadPool(
  44     numberOfThreads,
  45     './tests/worker-files/thread/asyncErrorWorker.js',
  46     {
  47       errorHandler: (e) => console.error(e)
  48     }
  49   )
  50   const asyncPool = new FixedThreadPool(
  51     numberOfThreads,
  52     './tests/worker-files/thread/asyncWorker.js'
  53   )
  54
  55   after('Destroy all pools', async () => {
  56     // We need to clean up the resources after our test
  57     await echoPool.destroy()
  58     await asyncPool.destroy()
  59     await errorPool.destroy()
  60     await asyncErrorPool.destroy()
  61     await emptyPool.destroy()
  62     await queuePool.destroy()
  63   })
  64
  65   it('Verify that the function is executed in a worker thread', async () => {
  66     let result = await pool.execute({
  67       function: TaskFunctions.fibonacci
  68     })
  69     expect(result).toBe(75025)
  70     result = await pool.execute({
  71       function: TaskFunctions.factorial
  72     })
  73     expect(result).toBe(9.33262154439441e157)
  74   })
  75
  76   it('Verify that is possible to invoke the execute() method without input', async () => {
  77     const result = await pool.execute()
  78     expect(result).toStrictEqual({ ok: 1 })
  79   })
  80
  81   it("Verify that 'ready' event is emitted", async () => {
  82     const pool1 = new FixedThreadPool(
  83       numberOfThreads,
  84       './tests/worker-files/thread/testWorker.js',
  85       {
  86         errorHandler: (e) => console.error(e)
  87       }
  88     )
  89     let poolReady = 0
  90     pool1.emitter.on(PoolEvents.ready, () => ++poolReady)
  91     await waitPoolEvents(pool1, 'ready', 1)
  92     expect(poolReady).toBe(1)
  93   })
  94
  95   it("Verify that 'busy' event is emitted", () => {
  96     let poolBusy = 0
  97     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
  98     for (let i = 0; i < numberOfThreads * 2; i++) {
  99       pool.execute()
 100     }
 101     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 102     // So in total numberOfThreads + 1 times for a loop submitting up to numberOfThreads * 2 tasks to the fixed pool.
 103     expect(poolBusy).toBe(numberOfThreads + 1)
 104   })
 105
 106   it('Verify that tasks queuing is working', async () => {
 107     const promises = new Set()
 108     const maxMultiplier = 3 // Must be greater than tasksConcurrency
 109     for (let i = 0; i < numberOfThreads * maxMultiplier; i++) {
 110       promises.add(queuePool.execute())
 111     }
 112     expect(promises.size).toBe(numberOfThreads * maxMultiplier)
 113     for (const workerNode of queuePool.workerNodes) {
 114       expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
 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).toBe(
 120         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
 121       )
 122       expect(workerNode.usage.tasks.maxQueued).toBe(
 123         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
 124       )
 125     }
 126     expect(queuePool.info.executingTasks).toBe(
 127       numberOfThreads * queuePool.opts.tasksQueueOptions.concurrency
 128     )
 129     expect(queuePool.info.queuedTasks).toBe(
 130       numberOfThreads *
 131         (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
 132     )
 133     expect(queuePool.info.maxQueuedTasks).toBe(
 134       numberOfThreads *
 135         (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
 136     )
 137     expect(queuePool.info.backPressure).toBe(false)
 138     await Promise.all(promises)
 139     for (const workerNode of queuePool.workerNodes) {
 140       expect(workerNode.usage.tasks.executing).toBe(0)
 141       expect(workerNode.usage.tasks.executed).toBe(maxMultiplier)
 142       expect(workerNode.usage.tasks.queued).toBe(0)
 143       expect(workerNode.usage.tasks.maxQueued).toBe(
 144         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
 145       )
 146     }
 147   })
 148
 149   it('Verify that is possible to have a worker that return undefined', async () => {
 150     const result = await emptyPool.execute()
 151     expect(result).toBeUndefined()
 152   })
 153
 154   it('Verify that data are sent to the worker correctly', async () => {
 155     const data = { f: 10 }
 156     const result = await echoPool.execute(data)
 157     expect(result).toStrictEqual(data)
 158   })
 159
 160   it('Verify that transferable objects are sent to the worker correctly', async () => {
 161     let error
 162     let result
 163     try {
 164       result = await pool.execute(undefined, undefined, [
 165         new ArrayBuffer(16),
 166         new MessageChannel().port1
 167       ])
 168     } catch (e) {
 169       error = e
 170     }
 171     expect(result).toStrictEqual({ ok: 1 })
 172     expect(error).toBeUndefined()
 173     try {
 174       result = await pool.execute(undefined, undefined, [
 175         new SharedArrayBuffer(16)
 176       ])
 177     } catch (e) {
 178       error = e
 179     }
 180     expect(result).toStrictEqual({ ok: 1 })
 181     expect(error).toStrictEqual(
 182       new TypeError('Found invalid object in transferList')
 183     )
 184   })
 185
 186   it('Verify that error handling is working properly:sync', async () => {
 187     const data = { f: 10 }
 188     let taskError
 189     errorPool.emitter.on(PoolEvents.taskError, (e) => {
 190       taskError = e
 191     })
 192     let inError
 193     try {
 194       await errorPool.execute(data)
 195     } catch (e) {
 196       inError = e
 197     }
 198     expect(inError).toBeDefined()
 199     expect(inError).toBeInstanceOf(Error)
 200     expect(inError.message).toBeDefined()
 201     expect(typeof inError.message === 'string').toBe(true)
 202     expect(inError.message).toBe('Error Message from ThreadWorker')
 203     expect(taskError).toStrictEqual({
 204       name: 'default',
 205       message: new Error('Error Message from ThreadWorker'),
 206       data
 207     })
 208     expect(
 209       errorPool.workerNodes.some(
 210         (workerNode) => workerNode.usage.tasks.failed === 1
 211       )
 212     ).toBe(true)
 213   })
 214
 215   it('Verify that error handling is working properly:async', async () => {
 216     const data = { f: 10 }
 217     let taskError
 218     asyncErrorPool.emitter.on(PoolEvents.taskError, (e) => {
 219       taskError = e
 220     })
 221     let inError
 222     try {
 223       await asyncErrorPool.execute(data)
 224     } catch (e) {
 225       inError = e
 226     }
 227     expect(inError).toBeDefined()
 228     expect(inError).toBeInstanceOf(Error)
 229     expect(inError.message).toBeDefined()
 230     expect(typeof inError.message === 'string').toBe(true)
 231     expect(inError.message).toBe('Error Message from ThreadWorker:async')
 232     expect(taskError).toStrictEqual({
 233       name: 'default',
 234       message: new Error('Error Message from ThreadWorker:async'),
 235       data
 236     })
 237     expect(
 238       asyncErrorPool.workerNodes.some(
 239         (workerNode) => workerNode.usage.tasks.failed === 1
 240       )
 241     ).toBe(true)
 242   })
 243
 244   it('Verify that async function is working properly', async () => {
 245     const data = { f: 10 }
 246     const startTime = performance.now()
 247     const result = await asyncPool.execute(data)
 248     const usedTime = performance.now() - startTime
 249     expect(result).toStrictEqual(data)
 250     expect(usedTime).toBeGreaterThanOrEqual(2000)
 251   })
 252
 253   it('Shutdown test', async () => {
 254     const exitPromise = waitWorkerEvents(pool, 'exit', numberOfThreads)
 255     let poolDestroy = 0
 256     pool.emitter.on(PoolEvents.destroy, () => ++poolDestroy)
 257     await pool.destroy()
 258     const numberOfExitEvents = await exitPromise
 259     expect(numberOfExitEvents).toBe(numberOfThreads)
 260     expect(poolDestroy).toBe(1)
 261   })
 262
 263   it('Verify that thread pool options are checked', async () => {
 264     const workerFilePath = './tests/worker-files/thread/testWorker.js'
 265     let pool1 = new FixedThreadPool(numberOfThreads, workerFilePath)
 266     expect(pool1.opts.workerOptions).toBeUndefined()
 267     await pool1.destroy()
 268     pool1 = new FixedThreadPool(numberOfThreads, workerFilePath, {
 269       workerOptions: {
 270         env: { TEST: 'test' },
 271         name: 'test'
 272       }
 273     })
 274     expect(pool1.opts.workerOptions).toStrictEqual({
 275       env: { TEST: 'test' },
 276       name: 'test'
 277     })
 278     await pool1.destroy()
 279   })
 280
 281   it('Should work even without opts in input', async () => {
 282     const pool1 = new FixedThreadPool(
 283       numberOfThreads,
 284       './tests/worker-files/thread/testWorker.js'
 285     )
 286     const res = await pool1.execute()
 287     expect(res).toStrictEqual({ ok: 1 })
 288     // We need to clean up the resources after our test
 289     await pool1.destroy()
 290   })
 291
 292   it('Verify that a pool with zero worker fails', async () => {
 293     expect(
 294       () => new FixedThreadPool(0, './tests/worker-files/thread/testWorker.js')
 295     ).toThrowError('Cannot instantiate a fixed pool with zero worker')
 296   })
 297 })