fix: ensure not more than one task is executed on a worker with tasks

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 1668813354538bfcf5d8a39c38007c5d0f42755d..a8104e9e26bdb31dd7d0374794538eafd9acb116 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -33,12 +33,12 @@ export abstract class AbstractPool<
   public readonly emitter?: PoolEmitter
 
   /**
-   * The promise response map.
+   * The execution response promise map.
    *
    * - `key`: The message id of each submitted task.
-   * - `value`: An object that contains the worker, the promise resolve and reject callbacks.
+   * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks.
    *
-   * When we receive a message from the worker we get a map entry with the promise resolve/reject bound to the message.
+   * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id.
    */
   protected promiseResponseMap: Map<
   string,
@@ -93,7 +93,11 @@ export abstract class AbstractPool<
     Worker,
     Data,
     Response
-    >(this, this.opts.workerChoiceStrategy)
+    >(
+      this,
+      this.opts.workerChoiceStrategy,
+      this.opts.workerChoiceStrategyOptions
+    )
   }
 
   private checkFilePath (filePath: string): void {
@@ -127,6 +131,8 @@ export abstract class AbstractPool<
     this.opts.workerChoiceStrategy =
       opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
     this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy)
+    this.opts.workerChoiceStrategyOptions =
+      opts.workerChoiceStrategyOptions ?? { medRunTime: false }
     this.opts.enableEvents = opts.enableEvents ?? true
     this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
   }
@@ -185,21 +191,16 @@ export abstract class AbstractPool<
   ): void {
     this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
     this.opts.workerChoiceStrategy = workerChoiceStrategy
-    for (const [index, workerNode] of this.workerNodes.entries()) {
-      this.setWorkerNode(
-        index,
-        workerNode.worker,
-        {
-          run: 0,
-          running: 0,
-          runTime: 0,
-          runTimeHistory: new CircularArray(),
-          avgRunTime: 0,
-          medRunTime: 0,
-          error: 0
-        },
-        workerNode.tasksQueue
-      )
+    for (const workerNode of this.workerNodes) {
+      this.setWorkerNodeTasksUsage(workerNode, {
+        run: 0,
+        running: 0,
+        runTime: 0,
+        runTimeHistory: new CircularArray(),
+        avgRunTime: 0,
+        medRunTime: 0,
+        error: 0
+      })
     }
     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
       workerChoiceStrategy
@@ -235,15 +236,14 @@ export abstract class AbstractPool<
       id: crypto.randomUUID()
     }
     const res = this.internalExecute(workerNodeKey, workerNode, submittedTask)
-    let currentTask: Task<Data> = submittedTask
     if (
       this.opts.enableTasksQueue === true &&
-      (this.busy || this.tasksQueueLength(workerNodeKey) > 0)
+      (this.busy || this.workerNodes[workerNodeKey].tasksUsage.running > 0)
     ) {
       this.enqueueTask(workerNodeKey, submittedTask)
-      currentTask = this.dequeueTask(workerNodeKey) as Task<Data>
+    } else {
+      this.sendToWorker(workerNode.worker, submittedTask)
     }
-    this.sendToWorker(workerNode.worker, currentTask)
     this.checkAndEmitEvents()
     // eslint-disable-next-line @typescript-eslint/return-await
     return res
@@ -423,7 +423,7 @@ export abstract class AbstractPool<
   protected workerListener (): (message: MessageValue<Response>) => void {
     return message => {
       if (message.id != null) {
-        // Task response received
+        // Task execution response received
         const promiseResponse = this.promiseResponseMap.get(message.id)
         if (promiseResponse != null) {
           if (message.error != null) {
@@ -436,7 +436,7 @@ export abstract class AbstractPool<
           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
           if (
             this.opts.enableTasksQueue === true &&
-            this.tasksQueueLength(workerNodeKey) > 0
+            this.tasksQueueSize(workerNodeKey) > 0
           ) {
             this.sendToWorker(
               promiseResponse.worker,
@@ -474,6 +474,19 @@ export abstract class AbstractPool<
     }
   }
 
+  /**
+   * Sets the given worker node its tasks usage in the pool.
+   *
+   * @param workerNode - The worker node.
+   * @param tasksUsage - The worker node tasks usage.
+   */
+  private setWorkerNodeTasksUsage (
+    workerNode: WorkerNode<Worker, Data>,
+    tasksUsage: TasksUsage
+  ): void {
+    workerNode.tasksUsage = tasksUsage
+  }
+
   /**
    * Gets the given worker its tasks usage in the pool.
    *
@@ -536,26 +549,26 @@ export abstract class AbstractPool<
    *
    * @param worker - The worker.
    */
-  protected removeWorkerNode (worker: Worker): void {
+  private removeWorkerNode (worker: Worker): void {
     const workerNodeKey = this.getWorkerNodeKey(worker)
     this.workerNodes.splice(workerNodeKey, 1)
     this.workerChoiceStrategyContext.remove(workerNodeKey)
   }
 
-  protected enqueueTask (workerNodeKey: number, task: Task<Data>): void {
+  private enqueueTask (workerNodeKey: number, task: Task<Data>): void {
     this.workerNodes[workerNodeKey].tasksQueue.push(task)
   }
 
-  protected dequeueTask (workerNodeKey: number): Task<Data> | undefined {
+  private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
     return this.workerNodes[workerNodeKey].tasksQueue.shift()
   }
 
-  protected tasksQueueLength (workerNodeKey: number): number {
+  private tasksQueueSize (workerNodeKey: number): number {
     return this.workerNodes[workerNodeKey].tasksQueue.length
   }
 
-  protected flushTasksQueue (workerNodeKey: number): void {
-    if (this.tasksQueueLength(workerNodeKey) > 0) {
+  private flushTasksQueue (workerNodeKey: number): void {
+    if (this.tasksQueueSize(workerNodeKey) > 0) {
       for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
         this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
       }
@@ -563,7 +576,7 @@ export abstract class AbstractPool<
     }
   }
 
-  protected flushTasksQueueByWorker (worker: Worker): void {
+  private flushTasksQueueByWorker (worker: Worker): void {
     const workerNodeKey = this.getWorkerNodeKey(worker)
     this.flushTasksQueue(workerNodeKey)
   }