test: add test for backPressure event emission

[poolifier.git] / tests / pools / abstract / abstract-pool.test.js

diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js
index ec552e986003d859f38fe0557d6d8727c23d7ceb..d1c58ddbb73658f495df83f345c013f16c631132 100644 (file)
--- a/tests/pools/abstract/abstract-pool.test.js
+++ b/tests/pools/abstract/abstract-pool.test.js
@@ -1,4 +1,5 @@
 const { expect } = require('expect')
+const sinon = require('sinon')
 const {
   DynamicClusterPool,
   DynamicThreadPool,
@@ -242,7 +243,9 @@ describe('Abstract pool test suite', () => {
             workerChoiceStrategy: 'invalidStrategy'
           }
         )
-    ).toThrowError("Invalid worker choice strategy 'invalidStrategy'")
+    ).toThrowError(
+      new Error("Invalid worker choice strategy 'invalidStrategy'")
+    )
     expect(
       () =>
         new FixedThreadPool(
@@ -253,7 +256,9 @@ describe('Abstract pool test suite', () => {
           }
         )
     ).toThrowError(
-      'Invalid worker choice strategy options: must have a weight for each worker node'
+      new Error(
+        'Invalid worker choice strategy options: must have a weight for each worker node'
+      )
     )
     expect(
       () =>
@@ -265,7 +270,9 @@ describe('Abstract pool test suite', () => {
           }
         )
     ).toThrowError(
-      "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
+      new Error(
+        "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
+      )
     )
     expect(
       () =>
@@ -277,7 +284,11 @@ describe('Abstract pool test suite', () => {
             tasksQueueOptions: { concurrency: 0 }
           }
         )
-    ).toThrowError("Invalid worker tasks concurrency '0'")
+    ).toThrowError(
+      new TypeError(
+        'Invalid worker tasks concurrency: 0 is a negative integer or zero'
+      )
+    )
     expect(
       () =>
         new FixedThreadPool(
@@ -288,7 +299,9 @@ describe('Abstract pool test suite', () => {
             tasksQueueOptions: 'invalidTasksQueueOptions'
           }
         )
-    ).toThrowError('Invalid tasks queue options: must be a plain object')
+    ).toThrowError(
+      new TypeError('Invalid tasks queue options: must be a plain object')
+    )
     expect(
       () =>
         new FixedThreadPool(
@@ -299,7 +312,9 @@ describe('Abstract pool test suite', () => {
             tasksQueueOptions: { concurrency: 0.2 }
           }
         )
-    ).toThrowError('Invalid worker tasks concurrency: must be an integer')
+    ).toThrowError(
+      new TypeError('Invalid worker tasks concurrency: must be an integer')
+    )
   })
 
   it('Verify that pool worker choice strategy options can be set', async () => {
@@ -439,17 +454,23 @@ describe('Abstract pool test suite', () => {
     expect(() =>
       pool.setWorkerChoiceStrategyOptions('invalidWorkerChoiceStrategyOptions')
     ).toThrowError(
-      'Invalid worker choice strategy options: must be a plain object'
+      new TypeError(
+        'Invalid worker choice strategy options: must be a plain object'
+      )
     )
     expect(() =>
       pool.setWorkerChoiceStrategyOptions({ weights: {} })
     ).toThrowError(
-      'Invalid worker choice strategy options: must have a weight for each worker node'
+      new Error(
+        'Invalid worker choice strategy options: must have a weight for each worker node'
+      )
     )
     expect(() =>
       pool.setWorkerChoiceStrategyOptions({ measurement: 'invalidMeasurement' })
     ).toThrowError(
-      "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
+      new Error(
+        "Invalid worker choice strategy options: invalid measurement 'invalidMeasurement'"
+      )
     )
     await pool.destroy()
   })
@@ -484,12 +505,21 @@ describe('Abstract pool test suite', () => {
     expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
     expect(() =>
       pool.setTasksQueueOptions('invalidTasksQueueOptions')
-    ).toThrowError('Invalid tasks queue options: must be a plain object')
+    ).toThrowError(
+      new TypeError('Invalid tasks queue options: must be a plain object')
+    )
     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
-      "Invalid worker tasks concurrency '0'"
+      new Error(
+        'Invalid worker tasks concurrency: 0 is a negative integer or zero'
+      )
+    )
+    expect(() => pool.setTasksQueueOptions({ concurrency: -1 })).toThrowError(
+      new Error(
+        'Invalid worker tasks concurrency: -1 is a negative integer or zero'
+      )
     )
     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
-      'Invalid worker tasks concurrency: must be an integer'
+      new TypeError('Invalid worker tasks concurrency: must be an integer')
     )
     await pool.destroy()
   })
@@ -726,7 +756,9 @@ describe('Abstract pool test suite', () => {
         }
       })
       expect(workerNode.usage.tasks.executed).toBeGreaterThan(0)
-      expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(maxMultiplier)
+      expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(
+        numberOfWorkers * maxMultiplier
+      )
       expect(workerNode.usage.runTime.history.length).toBe(0)
       expect(workerNode.usage.waitTime.history.length).toBe(0)
       expect(workerNode.usage.elu.idle.history.length).toBe(0)
@@ -851,9 +883,7 @@ describe('Abstract pool test suite', () => {
       promises.add(pool.execute())
     }
     await Promise.all(promises)
-    // The `full` event is triggered when the number of submitted tasks at once reach the maximum number of workers in the dynamic pool.
-    // So in total numberOfWorkers * 2 - 1 times for a loop submitting up to numberOfWorkers * 2 tasks to the dynamic pool with min = (max = numberOfWorkers) / 2.
-    expect(poolFull).toBe(numberOfWorkers * 2 - 1)
+    expect(poolFull).toBe(1)
     expect(poolInfo).toStrictEqual({
       version,
       type: PoolTypes.dynamic,
@@ -872,6 +902,49 @@ describe('Abstract pool test suite', () => {
     await pool.destroy()
   })
 
+  it("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
+    const pool = new FixedThreadPool(
+      numberOfWorkers,
+      './tests/worker-files/thread/testWorker.js',
+      {
+        enableTasksQueue: true
+      }
+    )
+    sinon.stub(pool, 'hasBackPressure').returns(true)
+    const promises = new Set()
+    let poolBackPressure = 0
+    let poolInfo
+    pool.emitter.on(PoolEvents.backPressure, (info) => {
+      ++poolBackPressure
+      poolInfo = info
+    })
+    for (let i = 0; i < numberOfWorkers * 2; i++) {
+      promises.add(pool.execute())
+    }
+    await Promise.all(promises)
+    expect(poolBackPressure).toBe(2)
+    expect(poolInfo).toStrictEqual({
+      version,
+      type: PoolTypes.fixed,
+      worker: WorkerTypes.thread,
+      ready: expect.any(Boolean),
+      strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      minSize: expect.any(Number),
+      maxSize: expect.any(Number),
+      workerNodes: expect.any(Number),
+      idleWorkerNodes: expect.any(Number),
+      busyWorkerNodes: expect.any(Number),
+      executedTasks: expect.any(Number),
+      executingTasks: expect.any(Number),
+      maxQueuedTasks: expect.any(Number),
+      queuedTasks: expect.any(Number),
+      backPressure: true,
+      failedTasks: expect.any(Number)
+    })
+    expect(pool.hasBackPressure.called).toBe(true)
+    await pool.destroy()
+  })
+
   it('Verify that listTaskFunctions() is working', async () => {
     const dynamicThreadPool = new DynamicThreadPool(
       Math.floor(numberOfWorkers / 2),