Merge branch 'master' of github.com:poolifier/poolifier

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 9da566212877c60c5135f095e812035c2d086260..192725082942b1aa7b8904c07eaf8a2731865f45 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -153,7 +153,25 @@ export abstract class AbstractPool<
         'Cannot instantiate a pool with a negative number of workers'
       )
     } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
-      throw new Error('Cannot instantiate a fixed pool with no worker')
+      throw new RangeError('Cannot instantiate a fixed pool with zero worker')
+    }
+  }
+
+  protected checkDynamicPoolSize (min: number, max: number): void {
+    if (this.type === PoolTypes.dynamic) {
+      if (min > max) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
+        )
+      } else if (min === 0 && max === 0) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a minimum pool size and a maximum pool size equal to zero'
+        )
+      } else if (min === max) {
+        throw new RangeError(
+          'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead'
+        )
+      }
     }
   }
 
@@ -252,6 +270,8 @@ export abstract class AbstractPool<
       version,
       type: this.type,
       worker: this.worker,
+      ready: this.ready,
+      strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy,
       minSize: this.minSize,
       maxSize: this.maxSize,
       ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
@@ -381,6 +401,19 @@ export abstract class AbstractPool<
     }
   }
 
+  private get starting (): boolean {
+    return (
+      !this.full ||
+      (this.full && this.workerNodes.some(workerNode => !workerNode.info.ready))
+    )
+  }
+
+  private get ready (): boolean {
+    return (
+      this.full && this.workerNodes.every(workerNode => workerNode.info.ready)
+    )
+  }
+
   /**
    * Gets the approximate pool utilization.
    *
@@ -435,6 +468,17 @@ export abstract class AbstractPool<
       ?.worker
   }
 
+  private checkMessageWorkerId (message: MessageValue<Response>): void {
+    if (
+      message.workerId != null &&
+      this.getWorkerById(message.workerId) == null
+    ) {
+      throw new Error(
+        `Worker message received from unknown worker '${message.workerId}'`
+      )
+    }
+  }
+
   /**
    * Gets the given worker its worker node key.
    *
@@ -546,6 +590,7 @@ export abstract class AbstractPool<
       // 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) => {
@@ -864,6 +909,13 @@ export abstract class AbstractPool<
   protected afterWorkerSetup (worker: Worker): void {
     // Listen to worker messages.
     this.registerWorkerMessageListener(worker, this.workerListener())
+    // Send startup message to worker.
+    this.sendToWorker(worker, {
+      ready: false,
+      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
+    })
+    // Setup worker task statistics computation.
+    this.setWorkerStatistics(worker)
   }
 
   /**
@@ -883,7 +935,7 @@ export abstract class AbstractPool<
       if (this.opts.enableTasksQueue === true) {
         this.redistributeQueuedTasks(worker)
       }
-      if (this.opts.restartWorkerOnError === true) {
+      if (this.opts.restartWorkerOnError === true && !this.starting) {
         if (this.getWorkerInfo(this.getWorkerNodeKey(worker)).dynamic) {
           this.createAndSetupDynamicWorker()
         } else {
@@ -899,8 +951,6 @@ export abstract class AbstractPool<
 
     this.pushWorkerNode(worker)
 
-    this.setWorkerStatistics(worker)
-
     this.afterWorkerSetup(worker)
 
     return worker
@@ -941,7 +991,6 @@ export abstract class AbstractPool<
    */
   protected createAndSetupDynamicWorker (): Worker {
     const worker = this.createAndSetupWorker()
-    this.getWorkerInfo(this.getWorkerNodeKey(worker)).dynamic = true
     this.registerWorkerMessageListener(worker, message => {
       const workerNodeKey = this.getWorkerNodeKey(worker)
       if (
@@ -957,7 +1006,12 @@ export abstract class AbstractPool<
         void (this.destroyWorker(worker) as Promise<void>)
       }
     })
-    this.sendToWorker(worker, { checkAlive: true })
+    const workerInfo = this.getWorkerInfo(this.getWorkerNodeKey(worker))
+    workerInfo.dynamic = true
+    this.sendToWorker(worker, {
+      checkAlive: true,
+      workerId: workerInfo.id as number
+    })
     return worker
   }
 
@@ -968,9 +1022,10 @@ export abstract class AbstractPool<
    */
   protected workerListener (): (message: MessageValue<Response>) => void {
     return message => {
-      if (message.workerId != null && message.started != null) {
-        // Worker started message received
-        this.handleWorkerStartedMessage(message)
+      this.checkMessageWorkerId(message)
+      if (message.ready != null && message.workerId != null) {
+        // Worker ready message received
+        this.handleWorkerReadyMessage(message)
       } else if (message.id != null) {
         // Task execution response received
         this.handleTaskExecutionResponse(message)
@@ -978,17 +1033,12 @@ export abstract class AbstractPool<
     }
   }
 
-  private handleWorkerStartedMessage (message: MessageValue<Response>): void {
-    const worker = this.getWorkerById(message.workerId as number)
-    if (worker != null) {
-      this.getWorkerInfo(this.getWorkerNodeKey(worker)).started =
-        message.started as boolean
-    } else {
-      throw new Error(
-        `Worker started message received from unknown worker '${
-          message.workerId as number
-        }'`
-      )
+  private handleWorkerReadyMessage (message: MessageValue<Response>): void {
+    const worker = this.getWorkerById(message.workerId)
+    this.getWorkerInfo(this.getWorkerNodeKey(worker as Worker)).ready =
+      message.ready as boolean
+    if (this.emitter != null && this.ready) {
+      this.emitter.emit(PoolEvents.ready, this.info)
     }
   }
 
@@ -1103,7 +1153,8 @@ export abstract class AbstractPool<
             .runTime.aggregate,
         elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
           .elu.aggregate
-      }
+      },
+      workerId: this.getWorkerInfo(this.getWorkerNodeKey(worker)).id as number
     })
   }
 }