Merge branch 'master' of github.com:jerome-benoit/poolifier

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index a8104e9e26bdb31dd7d0374794538eafd9acb116..a840dc740116057f8fa7867f85348a676caf04a4 100644 (file)
--- 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()
 
@@ -214,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 */
@@ -235,14 +232,20 @@ export abstract class AbstractPool<
       data: data ?? ({} as Data),
       id: crypto.randomUUID()
     }
-    const res = this.internalExecute(workerNodeKey, workerNode, 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.workerNodes[workerNodeKey].tasksUsage.running > 0)
     ) {
       this.enqueueTask(workerNodeKey, submittedTask)
     } else {
-      this.sendToWorker(workerNode.worker, submittedTask)
+      this.executeTask(workerNodeKey, submittedTask)
     }
     this.checkAndEmitEvents()
     // eslint-disable-next-line @typescript-eslint/return-await
@@ -252,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)
       })
     )
@@ -267,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
@@ -282,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 {
@@ -333,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 (
@@ -431,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>
             )
           }
@@ -448,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) {
@@ -555,8 +539,13 @@ export abstract class AbstractPool<
     this.workerChoiceStrategyContext.remove(workerNodeKey)
   }
 
-  private enqueueTask (workerNodeKey: number, task: Task<Data>): void {
-    this.workerNodes[workerNodeKey].tasksQueue.push(task)
+  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>): number {
+    return this.workerNodes[workerNodeKey].tasksQueue.push(task)
   }
 
   private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
@@ -570,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 = []
     }
   }