fix: close communication channel on error

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 6467f4eb120bad8655daa076e7f910ca3da1f42a..d55c3101bcaead44cab1b424a170383165dc52be 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -30,12 +30,12 @@ import {
 import type {
   IWorker,
   IWorkerNode,
-  MessageHandler,
   WorkerInfo,
   WorkerType,
   WorkerUsage
 } from './worker'
 import {
+  type MeasurementStatisticsRequirements,
   Measurements,
   WorkerChoiceStrategies,
   type WorkerChoiceStrategy,
@@ -516,7 +516,7 @@ export abstract class AbstractPool<
    * @param worker - The worker.
    * @returns The worker node key if found in the pool worker nodes, `-1` otherwise.
    */
-  private getWorkerNodeKey (worker: Worker): number {
+  protected getWorkerNodeKey (worker: Worker): number {
     return this.workerNodes.findIndex(
       workerNode => workerNode.worker === worker
     )
@@ -774,22 +774,21 @@ export abstract class AbstractPool<
     workerUsage: WorkerUsage,
     message: MessageValue<Response>
   ): void {
+    const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements =
+      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
     updateMeasurementStatistics(
       workerUsage.elu.active,
-      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu,
+      eluTaskStatisticsRequirements,
       message.taskPerformance?.elu?.active ?? 0,
       workerUsage.tasks.executed
     )
     updateMeasurementStatistics(
       workerUsage.elu.idle,
-      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu,
+      eluTaskStatisticsRequirements,
       message.taskPerformance?.elu?.idle ?? 0,
       workerUsage.tasks.executed
     )
-    if (
-      this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu
-        .aggregate
-    ) {
+    if (eluTaskStatisticsRequirements.aggregate) {
       if (message.taskPerformance?.elu != null) {
         if (workerUsage.elu.utilization != null) {
           workerUsage.elu.utilization =
@@ -863,6 +862,7 @@ export abstract class AbstractPool<
       const workerNodeKey = this.getWorkerNodeKey(worker)
       const workerInfo = this.getWorkerInfo(workerNodeKey)
       workerInfo.ready = false
+      this.workerNodes[workerNodeKey].closeChannel()
       this.emitter?.emit(PoolEvents.error, error)
       if (this.opts.restartWorkerOnError === true && !this.starting) {
         if (workerInfo.dynamic) {
@@ -910,7 +910,7 @@ export abstract class AbstractPool<
         void (this.destroyWorker(worker) as Promise<void>)
       }
     })
-    const workerInfo = this.getWorkerInfo(this.getWorkerNodeKey(worker))
+    const workerInfo = this.getWorkerInfoByWorker(worker)
     workerInfo.dynamic = true
     if (this.workerChoiceStrategyContext.getStrategyPolicy().useDynamicWorker) {
       workerInfo.ready = true
@@ -928,12 +928,9 @@ export abstract class AbstractPool<
    * @param worker - The worker which should register a listener.
    * @param listener - The message listener callback.
    */
-  private registerWorkerMessageListener<Message extends Data | Response>(
-    worker: Worker,
-    listener: (message: MessageValue<Message>) => void
-  ): void {
-    worker.on('message', listener as MessageHandler<Worker>)
-  }
+  protected abstract registerWorkerMessageListener<
+    Message extends Data | Response
+  >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
 
   /**
    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
@@ -944,18 +941,18 @@ export abstract class AbstractPool<
   protected afterWorkerSetup (worker: Worker): void {
     // Listen to worker messages.
     this.registerWorkerMessageListener(worker, this.workerListener())
-    // Send startup message to worker.
-    this.sendWorkerStartupMessage(worker)
+    // Send the startup message to worker.
+    this.sendStartupMessageToWorker(worker)
     // Setup worker task statistics computation.
     this.setWorkerStatistics(worker)
   }
 
-  private sendWorkerStartupMessage (worker: Worker): void {
-    this.sendToWorker(worker, {
-      ready: false,
-      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
-    })
-  }
+  /**
+   * Sends the startup message to the given worker.
+   *
+   * @param worker - The worker which should receive the startup message.
+   */
+  protected abstract sendStartupMessageToWorker (worker: Worker): void
 
   private redistributeQueuedTasks (workerNodeKey: number): void {
     while (this.tasksQueueSize(workerNodeKey) > 0) {
@@ -1006,9 +1003,9 @@ export abstract class AbstractPool<
   }
 
   private handleWorkerReadyResponse (message: MessageValue<Response>): void {
-    const worker = this.getWorkerById(message.workerId)
-    this.getWorkerInfo(this.getWorkerNodeKey(worker as Worker)).ready =
-      message.ready as boolean
+    this.getWorkerInfoByWorker(
+      this.getWorkerById(message.workerId) as Worker
+    ).ready = message.ready as boolean
     if (this.emitter != null && this.ready) {
       this.emitter.emit(PoolEvents.ready, this.info)
     }
@@ -1051,14 +1048,30 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Gets the worker information.
+   * Gets the worker information from the given worker node key.
    *
    * @param workerNodeKey - The worker node key.
+   * @returns The worker information.
    */
   private getWorkerInfo (workerNodeKey: number): WorkerInfo {
     return this.workerNodes[workerNodeKey].info
   }
 
+  /**
+   * Gets the worker information from the given worker.
+   *
+   * @param worker - The worker.
+   * @returns The worker information.
+   * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker is not found.
+   */
+  protected getWorkerInfoByWorker (worker: Worker): WorkerInfo {
+    const workerNodeKey = this.getWorkerNodeKey(worker)
+    if (workerNodeKey === -1) {
+      throw new Error('Worker not found')
+    }
+    return this.workerNodes[workerNodeKey].info
+  }
+
   /**
    * Adds the given worker in the pool worker nodes.
    *
@@ -1135,7 +1148,7 @@ export abstract class AbstractPool<
         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
           .elu.aggregate
       },
-      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
+      workerId: this.getWorkerInfoByWorker(worker).id as number
     })
   }
 }