repositories / poolifier.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
fix: fix task stealing related options handling at runtime
[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 a206948f6d19b168f910b55f98ba09e488266204..aeede25d7149a9c5918626ae9e860e7733cea4b4 100644 (file)
--- a/tests/pools/abstract/abstract-pool.test.js
+++ b/tests/pools/abstract/abstract-pool.test.js
@@ -1,4 +1,4 @@
-const { EventEmitter } = require('events')
+const { EventEmitter } = require('node:events')
 const { expect } = require('expect')
 const sinon = require('sinon')
 const {
@@ -16,6 +16,7 @@ const { Deque } = require('../../../lib/deque')
 const { DEFAULT_TASK_NAME } = require('../../../lib/utils')
 const { version } = require('../../../package.json')
 const { waitPoolEvents } = require('../../test-utils')
+const { WorkerNode } = require('../../../lib/pools/worker-node')
 
 describe('Abstract pool test suite', () => {
   const numberOfWorkers = 2
@@ -180,18 +181,18 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/thread/testWorker.js'
     )
     expect(pool.emitter).toBeInstanceOf(EventEmitter)
-    expect(pool.opts.enableEvents).toBe(true)
-    expect(pool.opts.restartWorkerOnError).toBe(true)
-    expect(pool.opts.enableTasksQueue).toBe(false)
-    expect(pool.opts.tasksQueueOptions).toBeUndefined()
-    expect(pool.opts.workerChoiceStrategy).toBe(
-      WorkerChoiceStrategies.ROUND_ROBIN
-    )
-    expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
-      retries: 6,
-      runTime: { median: false },
-      waitTime: { median: false },
-      elu: { median: false }
+    expect(pool.opts).toStrictEqual({
+      startWorkers: true,
+      enableEvents: true,
+      restartWorkerOnError: true,
+      enableTasksQueue: false,
+      workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      workerChoiceStrategyOptions: {
+        retries: 6,
+        runTime: { median: false },
+        waitTime: { median: false },
+        elu: { median: false }
+      }
     })
     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
       retries: 6,
@@ -208,10 +209,6 @@ describe('Abstract pool test suite', () => {
         elu: { median: false }
       })
     }
-    expect(pool.opts.messageHandler).toBeUndefined()
-    expect(pool.opts.errorHandler).toBeUndefined()
-    expect(pool.opts.onlineHandler).toBeUndefined()
-    expect(pool.opts.exitHandler).toBeUndefined()
     await pool.destroy()
     const testHandler = () => console.info('test handler executed')
     pool = new FixedThreadPool(
@@ -234,22 +231,29 @@ describe('Abstract pool test suite', () => {
       }
     )
     expect(pool.emitter).toBeUndefined()
-    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,
-      size: 4
-    })
-    expect(pool.opts.workerChoiceStrategy).toBe(
-      WorkerChoiceStrategies.LEAST_USED
-    )
-    expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
-      retries: 6,
-      runTime: { median: true },
-      waitTime: { median: false },
-      elu: { median: false },
-      weights: { 0: 300, 1: 200 }
+    expect(pool.opts).toStrictEqual({
+      startWorkers: true,
+      enableEvents: false,
+      restartWorkerOnError: false,
+      enableTasksQueue: true,
+      tasksQueueOptions: {
+        concurrency: 2,
+        size: 4,
+        taskStealing: true,
+        tasksStealingOnBackPressure: true
+      },
+      workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
+      workerChoiceStrategyOptions: {
+        retries: 6,
+        runTime: { median: true },
+        waitTime: { median: false },
+        elu: { median: false },
+        weights: { 0: 300, 1: 200 }
+      },
+      onlineHandler: testHandler,
+      messageHandler: testHandler,
+      errorHandler: testHandler,
+      exitHandler: testHandler
     })
     expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
       retries: 6,
@@ -268,10 +272,6 @@ describe('Abstract pool test suite', () => {
         weights: { 0: 300, 1: 200 }
       })
     }
