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