chore: v2.6.14

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index c699b5282a7c0b862a4cc16b63ee5dd34637a643..c200d5317da6ba47c1154cdeb15a9b5ca5c9ee30 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,7 +1,12 @@
 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,
   EMPTY_FUNCTION,
   isKillBehavior,
@@ -24,7 +29,6 @@ import type {
   IWorker,
   IWorkerNode,
   MessageHandler,
-  Task,
   WorkerInfo,
   WorkerType,
   WorkerUsage
@@ -308,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(
@@ -588,7 +592,7 @@ export abstract class AbstractPool<
     const timestamp = performance.now()
     const workerNodeKey = this.chooseWorkerNode()
     const submittedTask: Task<Data> = {
-      name,
+      name: name ?? DEFAULT_TASK_NAME,
       // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
       data: data ?? ({} as Data),
       timestamp,
@@ -671,6 +675,11 @@ export abstract class AbstractPool<
     const workerUsage = this.workerNodes[workerNodeKey].usage
     ++workerUsage.tasks.executing
     this.updateWaitTimeWorkerUsage(workerUsage, task)
+    const taskWorkerUsage = this.workerNodes[workerNodeKey].getTaskWorkerUsage(
+      task.name as string
+    ) as WorkerUsage
+    ++taskWorkerUsage.tasks.executing
+    this.updateWaitTimeWorkerUsage(taskWorkerUsage, task)
   }
 
   /**
@@ -684,10 +693,17 @@ export abstract class AbstractPool<
     worker: Worker,
     message: MessageValue<Response>
   ): void {
-    const workerUsage = this.workerNodes[this.getWorkerNodeKey(worker)].usage
+    const workerNodeKey = this.getWorkerNodeKey(worker)
+    const workerUsage = this.workerNodes[workerNodeKey].usage
     this.updateTaskStatisticsWorkerUsage(workerUsage, message)
     this.updateRunTimeWorkerUsage(workerUsage, message)
     this.updateEluWorkerUsage(workerUsage, 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 (