tests/pools/thread/fixed.test.mjs

   1 import { expect } from 'expect'
   2 import { FixedThreadPool, PoolEvents } from '../../../lib/index.cjs'
   3 import { TaskFunctions } from '../../test-types.cjs'
   4 import { waitPoolEvents, waitWorkerEvents } from '../../test-utils.cjs'
   5 import { DEFAULT_TASK_NAME } from '../../../lib/utils.cjs'
   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.mjs',
  13     {
  14       errorHandler: e => console.error(e)
  15     }
  16   )
  17   const queuePool = new FixedThreadPool(
  18     numberOfThreads,
  19     './tests/worker-files/thread/testWorker.mjs',
  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.mjs',
  31     { exitHandler: () => console.info('empty pool worker exited') }
  32   )
  33   const echoPool = new FixedThreadPool(
  34     numberOfThreads,
  35     './tests/worker-files/thread/echoWorker.mjs'
  36   )
  37   const errorPool = new FixedThreadPool(
  38     numberOfThreads,
  39     './tests/worker-files/thread/errorWorker.mjs',
  40     {
  41       errorHandler: e => console.error(e)
  42     }
  43   )
  44   const asyncErrorPool = new FixedThreadPool(
  45     numberOfThreads,
  46     './tests/worker-files/thread/asyncErrorWorker.mjs',
  47     {
  48       errorHandler: e => console.error(e)
  49     }
  50   )
  51   const asyncPool = new FixedThreadPool(
  52     numberOfThreads,
  53     './tests/worker-files/thread/asyncWorker.mjs'
  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 pool = new FixedThreadPool(
  84       numberOfThreads,
  85       './tests/worker-files/thread/testWorker.mjs',
  86       {
  87         errorHandler: e => console.error(e)
  88       }
  89     )
  90     expect(pool.emitter.eventNames()).toStrictEqual([])
  91     let poolReady = 0
  92     pool.emitter.on(PoolEvents.ready, () => ++poolReady)
  93     await waitPoolEvents(pool, PoolEvents.ready, 1)
  94     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.ready])
  95     expect(poolReady).toBe(1)
  96     await pool.destroy()
  97   })
  98
  99   it("Verify that 'busy' event is emitted", async () => {
 100     const promises = new Set()
 101     expect(pool.emitter.eventNames()).toStrictEqual([])
 102     let poolBusy = 0
 103     pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
 104     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
 105     for (let i = 0; i < numberOfThreads * 2; i++) {
 106       promises.add(pool.execute())
 107     }
 108     await Promise.all(promises)
 109     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 110     // So in total numberOfThreads + 1 times for a loop submitting up to numberOfThreads * 2 tasks to the fixed pool.
 111     expect(poolBusy).toBe(numberOfThreads + 1)
 112   })
 113
 114   it('Verify that tasks queuing is working', async () => {
 115     const promises = new Set()
 116     const maxMultiplier = 3 // Must be greater than tasksConcurrency
 117     for (let i = 0; i < numberOfThreads * maxMultiplier; i++) {
 118       promises.add(queuePool.execute())
 119     }
 120     expect(promises.size).toBe(numberOfThreads * maxMultiplier)
 121     for (const workerNode of queuePool.workerNodes) {
 122       expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
 123       expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
 124         queuePool.opts.tasksQueueOptions.concurrency
 125       )
 126       expect(workerNode.usage.tasks.executed).toBe(0)
 127       expect(workerNode.usage.tasks.queued).toBe(
 128         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
 129       )
 130       expect(workerNode.usage.tasks.maxQueued).toBe(
 131         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
 132       )
 133       expect(workerNode.usage.tasks.sequentiallyStolen).toBe(0)
 134       expect(workerNode.usage.tasks.stolen).toBe(0)
 135     }
 136     expect(queuePool.info.executedTasks).toBe(0)
 137     expect(queuePool.info.executingTasks).toBe(
 138       numberOfThreads * queuePool.opts.tasksQueueOptions.concurrency
 139     )
 140     expect(queuePool.info.queuedTasks).toBe(
 141       numberOfThreads *
 142         (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
 143     )
 144     expect(queuePool.info.maxQueuedTasks).toBe(
 145       numberOfThreads *
 146         (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
 147     )
 148     expect(queuePool.info.backPressure).toBe(false)
 149     expect(queuePool.info.stolenTasks).toBe(0)
 150     await Promise.all(promises)
 151     for (const workerNode of queuePool.workerNodes) {
 152       expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
 153       expect(workerNode.usage.tasks.executing).toBeLessThanOrEqual(
 154         numberOfThreads * maxMultiplier
 155       )
 156       expect(workerNode.usage.tasks.executed).toBe(maxMultiplier)
 157       expect(workerNode.usage.tasks.queued).toBe(0)
 158       expect(workerNode.usage.tasks.maxQueued).toBe(
 159         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
 160       )
 161       expect(workerNode.usage.tasks.sequentiallyStolen).toBeGreaterThanOrEqual(
 162         0
 163       )
 164       expect(workerNode.usage.tasks.sequentiallyStolen).toBeLessThanOrEqual(
 165         numberOfThreads * maxMultiplier
 166       )
 167       expect(workerNode.usage.tasks.stolen).toBeGreaterThanOrEqual(0)
 168       expect(workerNode.usage.tasks.stolen).toBeLessThanOrEqual(
 169         numberOfThreads * maxMultiplier
 170       )
 171     }
 172     expect(queuePool.info.executedTasks).toBe(numberOfThreads * maxMultiplier)
 173     expect(queuePool.info.backPressure).toBe(false)
 174     expect(queuePool.info.stolenTasks).toBeGreaterThanOrEqual(0)
 175     expect(queuePool.info.stolenTasks).toBeLessThanOrEqual(
 176       numberOfThreads * maxMultiplier
 177     )
 178   })
 179
 180   it('Verify that is possible to have a worker that return undefined', async () => {
 181     const result = await emptyPool.execute()
 182     expect(result).toBeUndefined()
 183   })
 184
 185   it('Verify that data are sent to the worker correctly', async () => {
 186     const data = { f: 10 }
 187     const result = await echoPool.execute(data)
 188     expect(result).toStrictEqual(data)
 189   })
 190
 191   it('Verify that transferable objects are sent to the worker correctly', async () => {
 192     let error
 193     let result
 194     try {
 195       result = await pool.execute(undefined, undefined, [
 196         new ArrayBuffer(16),
 197         new MessageChannel().port1
 198       ])
 199     } catch (e) {
 200       error = e
 201     }
 202     expect(result).toStrictEqual({ ok: 1 })
 203     expect(error).toBeUndefined()
 204     try {
 205       result = await pool.execute(undefined, undefined, [
 206         new SharedArrayBuffer(16)
 207       ])
 208     } catch (e) {
 209       error = e
 210     }
 211     expect(result).toStrictEqual({ ok: 1 })
 212     expect(error).toBeInstanceOf(Error)
 213     expect(error.message).toMatch(
 214       /Found invalid (object|value) in transferList/
 215     )
 216   })
 217
 218   it('Verify that error handling is working properly:sync', async () => {
 219     const data = { f: 10 }
 220     expect(errorPool.emitter.eventNames()).toStrictEqual([])
 221     let taskError
 222     errorPool.emitter.on(PoolEvents.taskError, e => {
 223       taskError = e
 224     })
 225     expect(errorPool.emitter.eventNames()).toStrictEqual([PoolEvents.taskError])
 226     let inError
 227     try {
 228       await errorPool.execute(data)
 229     } catch (e) {
 230       inError = e
 231     }
 232     expect(inError).toBeDefined()
 233     expect(inError).toBeInstanceOf(Error)
 234     expect(inError.message).toBeDefined()
 235     expect(typeof inError.message === 'string').toBe(true)
 236     expect(inError.message).toBe('Error Message from ThreadWorker')
 237     expect(taskError).toStrictEqual({
 238       name: DEFAULT_TASK_NAME,
 239       message: new Error('Error Message from ThreadWorker'),
 240       data
 241     })
 242     expect(
 243       errorPool.workerNodes.some(
 244         workerNode => workerNode.usage.tasks.failed === 1
 245       )
 246     ).toBe(true)
 247   })
 248
 249   it('Verify that error handling is working properly:async', async () => {
 250     const data = { f: 10 }
 251     expect(asyncErrorPool.emitter.eventNames()).toStrictEqual([])
 252     let taskError
 253     asyncErrorPool.emitter.on(PoolEvents.taskError, e => {
 254       taskError = e
 255     })
 256     expect(asyncErrorPool.emitter.eventNames()).toStrictEqual([
 257       PoolEvents.taskError
 258     ])
 259     let inError
 260     try {
 261       await asyncErrorPool.execute(data)
 262     } catch (e) {
 263       inError = e
 264     }
 265     expect(inError).toBeDefined()
 266     expect(inError).toBeInstanceOf(Error)
 267     expect(inError.message).toBeDefined()
 268     expect(typeof inError.message === 'string').toBe(true)
 269     expect(inError.message).toBe('Error Message from ThreadWorker:async')
 270     expect(taskError).toStrictEqual({
 271       name: DEFAULT_TASK_NAME,
 272       message: new Error('Error Message from ThreadWorker:async'),
 273       data
 274     })
 275     expect(
 276       asyncErrorPool.workerNodes.some(
 277         workerNode => workerNode.usage.tasks.failed === 1
 278       )
 279     ).toBe(true)
 280   })
 281
 282   it('Verify that async function is working properly', async () => {
 283     const data = { f: 10 }
 284     const startTime = performance.now()
 285     const result = await asyncPool.execute(data)
 286     const usedTime = performance.now() - startTime
 287     expect(result).toStrictEqual(data)
 288     expect(usedTime).toBeGreaterThanOrEqual(2000)
 289   })
 290
 291   it('Shutdown test', async () => {
 292     const exitPromise = waitWorkerEvents(pool, 'exit', numberOfThreads)
 293     expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
 294     let poolDestroy = 0
 295     pool.emitter.on(PoolEvents.destroy, () => ++poolDestroy)
 296     expect(pool.emitter.eventNames()).toStrictEqual([
 297       PoolEvents.busy,
 298       PoolEvents.destroy
 299     ])
 300     await pool.destroy()
 301     const numberOfExitEvents = await exitPromise
 302     expect(pool.started).toBe(false)
 303     expect(pool.readyEventEmitted).toBe(false)
 304     expect(pool.emitter.eventNames()).toStrictEqual([])
 305     expect(pool.workerNodes.length).toBe(0)
 306     expect(numberOfExitEvents).toBe(numberOfThreads)
 307     expect(poolDestroy).toBe(1)
 308   })
 309
 310   it('Verify that thread pool options are checked', async () => {
 311     const workerFilePath = './tests/worker-files/thread/testWorker.mjs'
 312     let pool = new FixedThreadPool(numberOfThreads, workerFilePath)
 313     expect(pool.opts.workerOptions).toBeUndefined()
 314     await pool.destroy()
 315     pool = new FixedThreadPool(numberOfThreads, workerFilePath, {
 316       workerOptions: {
 317         env: { TEST: 'test' },
 318         name: 'test'
 319       }
 320     })
 321     expect(pool.opts.workerOptions).toStrictEqual({
 322       env: { TEST: 'test' },
 323       name: 'test'
 324     })
 325     await pool.destroy()
 326   })
 327
 328   it('Should work even without opts in input', async () => {
 329     const workerFilePath = './tests/worker-files/thread/testWorker.mjs'
 330     const pool = new FixedThreadPool(numberOfThreads, workerFilePath)
 331     const res = await pool.execute()
 332     expect(res).toStrictEqual({ ok: 1 })
 333     // We need to clean up the resources after our test
 334     await pool.destroy()
 335   })
 336
 337   it('Verify destroyWorkerNode()', async () => {
 338     const workerFilePath = './tests/worker-files/thread/testWorker.mjs'
 339     const pool = new FixedThreadPool(numberOfThreads, workerFilePath)
 340     const workerNodeKey = 0
 341     let exitEvent = 0
 342     pool.workerNodes[workerNodeKey].worker.on('exit', () => {
 343       ++exitEvent
 344     })
 345     await expect(pool.destroyWorkerNode(workerNodeKey)).resolves.toBeUndefined()
 346     expect(exitEvent).toBe(1)
 347     expect(pool.workerNodes.length).toBe(numberOfThreads - 1)
 348     await pool.destroy()
 349   })
 350
 351   it('Verify that a pool with zero worker fails', () => {
 352     expect(
 353       () => new FixedThreadPool(0, './tests/worker-files/thread/testWorker.mjs')
 354     ).toThrow('Cannot instantiate a fixed pool with zero worker')
 355   })
 356 })