test: add kill behavior testing

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index f32aab0dc838409e2514bf0377c4ba9276180579..631434f37618e4727b2a8dcdbee119c309965da1 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -13,9 +13,12 @@ import {
   type IPool,
   PoolEmitter,
   PoolEvents,
+  type PoolInfo,
   type PoolOptions,
-  PoolType,
-  type TasksQueueOptions
+  type PoolType,
+  PoolTypes,
+  type TasksQueueOptions,
+  type WorkerType
 } from './pool'
 import type { IWorker, Task, TasksUsage, WorkerNode } from './worker'
 import {
@@ -133,7 +136,7 @@ export abstract class AbstractPool<
       throw new RangeError(
         'Cannot instantiate a pool with a negative number of workers'
       )
-    } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
+    } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) {
       throw new Error('Cannot instantiate a fixed pool with no worker')
     }
   }
@@ -149,6 +152,7 @@ export abstract class AbstractPool<
       this.checkValidWorkerChoiceStrategyOptions(
         this.opts.workerChoiceStrategyOptions
       )
+      this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true
       this.opts.enableEvents = opts.enableEvents ?? true
       this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
       if (this.opts.enableTasksQueue) {
@@ -184,7 +188,7 @@ export abstract class AbstractPool<
     }
     if (
       workerChoiceStrategyOptions.weights != null &&
-      Object.keys(workerChoiceStrategyOptions.weights).length !== this.size
+      Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize
     ) {
       throw new Error(
         'Invalid worker choice strategy options: must have a weight for each worker node'
@@ -211,30 +215,54 @@ export abstract class AbstractPool<
   public abstract get type (): PoolType
 
   /** @inheritDoc */
-  public abstract get size (): number
+  public get info (): PoolInfo {
+    return {
+      type: this.type,
+      worker: this.worker,
+      minSize: this.minSize,
+      maxSize: this.maxSize,
+      workerNodes: this.workerNodes.length,
+      idleWorkerNodes: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          workerNode.tasksUsage.running === 0 ? accumulator + 1 : accumulator,
+        0
+      ),
+      busyWorkerNodes: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          workerNode.tasksUsage.running > 0 ? accumulator + 1 : accumulator,
+        0
+      ),
+      runningTasks: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          accumulator + workerNode.tasksUsage.running,
+        0
+      ),
+      queuedTasks: this.workerNodes.reduce(
+        (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
+        0
+      ),
+      maxQueuedTasks: this.workerNodes.reduce(
+        (accumulator, workerNode) =>
+          accumulator + workerNode.tasksQueue.maxSize,
+        0
+      )
+    }
+  }
 
   /**
-   * Number of tasks running in the pool.
+   * Gets the worker type.
    */
-  private get numberOfRunningTasks (): number {
-    return this.workerNodes.reduce(
-      (accumulator, workerNode) => accumulator + workerNode.tasksUsage.running,
-      0
-    )
-  }
+  protected abstract get worker (): WorkerType
 
   /**
-   * Number of tasks queued in the pool.
+   * Pool minimum size.
    */
-  private get numberOfQueuedTasks (): number {
-    if (this.opts.enableTasksQueue === false) {
-      return 0
-    }
-    return this.workerNodes.reduce(
-      (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size,
-      0
-    )
-  }
+  protected abstract get minSize (): number
+
+  /**
+   * Pool maximum size.
+   */
+  protected abstract get maxSize (): number
 
   /**
    * Gets the given worker its worker node key.
@@ -382,6 +410,7 @@ export abstract class AbstractPool<
     await Promise.all(
       this.workerNodes.map(async (workerNode, workerNodeKey) => {
         this.flushTasksQueue(workerNodeKey)
+        // FIXME: wait for tasks to be finished
         await this.destroyWorker(workerNode.worker)
       })
     )
@@ -496,7 +525,7 @@ export abstract class AbstractPool<
    */
   protected chooseWorkerNode (): number {
     let workerNodeKey: number
-    if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) {
+    if (this.type === PoolTypes.dynamic && !this.full && this.internalBusy()) {
       const workerCreated = this.createAndSetupWorker()
       this.registerWorkerMessageListener(workerCreated, message => {
         const currentWorkerNodeKey = this.getWorkerNodeKey(workerCreated)
@@ -507,6 +536,7 @@ export abstract class AbstractPool<
         ) {
           // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime)
           this.flushTasksQueue(currentWorkerNodeKey)
+          // FIXME: wait for tasks to be finished
           void (this.destroyWorker(workerCreated) as Promise<void>)
         }
       })
@@ -562,6 +592,16 @@ export abstract class AbstractPool<
 
     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
+    worker.on('error', error => {
+      if (this.emitter != null) {
+        this.emitter.emit(PoolEvents.error, error)
+      }
+    })
+    if (this.opts.restartWorkerOnError === true) {
+      worker.on('error', () => {
+        this.createAndSetupWorker()
+      })
+    }
     worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
     worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
     worker.once('exit', () => {
@@ -588,6 +628,12 @@ export abstract class AbstractPool<
         if (promiseResponse != null) {
           if (message.error != null) {
             promiseResponse.reject(message.error)
+            if (this.emitter != null) {
+              this.emitter.emit(PoolEvents.taskError, {
+                error: message.error,
+                errorData: message.errorData
+              })
+            }
           } else {
             promiseResponse.resolve(message.data as Response)
           }
@@ -609,12 +655,12 @@ export abstract class AbstractPool<
   }
 
   private checkAndEmitEvents (): void {
-    if (this.opts.enableEvents === true) {
+    if (this.emitter != null) {
       if (this.busy) {
-        this.emitter?.emit(PoolEvents.busy)
+        this.emitter?.emit(PoolEvents.busy, this.info)
       }
-      if (this.type === PoolType.DYNAMIC && this.full) {
-        this.emitter?.emit(PoolEvents.full)
+      if (this.type === PoolTypes.dynamic && this.full) {
+        this.emitter?.emit(PoolEvents.full, this.info)
       }
     }
   }
@@ -686,8 +732,10 @@ export abstract class AbstractPool<
    */
   private removeWorkerNode (worker: Worker): void {
     const workerNodeKey = this.getWorkerNodeKey(worker)
-    this.workerNodes.splice(workerNodeKey, 1)
-    this.workerChoiceStrategyContext.remove(workerNodeKey)
+    if (workerNodeKey !== -1) {
+      this.workerNodes.splice(workerNodeKey, 1)
+      this.workerChoiceStrategyContext.remove(workerNodeKey)
+    }
   }
 
   private executeTask (workerNodeKey: number, task: Task<Data>): void {