X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fpools%2Fabstract-pool.ts;h=77991d7b95b76c67816d01faab881c1711f6c788;hb=52b71763f1a1240170e777049a80df0ad14ce496;hp=5481099f202c19f5ef57833d4f454a18757dce3c;hpb=008512c74c553d4400f3fabb46053e5e8a4a0cd1;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 5481099f..77991d7b 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -30,7 +30,6 @@ import {
 import type {
   IWorker,
   IWorkerNode,
-  MessageHandler,
   WorkerInfo,
   WorkerType,
   WorkerUsage
@@ -65,7 +64,7 @@ export abstract class AbstractPool<
   public readonly emitter?: PoolEmitter
 
   /**
-   * The execution response promise map.
+   * The task execution response promise map.
    *
    * - `key`: The message id of each submitted task.
    * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
@@ -517,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
     )
@@ -615,37 +614,35 @@ export abstract class AbstractPool<
 
   /** @inheritDoc */
   public async execute (data?: Data, name?: string): Promise<Response> {
-    const timestamp = performance.now()
-    const workerNodeKey = this.chooseWorkerNode()
-    const submittedTask: Task<Data> = {
-      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) => {
+    return await new Promise<Response>((resolve, reject) => {
+      const timestamp = performance.now()
+      const workerNodeKey = this.chooseWorkerNode()
+      const submittedTask: Task<Data> = {
+        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()
+      }
       this.promiseResponseMap.set(submittedTask.id as string, {
         resolve,
         reject,
         worker: this.workerNodes[workerNodeKey].worker
       })
+      if (
+        this.opts.enableTasksQueue === true &&
+        (this.busy ||
+          this.workerNodes[workerNodeKey].usage.tasks.executing >=
+            ((this.opts.tasksQueueOptions as TasksQueueOptions)
+              .concurrency as number))
+      ) {
+        this.enqueueTask(workerNodeKey, submittedTask)
+      } else {
+        this.executeTask(workerNodeKey, submittedTask)
+      }
+      this.checkAndEmitEvents()
     })
-    if (
-      this.opts.enableTasksQueue === true &&
-      (this.busy ||
-        this.workerNodes[workerNodeKey].usage.tasks.executing >=
-          ((this.opts.tasksQueueOptions as TasksQueueOptions)
-            .concurrency as number))
-    ) {
-      this.enqueueTask(workerNodeKey, submittedTask)
-    } else {
-      this.executeTask(workerNodeKey, submittedTask)
-    }
-    this.checkAndEmitEvents()
-    // eslint-disable-next-line @typescript-eslint/return-await
-    return res
   }
 
   /** @inheritDoc */
@@ -863,6 +860,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 +908,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 +926,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 +939,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 +1001,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 +1046,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 +1146,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
     })
   }
 }