refactor: cleanup variables namespace

[poolifier.git] / src / pools / abstract-pool.ts

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index bb37dbb70fe4aac89115fc5da929270f0a39dd8d..cb5d4ce7f60d81129ed9015108e10b14c4cbef80 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,6 +1,10 @@
 import { randomUUID } from 'node:crypto'
 import { performance } from 'node:perf_hooks'
-import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
+import type {
+  MessageValue,
+  PromiseResponseWrapper,
+  Task
+} from '../utility-types'
 import {
   DEFAULT_TASK_NAME,
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
@@ -25,7 +29,6 @@ import type {
   IWorker,
   IWorkerNode,
   MessageHandler,
-  Task,
   WorkerInfo,
   WorkerType,
   WorkerUsage
@@ -309,7 +312,7 @@ export abstract class AbstractPool<
       ),
       maxQueuedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.usage.tasks.maxQueued,
+          accumulator + (workerNode.usage.tasks?.maxQueued ?? 0),
         0
       ),
       failedTasks: this.workerNodes.reduce(
@@ -423,7 +426,7 @@ export abstract class AbstractPool<
    * @returns The pool utilization.
    */
   private get utilization (): number {
-    const poolRunTimeCapacity =
+    const poolTimeCapacity =
       (performance.now() - this.startTimestamp) * this.maxSize
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
@@ -435,7 +438,7 @@ export abstract class AbstractPool<
         accumulator + (workerNode.usage.waitTime?.aggregate ?? 0),
       0
     )
-    return (totalTasksRunTime + totalTasksWaitTime) / poolRunTimeCapacity
+    return (totalTasksRunTime + totalTasksWaitTime) / poolTimeCapacity
   }
 
   /**
@@ -471,6 +474,12 @@ export abstract class AbstractPool<
       ?.worker
   }
 
+  /**
+   * Checks if the worker id sent in the received message from a worker is valid.
+   *
+   * @param message - The received message.
+   * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
+   */
   private checkMessageWorkerId (message: MessageValue<Response>): void {
     if (
       message.workerId != null &&
@@ -672,11 +681,11 @@ export abstract class AbstractPool<
     const workerUsage = this.workerNodes[workerNodeKey].usage
     ++workerUsage.tasks.executing
     this.updateWaitTimeWorkerUsage(workerUsage, task)
-    const tasksWorkerUsage = this.workerNodes[
-      workerNodeKey
-    ].getTasksWorkerUsage(task.name as string) as WorkerUsage
-    ++tasksWorkerUsage.tasks.executing
-    this.updateWaitTimeWorkerUsage(tasksWorkerUsage, task)
+    const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
+      task.name as string
+    ) as WorkerUsage
+    ++taskWorkerUsage.tasks.executing
+    this.updateWaitTimeWorkerUsage(taskWorkerUsage, task)
   }
 
   /**
@@ -695,12 +704,12 @@ export abstract class AbstractPool<
     this.updateTaskStatisticsWorkerUsage(workerUsage, message)
     this.updateRunTimeWorkerUsage(workerUsage, message)
     this.updateEluWorkerUsage(workerUsage, message)
-    const tasksWorkerUsage = this.workerNodes[
-      workerNodeKey
-    ].getTasksWorkerUsage(message.name as string) as WorkerUsage
-    this.updateTaskStatisticsWorkerUsage(tasksWorkerUsage, message)
-    this.updateRunTimeWorkerUsage(tasksWorkerUsage, message)
-    this.updateEluWorkerUsage(tasksWorkerUsage, message)
+    const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
+      message.taskPerformance?.name ?? DEFAULT_TASK_NAME
+    ) as WorkerUsage
+    this.updateTaskStatisticsWorkerUsage(taskWorkerUsage, message)
+    this.updateRunTimeWorkerUsage(taskWorkerUsage, message)
+    this.updateEluWorkerUsage(taskWorkerUsage, message)
   }
 
   private updateTaskStatisticsWorkerUsage (
@@ -947,9 +956,6 @@ export abstract class AbstractPool<
       if (this.emitter != null) {
         this.emitter.emit(PoolEvents.error, error)
       }
-      if (this.opts.enableTasksQueue === true) {
-        this.redistributeQueuedTasks(worker)
-      }
       if (this.opts.restartWorkerOnError === true && !this.starting) {
         if (this.getWorkerInfo(this.getWorkerNodeKey(worker)).dynamic) {
           this.createAndSetupDynamicWorker()
@@ -957,6 +963,9 @@ export abstract class AbstractPool<
           this.createAndSetupWorker()
         }
       }
+      if (this.opts.enableTasksQueue === true) {
+        this.redistributeQueuedTasks(worker)
+      }
     })
     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)