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