fix: fix tasks queued count computation

Signed-off-by: Jérôme Benoit <jerome.benoit@sap.com>

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 23bb62168f8612d58b488763ad76af425b927983..3b544676be9239f8d78f889b190bb782884f972f 100644 (file)
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Ensure cluster pool destroy() gracefully shutdowns worker's server.
 - Ensure pool event is emitted before task error promise rejection.
+- Fix queued tasks count computation.
 
 ## [2.6.3] - 2023-06-19
 

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 058b43b8a37104b29d1d15978613c42735072547..07907be9c6c48bd6eaf73edad4144bba9ada8800 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -21,13 +21,7 @@ import {
   type TasksQueueOptions,
   type WorkerType
 } from './pool'
-import type {
-  IWorker,
-  Task,
-  TaskStatistics,
-  WorkerNode,
-  WorkerUsage
-} from './worker'
+import type { IWorker, Task, WorkerNode, WorkerUsage } from './worker'
 import {
   Measurements,
   WorkerChoiceStrategies,
@@ -332,10 +326,10 @@ export abstract class AbstractPool<
     if (workerChoiceStrategyOptions != null) {
       this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
     }
-    for (const workerNode of this.workerNodes) {
+    for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
       this.setWorkerNodeTasksUsage(
         workerNode,
-        this.getWorkerUsage(workerNode.worker)
+        this.getWorkerUsage(workerNodeKey)
       )
       this.setWorkerStatistics(workerNode.worker)
     }
@@ -823,11 +817,17 @@ export abstract class AbstractPool<
    * @returns The worker nodes length.
    */
   private pushWorkerNode (worker: Worker): number {
-    return this.workerNodes.push({
+    this.workerNodes.push({
       worker,
-      workerUsage: this.getWorkerUsage(worker),
+      workerUsage: this.getWorkerUsage(),
       tasksQueue: new Queue<Task<Data>>()
     })
+    const workerNodeKey = this.getWorkerNodeKey(worker)
+    this.setWorkerNodeTasksUsage(
+      this.workerNodes[workerNodeKey],
+      this.getWorkerUsage(workerNodeKey)
+    )
+    return this.workerNodes.length
   }
 
   // /**
@@ -910,9 +910,19 @@ export abstract class AbstractPool<
     })
   }
 
-  private getWorkerUsage (worker: Worker): WorkerUsage {
+  private getWorkerUsage (workerNodeKey?: number): WorkerUsage {
+    const getQueueSize = (workerNodeKey: number): number => {
+      return this.tasksQueueSize(workerNodeKey)
+    }
     return {
-      tasks: this.getTaskStatistics(worker),
+      tasks: {
+        executed: 0,
+        executing: 0,
+        get queued (): number {
+          return workerNodeKey == null ? 0 : getQueueSize(workerNodeKey)
+        },
+        failed: 0
+      },
       runTime: {
         aggregate: 0,
         average: 0,
@@ -942,17 +952,4 @@ export abstract class AbstractPool<
       }
     }
   }
-
-  private getTaskStatistics (worker: Worker): TaskStatistics {
-    const queueSize =
-      this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
-    return {
-      executed: 0,
-      executing: 0,
-      get queued (): number {
-        return queueSize ?? 0
-      },
-      failed: 0
-    }
-  }
 }

diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js
index b6219ce954e21c878df1c2a67ce0e86031ceddc0..a8cc145419c54826808a1faaf0a0d3e3a98b7c4b 100644 (file)
--- a/tests/pools/abstract/abstract-pool.test.js
+++ b/tests/pools/abstract/abstract-pool.test.js
@@ -498,6 +498,7 @@ describe('Abstract pool test suite', () => {
       expect(workerNode.tasksQueue).toBeDefined()
       expect(workerNode.tasksQueue).toBeInstanceOf(Queue)
       expect(workerNode.tasksQueue.size).toBe(0)
+      expect(workerNode.tasksQueue.maxSize).toBe(0)
     }
     await pool.destroy()
   })

diff --git a/tests/pools/cluster/fixed.test.js b/tests/pools/cluster/fixed.test.js
index cf0be68db1e028252c5556531c8f4d9f856e6d1e..d0ddcf044cfb711e19aaa75997633c673f764948 100644 (file)
--- a/tests/pools/cluster/fixed.test.js
+++ b/tests/pools/cluster/fixed.test.js
@@ -100,7 +100,7 @@ describe('Fixed cluster pool test suite', () => {
         queuePool.opts.tasksQueueOptions.concurrency
       )
       expect(workerNode.workerUsage.tasks.executed).toBe(0)
-      expect(workerNode.tasksQueue.size).toBeGreaterThan(0)
+      expect(workerNode.workerUsage.tasks.queued).toBeGreaterThan(0)
     }
     expect(queuePool.info.executingTasks).toBe(numberOfWorkers)
     expect(queuePool.info.queuedTasks).toBe(
@@ -116,7 +116,7 @@ describe('Fixed cluster pool test suite', () => {
       expect(workerNode.workerUsage.tasks.executed).toBeLessThanOrEqual(
         maxMultiplier
       )
-      expect(workerNode.tasksQueue.size).toBe(0)
+      expect(workerNode.workerUsage.tasks.queued).toBe(0)
     }
   })
 

diff --git a/tests/pools/thread/fixed.test.js b/tests/pools/thread/fixed.test.js
index 39ddc67d562c80d06cfdb86b5447c589feff8a25..f210d52016cac5115efff53991e465aebdd77621 100644 (file)
--- a/tests/pools/thread/fixed.test.js
+++ b/tests/pools/thread/fixed.test.js
@@ -100,7 +100,7 @@ describe('Fixed thread pool test suite', () => {
         queuePool.opts.tasksQueueOptions.concurrency
       )
       expect(workerNode.workerUsage.tasks.executed).toBe(0)
-      expect(workerNode.tasksQueue.size).toBeGreaterThan(0)
+      expect(workerNode.workerUsage.tasks.queued).toBeGreaterThan(0)
     }
     expect(queuePool.info.executingTasks).toBe(numberOfThreads)
     expect(queuePool.info.queuedTasks).toBe(
@@ -116,7 +116,7 @@ describe('Fixed thread pool test suite', () => {
       expect(workerNode.workerUsage.tasks.executed).toBeLessThanOrEqual(
         maxMultiplier
       )
-      expect(workerNode.tasksQueue.size).toBe(0)
+      expect(workerNode.workerUsage.tasks.queued).toBe(0)
     }
   })