feat: continuous task stealing

[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 4fd76d03585bd7f76966591ff0f54bfc2300594f..933e1f944f863bdbcdb4110c1d75359e8f571b4c 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,
@@ -10,7 +11,7 @@ const {
   WorkerTypes
 } = require('../../../lib')
 const { CircularArray } = require('../../../lib/circular-array')
-const { Queue } = require('../../../lib/queue')
+const { Deque } = require('../../../lib/deque')
 const { version } = require('../../../package.json')
 const { waitPoolEvents } = require('../../test-utils')
 
@@ -22,6 +23,10 @@ describe('Abstract pool test suite', () => {
     }
   }
 
+  afterEach(() => {
+    sinon.restore()
+  })
+
   it('Simulate pool creation from a non main thread/process', () => {
     expect(
       () =>
@@ -33,7 +38,9 @@ describe('Abstract pool test suite', () => {
           }
         )
     ).toThrowError(
-      'Cannot start a pool from a worker with the same type as the pool'
+      new Error(
+        'Cannot start a pool from a worker with the same type as the pool'
+      )
     )
   })
 
@@ -60,7 +67,9 @@ describe('Abstract pool test suite', () => {
 
   it('Verify that numberOfWorkers is checked', () => {
     expect(() => new FixedThreadPool()).toThrowError(
-      'Cannot instantiate a pool without specifying the number of workers'
+      new Error(
+        'Cannot instantiate a pool without specifying the number of workers'
+      )
     )
   })
 
@@ -207,7 +216,10 @@ describe('Abstract pool test suite', () => {
     expect(pool.opts.enableEvents).toBe(false)
     expect(pool.opts.restartWorkerOnError).toBe(false)
     expect(pool.opts.enableTasksQueue).toBe(true)
-    expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 2,
+      size: 4
+    })
     expect(pool.opts.workerChoiceStrategy).toBe(
       WorkerChoiceStrategies.LEAST_USED
     )
@@ -284,8 +296,8 @@ describe('Abstract pool test suite', () => {
           }
         )
     ).toThrowError(
-      new TypeError(
-        'Invalid worker tasks concurrency: 0 is a negative integer or zero'
+      new RangeError(
+        'Invalid worker node tasks concurrency: 0 is a negative integer or zero'
       )
     )
     expect(
@@ -312,7 +324,7 @@ describe('Abstract pool test suite', () => {
           }
         )
     ).toThrowError(
-      new TypeError('Invalid worker tasks concurrency: must be an integer')
+      new TypeError('Invalid worker node tasks concurrency: must be an integer')
     )
   })
 
@@ -483,10 +495,16 @@ describe('Abstract pool test suite', () => {
     expect(pool.opts.tasksQueueOptions).toBeUndefined()
     pool.enableTasksQueue(true)
     expect(pool.opts.enableTasksQueue).toBe(true)
-    expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 1,
+      size: 4
+    })
     pool.enableTasksQueue(true, { concurrency: 2 })
     expect(pool.opts.enableTasksQueue).toBe(true)
-    expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 2,
+      size: 4
+    })
     pool.enableTasksQueue(false)
     expect(pool.opts.enableTasksQueue).toBe(false)
     expect(pool.opts.tasksQueueOptions).toBeUndefined()
@@ -499,26 +517,47 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/thread/testWorker.js',
       { enableTasksQueue: true }
     )
-    expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 1 })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 1,
+      size: 4
+    })
     pool.setTasksQueueOptions({ concurrency: 2 })
-    expect(pool.opts.tasksQueueOptions).toStrictEqual({ concurrency: 2 })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 2,
+      size: 4
+    })
     expect(() =>
       pool.setTasksQueueOptions('invalidTasksQueueOptions')
     ).toThrowError(
       new TypeError('Invalid tasks queue options: must be a plain object')
     )
     expect(() => pool.setTasksQueueOptions({ concurrency: 0 })).toThrowError(
-      new Error(
-        'Invalid worker tasks concurrency: 0 is a negative integer or zero'
+      new RangeError(
+        'Invalid worker node 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'
+      new RangeError(
+        'Invalid worker node tasks concurrency: -1 is a negative integer or zero'
       )
     )
     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
-      new TypeError('Invalid worker tasks concurrency: must be an integer')
+      new TypeError('Invalid worker node tasks concurrency: must be an integer')
+    )
+    expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
+      new RangeError(
+        'Invalid worker node tasks queue max size: 0 is a negative integer or zero'
+      )
+    )
+    expect(() => pool.setTasksQueueOptions({ size: -1 })).toThrowError(
+      new RangeError(
+        'Invalid worker node tasks queue max size: -1 is a negative integer or zero'
+      )
+    )
+    expect(() => pool.setTasksQueueOptions({ size: 0.2 })).toThrowError(
+      new TypeError(
+        'Invalid worker node tasks queue max size: must be an integer'
+      )
     )
     await pool.destroy()
   })
