X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fabstract%2Fabstract-pool.test.js;h=e9c65dde91de251b6228c303861696a9f8629162;hb=41e3e08eb7fbcde0a69cea17e697aacb222990a6;hp=bedb1c6c42a9121697990121185c049c26ed87a4;hpb=c6f65aba7358678dc93cd04b573268564c5281d1;p=poolifier.git

diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js
index bedb1c6c..e9c65dde 100644
--- a/tests/pools/abstract/abstract-pool.test.js
+++ b/tests/pools/abstract/abstract-pool.test.js
@@ -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: Math.pow(numberOfWorkers, 2),
+        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: Math.pow(numberOfWorkers, 2),
+      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: Math.pow(numberOfWorkers, 2),
+      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,54 @@ describe('Abstract pool test suite', () => {
     )
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 1,
-      size: 4
+      size: Math.pow(numberOfWorkers, 2),
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
+    })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.tasksQueueBackPressureSize).toBe(
+        pool.opts.tasksQueueOptions.size
+      )
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
+    pool.setTasksQueueOptions({
+      concurrency: 2,
+      size: 2,
+      taskStealing: false,
+      tasksStealingOnBackPressure: false
     })
-    pool.setTasksQueueOptions({ concurrency: 2 })
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 2,
-      size: 4
+      size: 2,
+      taskStealing: false,
+      tasksStealingOnBackPressure: false
+    })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.tasksQueueBackPressureSize).toBe(
+        pool.opts.tasksQueueOptions.size
+      )
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
+    pool.setTasksQueueOptions({
+      concurrency: 1,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
+    })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 1,
+      size: Math.pow(numberOfWorkers, 2),
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.tasksQueueBackPressureSize).toBe(
+        pool.opts.tasksQueueOptions.size
+      )
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     expect(() =>
       pool.setTasksQueueOptions('invalidTasksQueueOptions')
     ).toThrowError(
@@ -696,11 +742,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 +767,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 +789,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 +810,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 +845,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 +857,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 +871,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 +886,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 +897,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 +939,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 +1108,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 +1146,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 +1183,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 +1223,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),