tests/pools/abstract/abstract-pool.test.js

   1 const { expect } = require('expect')
   2 const {
   3   FixedClusterPool,
   4   DynamicThreadPool,
   5   FixedThreadPool,
   6   WorkerChoiceStrategies
   7 } = require('../../../lib/index')
   8
   9 describe('Abstract pool test suite', () => {
  10   const numberOfWorkers = 1
  11   const workerNotFoundInPoolError = new Error(
  12     'Worker could not be found in the pool'
  13   )
  14   class StubPoolWithRemoveAllWorker extends FixedThreadPool {
  15     removeAllWorker () {
  16       this.workers = new Map()
  17       this.promiseMap.clear()
  18     }
  19   }
  20   class StubPoolWithIsMain extends FixedThreadPool {
  21     isMain () {
  22       return false
  23     }
  24   }
  25
  26   it('Simulate pool creation from a non main thread/process', () => {
  27     expect(
  28       () =>
  29         new StubPoolWithIsMain(
  30           numberOfWorkers,
  31           './tests/worker-files/thread/testWorker.js',
  32           {
  33             errorHandler: e => console.error(e)
  34           }
  35         )
  36     ).toThrowError(new Error('Cannot start a pool from a worker!'))
  37   })
  38
  39   it('Verify that filePath is checked', () => {
  40     const expectedError = new Error(
  41       'Please specify a file with a worker implementation'
  42     )
  43     expect(() => new FixedThreadPool(numberOfWorkers)).toThrowError(
  44       expectedError
  45     )
  46     expect(() => new FixedThreadPool(numberOfWorkers, '')).toThrowError(
  47       expectedError
  48     )
  49   })
  50
  51   it('Verify that numberOfWorkers is checked', () => {
  52     expect(() => new FixedThreadPool()).toThrowError(
  53       new Error(
  54         'Cannot instantiate a pool without specifying the number of workers'
  55       )
  56     )
  57   })
  58
  59   it('Verify that a negative number of workers is checked', () => {
  60     expect(
  61       () =>
  62         new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
  63     ).toThrowError(
  64       new RangeError(
  65         'Cannot instantiate a pool with a negative number of workers'
  66       )
  67     )
  68   })
  69
  70   it('Verify that a non integer number of workers is checked', () => {
  71     expect(
  72       () =>
  73         new FixedThreadPool(0.25, './tests/worker-files/thread/testWorker.js')
  74     ).toThrowError(
  75       new TypeError(
  76         'Cannot instantiate a pool with a non integer number of workers'
  77       )
  78     )
  79   })
  80
  81   it('Verify that pool options are checked', async () => {
  82     let pool = new FixedThreadPool(
  83       numberOfWorkers,
  84       './tests/worker-files/thread/testWorker.js'
  85     )
  86     expect(pool.opts.enableEvents).toBe(true)
  87     expect(pool.emitter).toBeDefined()
  88     expect(pool.opts.workerChoiceStrategy).toBe(
  89       WorkerChoiceStrategies.ROUND_ROBIN
  90     )
  91     expect(pool.opts.messageHandler).toBeUndefined()
  92     expect(pool.opts.errorHandler).toBeUndefined()
  93     expect(pool.opts.onlineHandler).toBeUndefined()
  94     expect(pool.opts.exitHandler).toBeUndefined()
  95     await pool.destroy()
  96     const testHandler = () => console.log('test handler executed')
  97     pool = new FixedThreadPool(
  98       numberOfWorkers,
  99       './tests/worker-files/thread/testWorker.js',
 100       {
 101         workerChoiceStrategy: WorkerChoiceStrategies.LESS_USED,
 102         enableEvents: false,
 103         messageHandler: testHandler,
 104         errorHandler: testHandler,
 105         onlineHandler: testHandler,
 106         exitHandler: testHandler
 107       }
 108     )
 109     expect(pool.opts.enableEvents).toBe(false)
 110     expect(pool.emitter).toBeUndefined()
 111     expect(pool.opts.workerChoiceStrategy).toBe(
 112       WorkerChoiceStrategies.LESS_USED
 113     )
 114     expect(pool.opts.messageHandler).toStrictEqual(testHandler)
 115     expect(pool.opts.errorHandler).toStrictEqual(testHandler)
 116     expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
 117     expect(pool.opts.exitHandler).toStrictEqual(testHandler)
 118     await pool.destroy()
 119   })
 120
 121   it('Simulate worker not found during getWorkerTasksUsage', async () => {
 122     const pool = new StubPoolWithRemoveAllWorker(
 123       numberOfWorkers,
 124       './tests/worker-files/cluster/testWorker.js',
 125       {
 126         errorHandler: e => console.error(e)
 127       }
 128     )
 129     // Simulate worker not found.
 130     pool.removeAllWorker()
 131     expect(() => pool.getWorkerTasksUsage()).toThrowError(
 132       workerNotFoundInPoolError
 133     )
 134     await pool.destroy()
 135   })
 136
 137   it('Verify that worker pool tasks usage are initialized', async () => {
 138     const pool = new FixedClusterPool(
 139       numberOfWorkers,
 140       './tests/worker-files/cluster/testWorker.js'
 141     )
 142     for (const value of pool.workers.values()) {
 143       expect(value.tasksUsage).toBeDefined()
 144       expect(value.tasksUsage.run).toBe(0)
 145       expect(value.tasksUsage.running).toBe(0)
 146       expect(value.tasksUsage.runTime).toBe(0)
 147       expect(value.tasksUsage.avgRunTime).toBe(0)
 148     }
 149     await pool.destroy()
 150   })
 151
 152   it('Verify that worker pool tasks usage are computed', async () => {
 153     const pool = new FixedClusterPool(
 154       numberOfWorkers,
 155       './tests/worker-files/cluster/testWorker.js'
 156     )
 157     const promises = []
 158     for (let i = 0; i < numberOfWorkers * 2; i++) {
 159       promises.push(pool.execute())
 160     }
 161     for (const value of pool.workers.values()) {
 162       expect(value.tasksUsage).toBeDefined()
 163       expect(value.tasksUsage.run).toBe(0)
 164       expect(value.tasksUsage.running).toBe(numberOfWorkers * 2)
 165       expect(value.tasksUsage.runTime).toBe(0)
 166       expect(value.tasksUsage.avgRunTime).toBe(0)
 167     }
 168     await Promise.all(promises)
 169     for (const value of pool.workers.values()) {
 170       expect(value.tasksUsage).toBeDefined()
 171       expect(value.tasksUsage.run).toBe(numberOfWorkers * 2)
 172       expect(value.tasksUsage.running).toBe(0)
 173       expect(value.tasksUsage.runTime).toBeGreaterThanOrEqual(0)
 174       expect(value.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0)
 175     }
 176     await pool.destroy()
 177   })
 178
 179   it('Verify that worker pool tasks usage are reset at worker choice strategy change', async () => {
 180     const pool = new DynamicThreadPool(
 181       numberOfWorkers,
 182       numberOfWorkers,
 183       './tests/worker-files/thread/testWorker.js'
 184     )
 185     const promises = []
 186     for (let i = 0; i < numberOfWorkers * 2; i++) {
 187       promises.push(pool.execute())
 188     }
 189     await Promise.all(promises)
 190     for (const value of pool.workers.values()) {
 191       expect(value.tasksUsage).toBeDefined()
 192       expect(value.tasksUsage.run).toBe(numberOfWorkers * 2)
 193       expect(value.tasksUsage.running).toBe(0)
 194       expect(value.tasksUsage.runTime).toBeGreaterThanOrEqual(0)
 195       expect(value.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0)
 196     }
 197     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 198     for (const value of pool.workers.values()) {
 199       expect(value.tasksUsage).toBeDefined()
 200       expect(value.tasksUsage.run).toBe(0)
 201       expect(value.tasksUsage.running).toBe(0)
 202       expect(value.tasksUsage.runTime).toBe(0)
 203       expect(value.tasksUsage.avgRunTime).toBe(0)
 204     }
 205     await pool.destroy()
 206   })
 207
 208   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 209     const pool = new FixedThreadPool(
 210       numberOfWorkers,
 211       './tests/worker-files/thread/testWorker.js'
 212     )
 213     const promises = []
 214     let poolBusy = 0
 215     pool.emitter.on('busy', () => poolBusy++)
 216     for (let i = 0; i < numberOfWorkers * 2; i++) {
 217       promises.push(pool.execute())
 218     }
 219     await Promise.all(promises)
 220     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 221     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 222     expect(poolBusy).toBe(numberOfWorkers + 1)
 223     await pool.destroy()
 224   })
 225 })