-    expect(pool.opts.messageHandler).toStrictEqual(testHandler)
-    expect(pool.opts.errorHandler).toStrictEqual(testHandler)
-    expect(pool.opts.onlineHandler).toStrictEqual(testHandler)
-    expect(pool.opts.exitHandler).toStrictEqual(testHandler)
     await pool.destroy()
   })
 
@@ -404,21 +404,6 @@ describe('Abstract pool test suite', () => {
     ).toThrowError(
       new TypeError('Invalid worker node tasks concurrency: must be an integer')
     )
-    expect(
-      () =>
-        new FixedThreadPool(
-          numberOfWorkers,
-          './tests/worker-files/thread/testWorker.js',
-          {
-            enableTasksQueue: true,
-            tasksQueueOptions: { queueMaxSize: 2 }
-          }
-        )
-    ).toThrowError(
-      new Error(
-        'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
-      )
-    )
     expect(
       () =>
         new FixedThreadPool(
@@ -645,21 +630,41 @@ describe('Abstract pool test suite', () => {
     )
     expect(pool.opts.enableTasksQueue).toBe(false)
     expect(pool.opts.tasksQueueOptions).toBeUndefined()
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
     pool.enableTasksQueue(true)
     expect(pool.opts.enableTasksQueue).toBe(true)
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 1,
-      size: 4
+      size: 4,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     pool.enableTasksQueue(true, { concurrency: 2 })
     expect(pool.opts.enableTasksQueue).toBe(true)
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 2,
-      size: 4
+      size: 4,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     pool.enableTasksQueue(false)
     expect(pool.opts.enableTasksQueue).toBe(false)
     expect(pool.opts.tasksQueueOptions).toBeUndefined()
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
     await pool.destroy()
   })
 
@@ -671,13 +676,44 @@ describe('Abstract pool test suite', () => {
     )
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 1,
-      size: 4
+      size: 4,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
+    })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
+    pool.setTasksQueueOptions({
+      concurrency: 2,
+      taskStealing: false,
+      tasksStealingOnBackPressure: false
     })
-    pool.setTasksQueueOptions({ concurrency: 2 })
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 2,
-      size: 4
+      size: 4,
+      taskStealing: false,
+      tasksStealingOnBackPressure: false
+    })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
+    pool.setTasksQueueOptions({
+      concurrency: 1,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
+    })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 1,
+      size: 4,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     expect(() =>
       pool.setTasksQueueOptions('invalidTasksQueueOptions')
     ).toThrowError(
@@ -696,11 +732,6 @@ describe('Abstract pool test suite', () => {
     expect(() => pool.setTasksQueueOptions({ concurrency: 0.2 })).toThrowError(
       new TypeError('Invalid worker node tasks concurrency: must be an integer')
     )
-    expect(() => pool.setTasksQueueOptions({ queueMaxSize: 2 })).toThrowError(
-      new Error(
-        'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead'
-      )
-    )
     expect(() => pool.setTasksQueueOptions({ size: 0 })).toThrowError(
       new RangeError(
         'Invalid worker node tasks queue size: 0 is a negative integer or zero'
@@ -726,6 +757,7 @@ describe('Abstract pool test suite', () => {
       version,
       type: PoolTypes.fixed,
       worker: WorkerTypes.thread,
+      started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
       minSize: numberOfWorkers,
@@ -747,6 +779,7 @@ describe('Abstract pool test suite', () => {
       version,
       type: PoolTypes.dynamic,
       worker: WorkerTypes.cluster,
+      started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
       minSize: Math.floor(numberOfWorkers / 2),
@@ -767,6 +800,7 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/cluster/testWorker.js'
     )
     for (const workerNode of pool.workerNodes) {
+      expect(workerNode).toBeInstanceOf(WorkerNode)
       expect(workerNode.usage).toStrictEqual({
         tasks: {
           executed: 0,
@@ -801,7 +835,7 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/cluster/testWorker.js'
     )
     for (const workerNode of pool.workerNodes) {
-      expect(workerNode.tasksQueue).toBeDefined()
+      expect(workerNode).toBeInstanceOf(WorkerNode)
       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
       expect(workerNode.tasksQueue.size).toBe(0)
       expect(workerNode.tasksQueue.maxSize).toBe(0)
@@ -813,7 +847,7 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/thread/testWorker.js'
     )
     for (const workerNode of pool.workerNodes) {
-      expect(workerNode.tasksQueue).toBeDefined()
+      expect(workerNode).toBeInstanceOf(WorkerNode)
       expect(workerNode.tasksQueue).toBeInstanceOf(Deque)
       expect(workerNode.tasksQueue.size).toBe(0)
       expect(workerNode.tasksQueue.maxSize).toBe(0)
@@ -827,6 +861,7 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/cluster/testWorker.js'
     )
     for (const workerNode of pool.workerNodes) {
+      expect(workerNode).toBeInstanceOf(WorkerNode)
       expect(workerNode.info).toStrictEqual({
         id: expect.any(Number),
         type: WorkerTypes.cluster,
@@ -841,6 +876,7 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/thread/testWorker.js'
     )
     for (const workerNode of pool.workerNodes) {
+      expect(workerNode).toBeInstanceOf(WorkerNode)
       expect(workerNode.info).toStrictEqual({
         id: expect.any(Number),
         type: WorkerTypes.thread,
@@ -851,6 +887,30 @@ describe('Abstract pool test suite', () => {
     await pool.destroy()
   })
 
+  it('Verify that pool can be started after initialization', async () => {
+    const pool = new FixedClusterPool(
+      numberOfWorkers,
+      './tests/worker-files/cluster/testWorker.js',
+      {
+        startWorkers: false
+      }
+    )
+    expect(pool.info.started).toBe(false)
+    expect(pool.info.ready).toBe(false)
+    expect(pool.workerNodes).toStrictEqual([])
+    await expect(pool.execute()).rejects.toThrowError(
+      new Error('Cannot execute a task on not started pool')
+    )
+    pool.start()
+    expect(pool.info.started).toBe(true)
+    expect(pool.info.ready).toBe(true)
+    expect(pool.workerNodes.length).toBe(numberOfWorkers)
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode).toBeInstanceOf(WorkerNode)
+    }
+    await pool.destroy()
+  })
+
   it('Verify that pool execute() arguments are checked', async () => {
     const pool = new FixedClusterPool(
       numberOfWorkers,
@@ -869,8 +929,8 @@ describe('Abstract pool test suite', () => {
       "Task function 'unknown' not found"
     )
     await pool.destroy()
-    await expect(pool.execute(undefined, undefined, {})).rejects.toThrowError(
-      new Error('Cannot execute a task on destroyed pool')
+    await expect(pool.execute()).rejects.toThrowError(
+      new Error('Cannot execute a task on not started pool')
     )
   })
 
@@ -1038,6 +1098,7 @@ describe('Abstract pool test suite', () => {
       version,
       type: PoolTypes.dynamic,
       worker: WorkerTypes.cluster,
+      started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
       minSize: expect.any(Number),
@@ -1075,7 +1136,8 @@ describe('Abstract pool test suite', () => {
       version,
       type: PoolTypes.fixed,
       worker: WorkerTypes.thread,
-      ready: expect.any(Boolean),
+      started: true,
+      ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
@@ -1111,7 +1173,8 @@ describe('Abstract pool test suite', () => {
       version,
       type: PoolTypes.dynamic,
       worker: WorkerTypes.thread,
-      ready: expect.any(Boolean),
+      started: true,
+      ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
@@ -1150,7 +1213,8 @@ describe('Abstract pool test suite', () => {
       version,
       type: PoolTypes.fixed,
       worker: WorkerTypes.thread,
-      ready: expect.any(Boolean),
+      started: true,
+      ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
Fast and small Node.js thread pool implementation mirror - https://github.com/poolifier/poolifier