X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fabstract-pool.test.mjs;h=89cc8995033c12c55f64e4cf8f1c7b1d59d66f42;hb=773881ea2c1b9aa8cd9698cc3ccc1978609ad98f;hp=1b80aa4cdeb92efd76ed0c5f4e56e7357342c8a0;hpb=568d0075825104b5a5ccc38dea2beeb55b55f3d0;p=poolifier.git

diff --git a/tests/pools/abstract-pool.test.mjs b/tests/pools/abstract-pool.test.mjs
index 1b80aa4c..89cc8995 100644
--- a/tests/pools/abstract-pool.test.mjs
+++ b/tests/pools/abstract-pool.test.mjs
@@ -1,10 +1,14 @@
+import { createHook, executionAsyncId } from 'node:async_hooks'
 import { EventEmitterAsyncResource } from 'node:events'
-import { dirname, join } from 'node:path'
 import { readFileSync } from 'node:fs'
+import { dirname, join } from 'node:path'
 import { fileURLToPath } from 'node:url'
-import { createHook, executionAsyncId } from 'node:async_hooks'
+
 import { expect } from 'expect'
 import { restore, stub } from 'sinon'
+
+import { CircularArray } from '../../lib/circular-array.cjs'
+import { Deque } from '../../lib/deque.cjs'
 import {
   DynamicClusterPool,
   DynamicThreadPool,
@@ -14,12 +18,10 @@ import {
   PoolTypes,
   WorkerChoiceStrategies,
   WorkerTypes
-} from '../../lib/index.js'
-import { CircularArray } from '../../lib/circular-array.js'
-import { Deque } from '../../lib/deque.js'
-import { DEFAULT_TASK_NAME } from '../../lib/utils.js'
-import { waitPoolEvents } from '../test-utils.js'
-import { WorkerNode } from '../../lib/pools/worker-node.js'
+} from '../../lib/index.cjs'
+import { WorkerNode } from '../../lib/pools/worker-node.cjs'
+import { DEFAULT_TASK_NAME } from '../../lib/utils.cjs'
+import { waitPoolEvents } from '../test-utils.cjs'
 
 describe('Abstract pool test suite', () => {
   const version = JSON.parse(
@@ -105,7 +107,7 @@ describe('Abstract pool test suite', () => {
   it('Verify that a negative number of workers is checked', () => {
     expect(
       () =>
-        new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.js')
+        new FixedClusterPool(-1, './tests/worker-files/cluster/testWorker.cjs')
     ).toThrow(
       new RangeError(
         'Cannot instantiate a pool with a negative number of workers'
@@ -124,13 +126,29 @@ describe('Abstract pool test suite', () => {
     )
   })
 
+  it('Verify that pool arguments number and pool type are checked', () => {
+    expect(
+      () =>
+        new FixedThreadPool(
+          numberOfWorkers,
+          './tests/worker-files/thread/testWorker.mjs',
+          undefined,
+          numberOfWorkers * 2
+        )
+    ).toThrow(
+      new Error(
+        'Cannot instantiate a fixed pool with a maximum number of workers specified at initialization'
+      )
+    )
+  })
+
   it('Verify that dynamic pool sizing is checked', () => {
     expect(
       () =>
         new DynamicClusterPool(
           1,
           undefined,
-          './tests/worker-files/cluster/testWorker.js'
+          './tests/worker-files/cluster/testWorker.cjs'
         )
     ).toThrow(
       new TypeError(
@@ -154,7 +172,7 @@ describe('Abstract pool test suite', () => {
         new DynamicClusterPool(
           0,
           0.5,
-          './tests/worker-files/cluster/testWorker.js'
+          './tests/worker-files/cluster/testWorker.cjs'
         )
     ).toThrow(
       new TypeError(
@@ -190,7 +208,7 @@ describe('Abstract pool test suite', () => {
         new DynamicClusterPool(
           1,
           1,
-          './tests/worker-files/cluster/testWorker.js'
+          './tests/worker-files/cluster/testWorker.cjs'
         )
     ).toThrow(
       new RangeError(
@@ -205,32 +223,24 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/thread/testWorker.mjs'
     )
     expect(pool.emitter).toBeInstanceOf(EventEmitterAsyncResource)
+    expect(pool.emitter.eventNames()).toStrictEqual([])
     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,
-      runTime: { median: false },
-      waitTime: { median: false },
-      elu: { median: false }
+      workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN
     })
     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
       .workerChoiceStrategies) {
       expect(workerChoiceStrategy.opts).toStrictEqual({
-        retries: 6,
         runTime: { median: false },
         waitTime: { median: false },
-        elu: { median: false }
+        elu: { median: false },
+        weights: expect.objectContaining({
+          0: expect.any(Number),
+          [pool.info.maxSize - 1]: expect.any(Number)
+        })
       })
     }
     await pool.destroy()
@@ -269,10 +279,7 @@ describe('Abstract pool test suite', () => {
       },
       workerChoiceStrategy: WorkerChoiceStrategies.LEAST_USED,
       workerChoiceStrategyOptions: {
-        retries: 6,
         runTime: { median: true },
-        waitTime: { median: false },
-        elu: { median: false },
         weights: { 0: 300, 1: 200 }
       },
       onlineHandler: testHandler,
@@ -280,17 +287,9 @@ describe('Abstract pool test suite', () => {
       errorHandler: testHandler,
       exitHandler: testHandler
     })
-    expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
-      retries: 6,
-      runTime: { median: true },
-      waitTime: { median: false },
-      elu: { median: false },
-      weights: { 0: 300, 1: 200 }
-    })
     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
       .workerChoiceStrategies) {
       expect(workerChoiceStrategy.opts).toStrictEqual({
-        retries: 6,
         runTime: { median: true },
         waitTime: { median: false },
         elu: { median: false },
@@ -311,38 +310,6 @@ describe('Abstract pool test suite', () => {
           }
         )
     ).toThrow(new Error("Invalid worker choice strategy 'invalidStrategy'"))
-    expect(
-      () =>
-        new FixedThreadPool(
-          numberOfWorkers,
-          './tests/worker-files/thread/testWorker.mjs',
-          {
-            workerChoiceStrategyOptions: {
-              retries: 'invalidChoiceRetries'
-            }
-          }
-        )
-    ).toThrow(
-      new TypeError(
-        'Invalid worker choice strategy options: retries must be an integer'
-      )
-    )
-    expect(
-      () =>
-        new FixedThreadPool(
-          numberOfWorkers,
-          './tests/worker-files/thread/testWorker.mjs',
-          {
-            workerChoiceStrategyOptions: {
-              retries: -1
-            }
-          }
-        )
-    ).toThrow(
-      new RangeError(
-        "Invalid worker choice strategy options: retries '-1' must be greater or equal than zero"
-      )
-    )
     expect(
       () =>
         new FixedThreadPool(
@@ -478,25 +445,17 @@ describe('Abstract pool test suite', () => {
       './tests/worker-files/thread/testWorker.mjs',
       { workerChoiceStrategy: WorkerChoiceStrategies.FAIR_SHARE }
     )
-    expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
-      retries: 6,
-      runTime: { median: false },
-      waitTime: { median: false },
-      elu: { median: false }
-    })
-    expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
-      retries: 6,
-      runTime: { median: false },
-      waitTime: { median: false },
-      elu: { median: false }
-    })
+    expect(pool.opts.workerChoiceStrategyOptions).toBeUndefined()
     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
       .workerChoiceStrategies) {
       expect(workerChoiceStrategy.opts).toStrictEqual({
-        retries: 6,
         runTime: { median: false },
         waitTime: { median: false },
-        elu: { median: false }
+        elu: { median: false },
+        weights: expect.objectContaining({
+          0: expect.any(Number),
+          [pool.info.maxSize - 1]: expect.any(Number)
+        })
       })
     }
     expect(
@@ -523,24 +482,19 @@ describe('Abstract pool test suite', () => {
       elu: { median: true }
     })
     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
-      retries: 6,
       runTime: { median: true },
-      waitTime: { median: false },
-      elu: { median: true }
-    })
-    expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
-      retries: 6,
-      runTime: { median: true },
-      waitTime: { median: false },
       elu: { median: true }
     })
     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
       .workerChoiceStrategies) {
       expect(workerChoiceStrategy.opts).toStrictEqual({
-        retries: 6,
         runTime: { median: true },
         waitTime: { median: false },
-        elu: { median: true }
+        elu: { median: true },
+        weights: expect.objectContaining({
+          0: expect.any(Number),
+          [pool.info.maxSize - 1]: expect.any(Number)
+        })
       })
     }
     expect(
@@ -567,24 +521,19 @@ describe('Abstract pool test suite', () => {
       elu: { median: false }
     })
     expect(pool.opts.workerChoiceStrategyOptions).toStrictEqual({
-      retries: 6,
       runTime: { median: false },
-      waitTime: { median: false },
-      elu: { median: false }
-    })
-    expect(pool.workerChoiceStrategyContext.opts).toStrictEqual({
-      retries: 6,
-      runTime: { median: false },
-      waitTime: { median: false },
       elu: { median: false }
     })
     for (const [, workerChoiceStrategy] of pool.workerChoiceStrategyContext
       .workerChoiceStrategies) {
       expect(workerChoiceStrategy.opts).toStrictEqual({
-        retries: 6,
         runTime: { median: false },
         waitTime: { median: false },
-        elu: { median: false }
+        elu: { median: false },
+        weights: expect.objectContaining({
+          0: expect.any(Number),
+          [pool.info.maxSize - 1]: expect.any(Number)
+        })
       })
     }
     expect(
@@ -613,20 +562,6 @@ describe('Abstract pool test suite', () => {
         'Invalid worker choice strategy options: must be a plain object'
       )
     )
-    expect(() =>
-      pool.setWorkerChoiceStrategyOptions({
-        retries: 'invalidChoiceRetries'
-      })
-    ).toThrow(
-      new TypeError(
-        'Invalid worker choice strategy options: retries must be an integer'
-      )
-    )
-    expect(() => pool.setWorkerChoiceStrategyOptions({ retries: -1 })).toThrow(
-      new RangeError(
-        "Invalid worker choice strategy options: retries '-1' must be greater or equal than zero"
-      )
-    )
     expect(() => pool.setWorkerChoiceStrategyOptions({ weights: {} })).toThrow(
       new Error(
         'Invalid worker choice strategy options: must have a weight for each worker node'
@@ -771,6 +706,7 @@ describe('Abstract pool test suite', () => {
       started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      strategyRetries: 0,
       minSize: numberOfWorkers,
       maxSize: numberOfWorkers,
       workerNodes: numberOfWorkers,
@@ -784,7 +720,7 @@ describe('Abstract pool test suite', () => {
     pool = new DynamicClusterPool(
       Math.floor(numberOfWorkers / 2),
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     expect(pool.info).toStrictEqual({
       version,
@@ -793,6 +729,7 @@ describe('Abstract pool test suite', () => {
       started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      strategyRetries: 0,
       minSize: Math.floor(numberOfWorkers / 2),
       maxSize: numberOfWorkers,
       workerNodes: Math.floor(numberOfWorkers / 2),
@@ -808,7 +745,7 @@ describe('Abstract pool test suite', () => {
   it('Verify that pool worker tasks usage are initialized', async () => {
     const pool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     for (const workerNode of pool.workerNodes) {
       expect(workerNode).toBeInstanceOf(WorkerNode)
@@ -844,7 +781,7 @@ describe('Abstract pool test suite', () => {
   it('Verify that pool worker tasks queue are initialized', async () => {
     let pool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     for (const workerNode of pool.workerNodes) {
       expect(workerNode).toBeInstanceOf(WorkerNode)
@@ -870,7 +807,7 @@ describe('Abstract pool test suite', () => {
   it('Verify that pool worker info are initialized', async () => {
     let pool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     for (const workerNode of pool.workerNodes) {
       expect(workerNode).toBeInstanceOf(WorkerNode)
@@ -878,7 +815,8 @@ describe('Abstract pool test suite', () => {
         id: expect.any(Number),
         type: WorkerTypes.cluster,
         dynamic: false,
-        ready: true
+        ready: true,
+        stealing: false
       })
     }
     await pool.destroy()
@@ -893,7 +831,8 @@ describe('Abstract pool test suite', () => {
         id: expect.any(Number),
         type: WorkerTypes.thread,
         dynamic: false,
-        ready: true
+        ready: true,
+        stealing: false
       })
     }
     await pool.destroy()
@@ -920,15 +859,15 @@ describe('Abstract pool test suite', () => {
   it('Verify that pool can be started after initialization', async () => {
     const pool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js',
+      './tests/worker-files/cluster/testWorker.cjs',
       {
         startWorkers: false
       }
     )
     expect(pool.info.started).toBe(false)
     expect(pool.info.ready).toBe(false)
-    expect(pool.readyEventEmitted).toBe(false)
     expect(pool.workerNodes).toStrictEqual([])
+    expect(pool.readyEventEmitted).toBe(false)
     await expect(pool.execute()).rejects.toThrow(
       new Error('Cannot execute a task on not started pool')
     )
@@ -947,7 +886,7 @@ describe('Abstract pool test suite', () => {
   it('Verify that pool execute() arguments are checked', async () => {
     const pool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     await expect(pool.execute(undefined, 0)).rejects.toThrow(
       new TypeError('name argument must be a string')
@@ -970,7 +909,7 @@ describe('Abstract pool test suite', () => {
   it('Verify that pool worker tasks usage are computed', async () => {
     const pool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     const promises = new Set()
     const maxMultiplier = 2
@@ -1121,7 +1060,7 @@ describe('Abstract pool test suite', () => {
     const pool = new DynamicClusterPool(
       Math.floor(numberOfWorkers / 2),
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     expect(pool.emitter.eventNames()).toStrictEqual([])
     let poolInfo
@@ -1140,6 +1079,7 @@ describe('Abstract pool test suite', () => {
       started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      strategyRetries: expect.any(Number),
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
       workerNodes: expect.any(Number),
@@ -1180,6 +1120,7 @@ describe('Abstract pool test suite', () => {
       started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      strategyRetries: expect.any(Number),
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
       workerNodes: expect.any(Number),
@@ -1219,6 +1160,7 @@ describe('Abstract pool test suite', () => {
       started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      strategyRetries: expect.any(Number),
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
       workerNodes: expect.any(Number),
@@ -1261,10 +1203,12 @@ describe('Abstract pool test suite', () => {
       started: true,
       ready: true,
       strategy: WorkerChoiceStrategies.ROUND_ROBIN,
+      strategyRetries: expect.any(Number),
       minSize: expect.any(Number),
       maxSize: expect.any(Number),
       workerNodes: expect.any(Number),
       idleWorkerNodes: expect.any(Number),
+      stealingWorkerNodes: expect.any(Number),
       busyWorkerNodes: expect.any(Number),
       executedTasks: expect.any(Number),
       executingTasks: expect.any(Number),
@@ -1274,8 +1218,65 @@ describe('Abstract pool test suite', () => {
       stolenTasks: expect.any(Number),
       failedTasks: expect.any(Number)
     })
-    expect(pool.hasBackPressure.callCount).toBe(5)
+    expect(pool.hasBackPressure.callCount).toBeGreaterThanOrEqual(7)
+    await pool.destroy()
+  })
+
+  it('Verify that destroy() waits for queued tasks to finish', async () => {
+    const tasksFinishedTimeout = 2500
+    const pool = new FixedThreadPool(
+      numberOfWorkers,
+      './tests/worker-files/thread/asyncWorker.mjs',
+      {
+        enableTasksQueue: true,
+        tasksQueueOptions: { tasksFinishedTimeout }
+      }
+    )
+    const maxMultiplier = 4
+    let tasksFinished = 0
+    for (const workerNode of pool.workerNodes) {
+      workerNode.on('taskFinished', () => {
+        ++tasksFinished
+      })
+    }
+    for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
+      pool.execute()
+    }
+    expect(pool.info.queuedTasks).toBeGreaterThan(0)
+    const startTime = performance.now()
+    await pool.destroy()
+    const elapsedTime = performance.now() - startTime
+    expect(tasksFinished).toBeLessThanOrEqual(numberOfWorkers * maxMultiplier)
+    expect(elapsedTime).toBeGreaterThanOrEqual(2000)
+    expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 800)
+  })
+
+  it('Verify that destroy() waits until the tasks finished timeout is reached', async () => {
+    const tasksFinishedTimeout = 1000
+    const pool = new FixedThreadPool(
+      numberOfWorkers,
+      './tests/worker-files/thread/asyncWorker.mjs',
+      {
+        enableTasksQueue: true,
+        tasksQueueOptions: { tasksFinishedTimeout }
+      }
+    )
+    const maxMultiplier = 4
+    let tasksFinished = 0
+    for (const workerNode of pool.workerNodes) {
+      workerNode.on('taskFinished', () => {
+        ++tasksFinished
+      })
+    }
+    for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) {
+      pool.execute()
+    }
+    expect(pool.info.queuedTasks).toBeGreaterThan(0)
+    const startTime = performance.now()
     await pool.destroy()
+    const elapsedTime = performance.now() - startTime
+    expect(tasksFinished).toBe(0)
+    expect(elapsedTime).toBeLessThanOrEqual(tasksFinishedTimeout + 800)
   })
 
   it('Verify that pool asynchronous resource track tasks execution', async () => {
@@ -1332,7 +1333,7 @@ describe('Abstract pool test suite', () => {
     await dynamicThreadPool.destroy()
     const fixedClusterPool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
+      './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
     )
     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
     expect(fixedClusterPool.hasTaskFunction(DEFAULT_TASK_NAME)).toBe(true)
@@ -1472,7 +1473,7 @@ describe('Abstract pool test suite', () => {
     await dynamicThreadPool.destroy()
     const fixedClusterPool = new FixedClusterPool(
       numberOfWorkers,
-      './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
+      './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
     )
     await waitPoolEvents(fixedClusterPool, PoolEvents.ready, 1)
     expect(fixedClusterPool.listTaskFunctionNames()).toStrictEqual([
@@ -1542,7 +1543,7 @@ describe('Abstract pool test suite', () => {
     const pool = new DynamicClusterPool(
       Math.floor(numberOfWorkers / 2),
       numberOfWorkers,
-      './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.js'
+      './tests/worker-files/cluster/testMultipleTaskFunctionsWorker.cjs'
     )
     const data = { n: 10 }
     const result0 = await pool.execute(data)
@@ -1607,7 +1608,7 @@ describe('Abstract pool test suite', () => {
     const pool = new DynamicClusterPool(
       Math.floor(numberOfWorkers / 2),
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     const workerNodeKey = 0
     await expect(
@@ -1620,7 +1621,7 @@ describe('Abstract pool test suite', () => {
     const pool = new DynamicClusterPool(
       Math.floor(numberOfWorkers / 2),
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     const workerNodeKey = 0
     await expect(
@@ -1640,7 +1641,7 @@ describe('Abstract pool test suite', () => {
     const pool = new DynamicClusterPool(
       Math.floor(numberOfWorkers / 2),
       numberOfWorkers,
-      './tests/worker-files/cluster/testWorker.js'
+      './tests/worker-files/cluster/testWorker.cjs'
     )
     await expect(
       pool.sendTaskFunctionOperationToWorkers({