X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=da0f2e77358525336a5936a42841edce9016f44b;hb=70726c2810ba4768dae5ae1f40ed6ddcecf142ba;hp=058b43b8a37104b29d1d15978613c42735072547;hpb=cd9580e7fd7edad1758e8797f6168c05d6701621;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 058b43b8..da0f2e77 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -23,8 +23,8 @@ import {
 } from './pool'
 import type {
   IWorker,
+  MessageHandler,
   Task,
-  TaskStatistics,
   WorkerNode,
   WorkerUsage
 } from './worker'
@@ -40,8 +40,8 @@ import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choic
  * Base class that implements some shared logic for all poolifier pools.
  *
  * @typeParam Worker - Type of worker which manages this pool.
- * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
- * @typeParam Response - Type of execution response. This can only be serializable data.
+ * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
+ * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  */
 export abstract class AbstractPool<
   Worker extends IWorker,
@@ -269,12 +269,13 @@ export abstract class AbstractPool<
         0
       ),
       queuedTasks: this.workerNodes.reduce(
-        (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
+        (accumulator, workerNode) =>
+          accumulator + workerNode.workerUsage.tasks.queued,
         0
       ),
       maxQueuedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
-          accumulator + workerNode.tasksQueue.maxSize,
+          accumulator + workerNode.workerUsage.tasks.maxQueued,
         0
       ),
       failedTasks: this.workerNodes.reduce(
@@ -332,10 +333,10 @@ export abstract class AbstractPool<
     if (workerChoiceStrategyOptions != null) {
       this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions)
     }
-    for (const workerNode of this.workerNodes) {
+    for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) {
       this.setWorkerNodeTasksUsage(
         workerNode,
-        this.getWorkerUsage(workerNode.worker)
+        this.getWorkerUsage(workerNodeKey)
       )
       this.setWorkerStatistics(workerNode.worker)
     }
@@ -465,8 +466,8 @@ export abstract class AbstractPool<
   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 
   /**
-   * Setup hook to execute code before worker node are created in the abstract constructor.
-   * Can be overridden
+   * Setup hook to execute code before worker nodes are created in the abstract constructor.
+   * Can be overridden.
    *
    * @virtual
    */
@@ -675,9 +676,12 @@ export abstract class AbstractPool<
    * @param worker - The worker which should register a listener.
    * @param listener - The message listener callback.
    */
-  protected abstract registerWorkerMessageListener<
-    Message extends Data | Response
-  >(worker: Worker, listener: (message: MessageValue<Message>) => void): void
+  private registerWorkerMessageListener<Message extends Data | Response>(
+    worker: Worker,
+    listener: (message: MessageValue<Message>) => void
+  ): void {
+    worker.on('message', listener as MessageHandler<Worker>)
+  }
 
   /**
    * Creates a new worker.
@@ -688,12 +692,14 @@ export abstract class AbstractPool<
 
   /**
    * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes.
-   *
-   * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default.
+   * Can be overridden.
    *
    * @param worker - The newly created worker.
    */
-  protected abstract afterWorkerSetup (worker: Worker): void
+  protected afterWorkerSetup (worker: Worker): void {
+    // Listen to worker messages.
+    this.registerWorkerMessageListener(worker, this.workerListener())
+  }
 
   /**
    * Creates a new worker and sets it up completely in the pool worker nodes.
@@ -823,11 +829,17 @@ export abstract class AbstractPool<
    * @returns The worker nodes length.
    */
   private pushWorkerNode (worker: Worker): number {
-    return this.workerNodes.push({
+    this.workerNodes.push({
       worker,
-      workerUsage: this.getWorkerUsage(worker),
+      workerUsage: this.getWorkerUsage(),
       tasksQueue: new Queue<Task<Data>>()
     })
+    const workerNodeKey = this.getWorkerNodeKey(worker)
+    this.setWorkerNodeTasksUsage(
+      this.workerNodes[workerNodeKey],
+      this.getWorkerUsage(workerNodeKey)
+    )
+    return this.workerNodes.length
   }
 
   // /**
@@ -881,6 +893,10 @@ export abstract class AbstractPool<
     return this.workerNodes[workerNodeKey].tasksQueue.size
   }
 
+  private tasksMaxQueueSize (workerNodeKey: number): number {
+    return this.workerNodes[workerNodeKey].tasksQueue.maxSize
+  }
+
   private flushTasksQueue (workerNodeKey: number): void {
     if (this.tasksQueueSize(workerNodeKey) > 0) {
       for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
@@ -890,6 +906,7 @@ export abstract class AbstractPool<
         )
       }
     }
+    this.workerNodes[workerNodeKey].tasksQueue.clear()
   }
 
   private flushTasksQueues (): void {
@@ -910,9 +927,25 @@ export abstract class AbstractPool<
     })
   }
 
-  private getWorkerUsage (worker: Worker): WorkerUsage {
+  private getWorkerUsage (workerNodeKey?: number): WorkerUsage {
+    const getTasksQueueSize = (workerNodeKey?: number): number => {
+      return workerNodeKey != null ? this.tasksQueueSize(workerNodeKey) : 0
+    }
+    const getTasksMaxQueueSize = (workerNodeKey?: number): number => {
+      return workerNodeKey != null ? this.tasksMaxQueueSize(workerNodeKey) : 0
+    }
     return {
-      tasks: this.getTaskStatistics(worker),
+      tasks: {
+        executed: 0,
+        executing: 0,
+        get queued (): number {
+          return getTasksQueueSize(workerNodeKey)
+        },
+        get maxQueued (): number {
+          return getTasksMaxQueueSize(workerNodeKey)
+        },
+        failed: 0
+      },
       runTime: {
         aggregate: 0,
         average: 0,
@@ -942,17 +975,4 @@ export abstract class AbstractPool<
       }
     }
   }
-
-  private getTaskStatistics (worker: Worker): TaskStatistics {
-    const queueSize =
-      this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size
-    return {
-      executed: 0,
-      executing: 0,
-      get queued (): number {
-        return queueSize ?? 0
-      },
-      failed: 0
-    }
-  }
 }