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 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_RECENTLY_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_RECENTLY_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 increaseWorkerRunningTasks', async () => {
 122     const pool = new StubPoolWithRemoveAllWorker(
 123       numberOfWorkers,
 124       './tests/worker-files/cluster/testWorker.js'
 125     )
 126     // Simulate worker not found.
 127     pool.removeAllWorker()
 128     expect(() => pool.increaseWorkerRunningTasks()).toThrowError(
 129       workerNotFoundInTasksUsageMapError
 130     )
 131     await pool.destroy()
 132   })
 133
 134   it('Simulate worker not found during decreaseWorkerRunningTasks', async () => {
 135     const pool = new StubPoolWithRemoveAllWorker(
 136       numberOfWorkers,
 137       './tests/worker-files/cluster/testWorker.js',
 138       {
 139         errorHandler: e => console.error(e)
 140       }
 141     )
 142     // Simulate worker not found.
 143     pool.removeAllWorker()
 144     expect(() => pool.decreaseWorkerRunningTasks()).toThrowError(
 145       workerNotFoundInTasksUsageMapError
 146     )
 147     await pool.destroy()
 148   })
 149
 150   it('Simulate worker not found during stepWorkerRunTasks', async () => {
 151     const pool = new StubPoolWithRemoveAllWorker(
 152       numberOfWorkers,
 153       './tests/worker-files/cluster/testWorker.js',
 154       {
 155         errorHandler: e => console.error(e)
 156       }
 157     )
 158     // Simulate worker not found.
 159     pool.removeAllWorker()
 160     expect(() => pool.stepWorkerRunTasks()).toThrowError(
 161       workerNotFoundInTasksUsageMapError
 162     )
 163     await pool.destroy()
 164   })
 165
 166   it('Simulate worker not found during updateWorkerTasksRunTime with strategy not requiring it', async () => {
 167     const pool = new StubPoolWithRemoveAllWorker(
 168       numberOfWorkers,
 169       './tests/worker-files/cluster/testWorker.js',
 170       {
 171         errorHandler: e => console.error(e)
 172       }
 173     )
 174     // Simulate worker not found.
 175     pool.removeAllWorker()
 176     expect(() => pool.updateWorkerTasksRunTime()).not.toThrowError()
 177     await pool.destroy()
 178   })
 179
 180   it('Simulate worker not found during updateWorkerTasksRunTime with strategy requiring it', async () => {
 181     const pool = new StubPoolWithRemoveAllWorker(
 182       numberOfWorkers,
 183       './tests/worker-files/cluster/testWorker.js',
 184       {
 185         workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE,
 186         errorHandler: e => console.error(e)
 187       }
 188     )
 189     // Simulate worker not found.
 190     pool.removeAllWorker()
 191     expect(() => pool.updateWorkerTasksRunTime()).toThrowError(
 192       workerNotFoundInTasksUsageMapError
 193     )
 194     await pool.destroy()
 195   })
 196
 197   it('Verify that worker pool tasks usage are initialized', async () => {
 198     const pool = new FixedClusterPool(
 199       numberOfWorkers,
 200       './tests/worker-files/cluster/testWorker.js'
 201     )
 202     for (const value of pool.workers.values()) {
 203       expect(value.tasksUsage).toBeDefined()
 204       expect(value.tasksUsage.run).toBe(0)
 205       expect(value.tasksUsage.running).toBe(0)
 206       expect(value.tasksUsage.runTime).toBe(0)
 207       expect(value.tasksUsage.avgRunTime).toBe(0)
 208     }
 209     await pool.destroy()
 210   })
 211
 212   it('Verify that worker pool tasks usage are computed', async () => {
 213     const pool = new FixedClusterPool(
 214       numberOfWorkers,
 215       './tests/worker-files/cluster/testWorker.js'
 216     )
 217     const promises = []
 218     for (let i = 0; i < numberOfWorkers * 2; i++) {
 219       promises.push(pool.execute())
 220     }
 221     for (const value of pool.workers.values()) {
 222       expect(value.tasksUsage).toBeDefined()
 223       expect(value.tasksUsage.run).toBe(0)
 224       expect(value.tasksUsage.running).toBe(numberOfWorkers * 2)
 225       expect(value.tasksUsage.runTime).toBe(0)
 226       expect(value.tasksUsage.avgRunTime).toBe(0)
 227     }
 228     await Promise.all(promises)
 229     for (const value of pool.workers.values()) {
 230       expect(value.tasksUsage).toBeDefined()
 231       expect(value.tasksUsage.run).toBe(numberOfWorkers * 2)
 232       expect(value.tasksUsage.running).toBe(0)
 233       expect(value.tasksUsage.runTime).toBeGreaterThanOrEqual(0)
 234       expect(value.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0)
 235     }
 236     await pool.destroy()
 237   })
 238
 239   it('Verify that worker pool tasks usage are reset at worker choice strategy change', async () => {
 240     const pool = new DynamicThreadPool(
 241       numberOfWorkers,
 242       numberOfWorkers,
 243       './tests/worker-files/thread/testWorker.js'
 244     )
 245     const promises = []
 246     for (let i = 0; i < numberOfWorkers * 2; i++) {
 247       promises.push(pool.execute())
 248     }
 249     await Promise.all(promises)
 250     for (const value of pool.workers.values()) {
 251       expect(value.tasksUsage).toBeDefined()
 252       expect(value.tasksUsage.run).toBe(numberOfWorkers * 2)
 253       expect(value.tasksUsage.running).toBe(0)
 254       expect(value.tasksUsage.runTime).toBeGreaterThanOrEqual(0)
 255       expect(value.tasksUsage.avgRunTime).toBeGreaterThanOrEqual(0)
 256     }
 257     pool.setWorkerChoiceStrategy(WorkerChoiceStrategies.FAIR_SHARE)
 258     for (const value of pool.workers.values()) {
 259       expect(value.tasksUsage).toBeDefined()
 260       expect(value.tasksUsage.run).toBe(0)
 261       expect(value.tasksUsage.running).toBe(0)
 262       expect(value.tasksUsage.runTime).toBe(0)
 263       expect(value.tasksUsage.avgRunTime).toBe(0)
 264     }
 265     await pool.destroy()
 266   })
 267
 268   it("Verify that pool event emitter 'busy' event can register a callback", async () => {
 269     const pool = new FixedThreadPool(
 270       numberOfWorkers,
 271       './tests/worker-files/thread/testWorker.js'
 272     )
 273     const promises = []
 274     let poolBusy = 0
 275     pool.emitter.on('busy', () => poolBusy++)
 276     for (let i = 0; i < numberOfWorkers * 2; i++) {
 277       promises.push(pool.execute())
 278     }
 279     await Promise.all(promises)
 280     // The `busy` event is triggered when the number of submitted tasks at once reach the number of fixed pool workers.
 281     // So in total numberOfWorkers + 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the fixed pool.
 282     expect(poolBusy).toBe(numberOfWorkers + 1)
 283     await pool.destroy()
 284   })
 285 })