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