@@ -579,6 +618,7 @@ describe('Abstract pool test suite', () => {
           executing: 0,
           queued: 0,
           maxQueued: 0,
+          stolen: 0,
           failed: 0
         },
         runTime: {
@@ -607,7 +647,7 @@ describe('Abstract pool test suite', () => {
     )
     for (const workerNode of pool.workerNodes) {
       expect(workerNode.tasksQueue).toBeDefined()
-      expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
+      expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
       expect(workerNode.tasksQueue.size).toBe(0)
       expect(workerNode.tasksQueue.maxSize).toBe(0)
     }
@@ -619,7 +659,7 @@ describe('Abstract pool test suite', () => {
     )
     for (const workerNode of pool.workerNodes) {
       expect(workerNode.tasksQueue).toBeDefined()
-      expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
+      expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
       expect(workerNode.tasksQueue.size).toBe(0)
       expect(workerNode.tasksQueue.maxSize).toBe(0)
     }
@@ -671,6 +711,7 @@ describe('Abstract pool test suite', () => {
           executing: maxMultiplier,
           queued: 0,
           maxQueued: 0,
+          stolen: 0,
           failed: 0
         },
         runTime: {
@@ -697,6 +738,7 @@ describe('Abstract pool test suite', () => {
           executing: 0,
           queued: 0,
           maxQueued: 0,
+          stolen: 0,
           failed: 0
         },
         runTime: {
@@ -737,6 +779,7 @@ describe('Abstract pool test suite', () => {
           executing: 0,
           queued: 0,
           maxQueued: 0,
+          stolen: 0,
           failed: 0
         },
         runTime: {
@@ -755,7 +798,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)
@@ -769,6 +814,7 @@ describe('Abstract pool test suite', () => {
           executing: 0,
           queued: 0,
           maxQueued: 0,
+          stolen: 0,
           failed: 0
         },
         runTime: {
@@ -899,15 +945,15 @@ describe('Abstract pool test suite', () => {
     await pool.destroy()
   })
 
-  it.skip("Verify that pool event emitter 'backPressure' event can register a callback", async () => {
-    const pool = new DynamicThreadPool(
-      Math.floor(numberOfWorkers / 2),
+  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
@@ -915,14 +961,14 @@ describe('Abstract pool test suite', () => {
       ++poolBackPressure
       poolInfo = info
     })
-    for (let i = 0; i < Math.pow(numberOfWorkers, 2); i++) {
+    for (let i = 0; i < numberOfWorkers + 1; i++) {
       promises.add(pool.execute())
     }
     await Promise.all(promises)
     expect(poolBackPressure).toBe(1)
     expect(poolInfo).toStrictEqual({
       version,
-      type: PoolTypes.dynamic,
+      type: PoolTypes.fixed,
       worker: WorkerTypes.thread,
       ready: expect.any(Boolean),
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
@@ -933,8 +979,13 @@ describe('Abstract pool test suite', () => {
       busyWorkerNodes: expect.any(Number),
       executedTasks: expect.any(Number),
       executingTasks: expect.any(Number),
+      maxQueuedTasks: expect.any(Number),
+      queuedTasks: expect.any(Number),
+      backPressure: true,
+      stolenTasks: expect.any(Number),
       failedTasks: expect.any(Number)
     })
+    expect(pool.hasBackPressure.called).toBe(true)
     await pool.destroy()
   })
 
@@ -995,7 +1046,8 @@ describe('Abstract pool test suite', () => {
             executed: expect.any(Number),
             executing: expect.any(Number),
             failed: 0,
-            queued: 0
+            queued: 0,
+            stolen: 0
           },
           runTime: {
             history: expect.any(CircularArray)