X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=a840dc740116057f8fa7867f85348a676caf04a4;hb=refs%2Ftags%2Fv2.4.6;hp=f4f3e8337056a1cb47edba30e0b4982ef339f5e0;hpb=a34454964733f585af19a04f21f9e35eb18c77b1;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index f4f3e833..a840dc74 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -76,9 +76,9 @@ export abstract class AbstractPool<
     this.checkPoolOptions(this.opts)
 
     this.chooseWorkerNode.bind(this)
-    this.internalExecute.bind(this)
+    this.executeTask.bind(this)
+    this.enqueueTask.bind(this)
     this.checkAndEmitEvents.bind(this)
-    this.sendToWorker.bind(this)
 
     this.setupHook()
 
@@ -191,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
@@ -219,10 +214,7 @@ export abstract class AbstractPool<
   public abstract get busy (): boolean
 
   protected internalBusy (): boolean {
-    return (
-      this.numberOfRunningTasks >= this.numberOfWorkers &&
-      this.findFreeWorkerNodeKey() === -1
-    )
+    return this.findFreeWorkerNodeKey() === -1
   }
 
   /** @inheritDoc */
@@ -240,18 +232,21 @@ export abstract class AbstractPool<
       data: data ?? ({} as Data),
       id: crypto.randomUUID()
     }
-    const res = this.internalExecute(workerNodeKey, workerNode, submittedTask)
-    let currentTask: Task<Data> = submittedTask
+    const res = new Promise<Response>((resolve, reject) => {
+      this.promiseResponseMap.set(submittedTask.id, {
+        resolve,
+        reject,
+        worker: workerNode.worker
+      })
+    })
     if (
       this.opts.enableTasksQueue === true &&
-      (this.busy || this.tasksQueueSize(workerNodeKey) > 0)
+      (this.busy || this.workerNodes[workerNodeKey].tasksUsage.running > 0)
     ) {
-      currentTask = this.enqueueDequeueTask(
-        workerNodeKey,
-        submittedTask
-      ) as Task<Data>
+      this.enqueueTask(workerNodeKey, submittedTask)
+    } else {
+      this.executeTask(workerNodeKey, submittedTask)
     }
-    this.sendToWorker(workerNode.worker, currentTask)
     this.checkAndEmitEvents()
     // eslint-disable-next-line @typescript-eslint/return-await
     return res
@@ -260,8 +255,8 @@ export abstract class AbstractPool<
   /** @inheritDoc */
   public async destroy (): Promise<void> {
     await Promise.all(
-      this.workerNodes.map(async workerNode => {
-        this.flushTasksQueueByWorker(workerNode.worker)
+      this.workerNodes.map(async (workerNode, workerNodeKey) => {
+        this.flushTasksQueue(workerNodeKey)
         await this.destroyWorker(workerNode.worker)
       })
     )
@@ -275,7 +270,7 @@ export abstract class AbstractPool<
   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 
   /**
-   * Setup hook to run code before worker node are created in the abstract constructor.
+   * Setup hook to execute code before worker node are created in the abstract constructor.
    * Can be overridden
    *
    * @virtual
@@ -290,23 +285,23 @@ export abstract class AbstractPool<
   protected abstract isMain (): boolean
 
   /**
-   * Hook executed before the worker task promise resolution.
+   * Hook executed before the worker task execution.
    * Can be overridden.
    *
    * @param workerNodeKey - The worker node key.
    */
-  protected beforePromiseResponseHook (workerNodeKey: number): void {
+  protected beforeTaskExecutionHook (workerNodeKey: number): void {
     ++this.workerNodes[workerNodeKey].tasksUsage.running
   }
 
   /**
-   * Hook executed after the worker task promise resolution.
+   * Hook executed after the worker task execution.
    * Can be overridden.
    *
    * @param worker - The worker.
    * @param message - The received message.
    */
-  protected afterPromiseResponseHook (
+  protected afterTaskExecutionHook (
     worker: Worker,
     message: MessageValue<Response>
   ): void {
@@ -341,11 +336,7 @@ export abstract class AbstractPool<
    */
   protected chooseWorkerNode (): [number, WorkerNode<Worker, Data>] {
     let workerNodeKey: number
-    if (
-      this.type === PoolType.DYNAMIC &&
-      !this.full &&
-      this.findFreeWorkerNodeKey() === -1
-    ) {
+    if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
       const workerCreated = this.createAndSetupWorker()
       this.registerWorkerMessageListener(workerCreated, message => {
         if (
@@ -439,15 +430,15 @@ export abstract class AbstractPool<
           } else {
             promiseResponse.resolve(message.data as Response)
           }
-          this.afterPromiseResponseHook(promiseResponse.worker, message)
+          this.afterTaskExecutionHook(promiseResponse.worker, message)
           this.promiseResponseMap.delete(message.id)
           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
           if (
             this.opts.enableTasksQueue === true &&
             this.tasksQueueSize(workerNodeKey) > 0
           ) {
-            this.sendToWorker(
-              promiseResponse.worker,
+            this.executeTask(
+              workerNodeKey,
               this.dequeueTask(workerNodeKey) as Task<Data>
             )
           }
@@ -456,21 +447,6 @@ export abstract class AbstractPool<
     }
   }
 
-  private async internalExecute (
-    workerNodeKey: number,
-    workerNode: WorkerNode<Worker, Data>,
-    task: Task<Data>
-  ): Promise<Response> {
-    this.beforePromiseResponseHook(workerNodeKey)
-    return await new Promise<Response>((resolve, reject) => {
-      this.promiseResponseMap.set(task.id, {
-        resolve,
-        reject,
-        worker: workerNode.worker
-      })
-    })
-  }
-
   private checkAndEmitEvents (): void {
     if (this.opts.enableEvents === true) {
       if (this.busy) {
@@ -482,6 +458,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.
    *
@@ -550,16 +539,13 @@ export abstract class AbstractPool<
     this.workerChoiceStrategyContext.remove(workerNodeKey)
   }
 
-  private enqueueDequeueTask (
-    workerNodeKey: number,
-    task: Task<Data>
-  ): Task<Data> | undefined {
-    this.enqueueTask(workerNodeKey, task)
-    return this.dequeueTask(workerNodeKey)
+  private executeTask (workerNodeKey: number, task: Task<Data>): void {
+    this.beforeTaskExecutionHook(workerNodeKey)
+    this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
   }
 
-  private enqueueTask (workerNodeKey: number, task: Task<Data>): void {
-    this.workerNodes[workerNodeKey].tasksQueue.push(task)
+  private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
+    return this.workerNodes[workerNodeKey].tasksQueue.push(task)
   }
 
   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
@@ -573,9 +559,8 @@ export abstract class AbstractPool<
   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)
+        this.executeTask(workerNodeKey, task)
       }
-      this.workerNodes[workerNodeKey].tasksQueue = []
     }
   }