Merge branch 'master' of github.com:poolifier/poolifier

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 6c9a0fd185f7c560f4f7fa5ac597ab784f6d6dbd..a7f1579fde52512735224c1f47738f8ab7c211f2 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto'
 import { performance } from 'node:perf_hooks'
 import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
 import {
+  DEFAULT_TASK_NAME,
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
   isKillBehavior,
@@ -158,14 +159,20 @@ export abstract class AbstractPool<
   }
 
   protected checkDynamicPoolSize (min: number, max: number): void {
-    if (this.type === PoolTypes.dynamic && min > max) {
-      throw new RangeError(
-        'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
-      )
-    } else if (this.type === PoolTypes.dynamic && min === max) {
-      throw new RangeError(
-        'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
-      )
+    if (this.type === PoolTypes.dynamic) {
+      if (min > max) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
+        )
+      } else if (min === 0 && max === 0) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
+        )
+      } else if (min === max) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
+        )
+      }
     }
   }
 
@@ -302,7 +309,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(
@@ -397,14 +404,16 @@ export abstract class AbstractPool<
 
   private get starting (): boolean {
     return (
-      !this.full ||
-      (this.full && this.workerNodes.some(workerNode => !workerNode.info.ready))
+      this.workerNodes.length < this.minSize ||
+      (this.workerNodes.length >= this.minSize &&
+        this.workerNodes.some(workerNode => !workerNode.info.ready))
     )
   }
 
   private get ready (): boolean {
     return (
-      this.full && this.workerNodes.every(workerNode => workerNode.info.ready)
+      this.workerNodes.length >= this.minSize &&
+      this.workerNodes.every(workerNode => workerNode.info.ready)
     )
   }
 
@@ -462,6 +471,17 @@ export abstract class AbstractPool<
       ?.worker
   }
 
+  private checkMessageWorkerId (message: MessageValue<Response>): void {
+    if (
+      message.workerId != null &&
+      this.getWorkerById(message.workerId) == null
+    ) {
+      throw new Error(
+        `Worker message received from unknown worker '${message.workerId}'`
+      )
+    }
+  }
+
   /**
    * Gets the given worker its worker node key.
    *
@@ -569,10 +589,11 @@ 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,
+      workerId: this.getWorkerInfo(workerNodeKey).id as number,
       id: randomUUID()
     }
     const res = new Promise<Response>((resolve, reject) => {
@@ -651,6 +672,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)
   }
 
   /**
@@ -664,10 +690,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 tasksWorkerUsage = this.workerNodes[
+      workerNodeKey
+    ].getTasksWorkerUsage(message.name as string) as WorkerUsage
+    this.updateTaskStatisticsWorkerUsage(tasksWorkerUsage, message)
+    this.updateRunTimeWorkerUsage(tasksWorkerUsage, message)
+    this.updateEluWorkerUsage(tasksWorkerUsage, message)
   }
 
   private updateTaskStatisticsWorkerUsage (
@@ -894,7 +927,7 @@ export abstract class AbstractPool<
     // Send startup message to worker.
     this.sendToWorker(worker, {
       ready: false,
-      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id
+      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
     })
     // Setup worker task statistics computation.
     this.setWorkerStatistics(worker)
@@ -988,8 +1021,12 @@ export abstract class AbstractPool<
         void (this.destroyWorker(worker) as Promise<void>)
       }
     })
-    this.getWorkerInfo(this.getWorkerNodeKey(worker)).dynamic = true
-    this.sendToWorker(worker, { checkAlive: true })
+    const workerInfo = this.getWorkerInfo(this.getWorkerNodeKey(worker))
+    workerInfo.dynamic = true
+    this.sendToWorker(worker, {
+      checkAlive: true,
+      workerId: workerInfo.id as number
+    })
     return worker
   }
 
@@ -1000,6 +1037,7 @@ export abstract class AbstractPool<
    */
   protected workerListener (): (message: MessageValue<Response>) => void {
     return message => {
+      this.checkMessageWorkerId(message)
       if (message.ready != null && message.workerId != null) {
         // Worker ready message received
         this.handleWorkerReadyMessage(message)
@@ -1011,17 +1049,9 @@ export abstract class AbstractPool<
   }
 
   private handleWorkerReadyMessage (message: MessageValue<Response>): void {
-    const worker = this.getWorkerById(message.workerId as number)
-    if (worker != null) {
-      this.getWorkerInfo(this.getWorkerNodeKey(worker)).ready =
-        message.ready as boolean
-    } else {
-      throw new Error(
-        `Worker ready message received from unknown worker '${
-          message.workerId as number
-        }'`
-      )
-    }
+    const worker = this.getWorkerById(message.workerId)
+    this.getWorkerInfo(this.getWorkerNodeKey(worker as Worker)).ready =
+      message.ready as boolean
     if (this.emitter != null && this.ready) {
       this.emitter.emit(PoolEvents.ready, this.info)
     }
@@ -1138,7 +1168,8 @@ export abstract class AbstractPool<
             .runTime.aggregate,
         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
           .elu.aggregate
-      }
+      },
+      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
     })
   }
 }