X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=c56beaa6d801843c043f091963cea385b72eccee;hb=refs%2Ftags%2Fv2.7.4;hp=9aeeebed772b37234c2a8f0e88c5cb45c65f0dce;hpb=b0b55f57cb5e2bc363bc75d84b483c9c29a5d22f;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 9aeeebed..c56beaa6 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,7 +1,7 @@
 import { randomUUID } from 'node:crypto'
 import { performance } from 'node:perf_hooks'
-import { existsSync } from 'node:fs'
-import { type TransferListItem } from 'node:worker_threads'
+import type { TransferListItem } from 'node:worker_threads'
+import { type EventEmitter, EventEmitterAsyncResource } from 'node:events'
 import type {
   MessageValue,
   PromiseResponseWrapper,
@@ -17,14 +17,12 @@ import {
   max,
   median,
   min,
-  round,
-  updateMeasurementStatistics
+  round
 } from '../utils'
 import { KillBehaviors } from '../worker/worker-options'
 import type { TaskFunction } from '../worker/task-functions'
 import {
   type IPool,
-  PoolEmitter,
   PoolEvents,
   type PoolInfo,
   type PoolOptions,
@@ -49,6 +47,12 @@ import {
 import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context'
 import { version } from './version'
 import { WorkerNode } from './worker-node'
+import {
+  checkFilePath,
+  checkValidTasksQueueOptions,
+  checkValidWorkerChoiceStrategy,
+  updateMeasurementStatistics
+} from './utils'
 
 /**
  * Base class that implements some shared logic for all poolifier pools.
@@ -66,7 +70,7 @@ export abstract class AbstractPool<
   public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
 
   /** @inheritDoc */
-  public readonly emitter?: PoolEmitter
+  public emitter?: EventEmitter | EventEmitterAsyncResource
 
   /**
    * The task execution response promise map:
@@ -129,8 +133,8 @@ export abstract class AbstractPool<
         'Cannot start a pool from a worker with the same type as the pool'
       )
     }
+    checkFilePath(this.filePath)
     this.checkNumberOfWorkers(this.numberOfWorkers)
-    this.checkFilePath(this.filePath)
     this.checkPoolOptions(this.opts)
 
     this.chooseWorkerNode = this.chooseWorkerNode.bind(this)
@@ -138,7 +142,7 @@ export abstract class AbstractPool<
     this.enqueueTask = this.enqueueTask.bind(this)
 
     if (this.opts.enableEvents === true) {
-      this.emitter = new PoolEmitter()
+      this.initializeEventEmitter()
     }
     this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext<
     Worker,
@@ -163,19 +167,6 @@ export abstract class AbstractPool<
     this.startTimestamp = performance.now()
   }
 
-  private checkFilePath (filePath: string): void {
-    if (
-      filePath == null ||
-      typeof filePath !== 'string' ||
-      (typeof filePath === 'string' && filePath.trim().length === 0)
-    ) {
-      throw new Error('Please specify a file with a worker implementation')
-    }
-    if (!existsSync(filePath)) {
-      throw new Error(`Cannot find the worker file '${filePath}'`)
-    }
-  }
-
   private checkNumberOfWorkers (numberOfWorkers: number): void {
     if (numberOfWorkers == null) {
       throw new Error(
@@ -194,36 +185,10 @@ export abstract class AbstractPool<
     }
   }
 
-  protected checkDynamicPoolSize (min: number, max: number): void {
-    if (this.type === PoolTypes.dynamic) {
-      if (max == null) {
-        throw new TypeError(
-          'Cannot instantiate a dynamic pool without specifying the maximum pool size'
-        )
-      } else if (!Number.isSafeInteger(max)) {
-        throw new TypeError(
-          'Cannot instantiate a dynamic pool with a non safe integer maximum pool size'
-        )
-      } else if (min > max) {
-        throw new RangeError(
-          'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size'
-        )
-      } else if (max === 0) {
-        throw new RangeError(
-          'Cannot instantiate a dynamic pool with 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'
-        )
-      }
-    }
-  }
-
   private checkPoolOptions (opts: PoolOptions<Worker>): void {
     if (isPlainObject(opts)) {
       this.opts.startWorkers = opts.startWorkers ?? true
-      this.checkValidWorkerChoiceStrategy(
+      checkValidWorkerChoiceStrategy(
         opts.workerChoiceStrategy as WorkerChoiceStrategy
       )
       this.opts.workerChoiceStrategy =
@@ -239,9 +204,7 @@ export abstract class AbstractPool<
       this.opts.enableEvents = opts.enableEvents ?? true
       this.opts.enableTasksQueue = opts.enableTasksQueue ?? false
       if (this.opts.enableTasksQueue) {
-        this.checkValidTasksQueueOptions(
-          opts.tasksQueueOptions as TasksQueueOptions
-        )
+        checkValidTasksQueueOptions(opts.tasksQueueOptions as TasksQueueOptions)
         this.opts.tasksQueueOptions = this.buildTasksQueueOptions(
           opts.tasksQueueOptions as TasksQueueOptions
         )
@@ -251,19 +214,6 @@ export abstract class AbstractPool<
     }
   }
 
-  private checkValidWorkerChoiceStrategy (
-    workerChoiceStrategy: WorkerChoiceStrategy
-  ): void {
-    if (
-      workerChoiceStrategy != null &&
-      !Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)
-    ) {
-      throw new Error(
-        `Invalid worker choice strategy '${workerChoiceStrategy}'`
-      )
-    }
-  }
-
   private checkValidWorkerChoiceStrategyOptions (
     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
   ): void {
@@ -311,41 +261,10 @@ export abstract class AbstractPool<
     }
   }
 
-  private checkValidTasksQueueOptions (
-    tasksQueueOptions: TasksQueueOptions
-  ): void {
-    if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) {
-      throw new TypeError('Invalid tasks queue options: must be a plain object')
-    }
-    if (
-      tasksQueueOptions?.concurrency != null &&
-      !Number.isSafeInteger(tasksQueueOptions.concurrency)
-    ) {
-      throw new TypeError(
-        'Invalid worker node tasks concurrency: must be an integer'
-      )
-    }
-    if (
-      tasksQueueOptions?.concurrency != null &&
-      tasksQueueOptions.concurrency <= 0
-    ) {
-      throw new RangeError(
-        `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
-      )
-    }
-    if (
-      tasksQueueOptions?.size != null &&
-      !Number.isSafeInteger(tasksQueueOptions.size)
-    ) {
-      throw new TypeError(
-        'Invalid worker node tasks queue size: must be an integer'
-      )
-    }
-    if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) {
-      throw new RangeError(
-        `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero`
-      )
-    }
+  private initializeEventEmitter (): void {
+    this.emitter = new EventEmitterAsyncResource({
+      name: `poolifier:${this.type}-${this.worker}-pool`
+    })
   }
 
   /** @inheritDoc */
@@ -572,7 +491,7 @@ export abstract class AbstractPool<
    * @param message - The received message.
    * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
    */
-  private checkMessageWorkerId (message: MessageValue<Response>): void {
+  private checkMessageWorkerId (message: MessageValue<Data | Response>): void {
     if (message.workerId == null) {
       throw new Error('Worker message received without worker id')
     } else if (
@@ -614,7 +533,7 @@ export abstract class AbstractPool<
     workerChoiceStrategy: WorkerChoiceStrategy,
     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
   ): void {
-    this.checkValidWorkerChoiceStrategy(workerChoiceStrategy)
+    checkValidWorkerChoiceStrategy(workerChoiceStrategy)
     this.opts.workerChoiceStrategy = workerChoiceStrategy
     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
       this.opts.workerChoiceStrategy
@@ -659,7 +578,7 @@ export abstract class AbstractPool<
   /** @inheritDoc */
   public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void {
     if (this.opts.enableTasksQueue === true) {
-      this.checkValidTasksQueueOptions(tasksQueueOptions)
+      checkValidTasksQueueOptions(tasksQueueOptions)
       this.opts.tasksQueueOptions =
         this.buildTasksQueueOptions(tasksQueueOptions)
       this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number)
@@ -769,28 +688,38 @@ export abstract class AbstractPool<
     message: MessageValue<Data>
   ): Promise<boolean> {
     return await new Promise<boolean>((resolve, reject) => {
-      const workerId = this.getWorkerInfo(workerNodeKey).id as number
-      this.registerWorkerMessageListener(workerNodeKey, message => {
+      const taskFunctionOperationListener = (
+        message: MessageValue<Response>
+      ): void => {
+        this.checkMessageWorkerId(message)
+        const workerId = this.getWorkerInfo(workerNodeKey).id as number
         if (
-          message.workerId === workerId &&
-          message.taskFunctionOperationStatus === true
-        ) {
-          resolve(true)
-        } else if (
-          message.workerId === workerId &&
-          message.taskFunctionOperationStatus === false
+          message.taskFunctionOperationStatus != null &&
+          message.workerId === workerId
         ) {
-          reject(
-            new Error(
-              `Task function operation '${
-                message.taskFunctionOperation as string
-              }' failed on worker ${message.workerId} with error: '${
-                message.workerError?.message as string
-              }'`
+          if (message.taskFunctionOperationStatus) {
+            resolve(true)
+          } else if (!message.taskFunctionOperationStatus) {
+            reject(
+              new Error(
+                `Task function operation '${
+                  message.taskFunctionOperation as string
+                }' failed on worker ${message.workerId} with error: '${
+                  message.workerError?.message as string
+                }'`
+              )
             )
+          }
+          this.deregisterWorkerMessageListener(
+            this.getWorkerNodeKeyByWorkerId(message.workerId),
+            taskFunctionOperationListener
           )
         }
-      })
+      }
+      this.registerWorkerMessageListener(
+        workerNodeKey,
+        taskFunctionOperationListener
+      )
       this.sendToWorker(workerNodeKey, message)
     })
   }
@@ -799,20 +728,21 @@ export abstract class AbstractPool<
     message: MessageValue<Data>
   ): Promise<boolean> {
     return await new Promise<boolean>((resolve, reject) => {
-      const responsesReceived = new Array<MessageValue<Data | Response>>()
-      for (const [workerNodeKey] of this.workerNodes.entries()) {
-        this.registerWorkerMessageListener(workerNodeKey, message => {
-          if (message.taskFunctionOperationStatus != null) {
-            responsesReceived.push(message)
+      const responsesReceived = new Array<MessageValue<Response>>()
+      const taskFunctionOperationsListener = (
+        message: MessageValue<Response>
+      ): void => {
+        this.checkMessageWorkerId(message)
+        if (message.taskFunctionOperationStatus != null) {
+          responsesReceived.push(message)
+          if (responsesReceived.length === this.workerNodes.length) {
             if (
-              responsesReceived.length === this.workerNodes.length &&
               responsesReceived.every(
                 message => message.taskFunctionOperationStatus === true
               )
             ) {
               resolve(true)
             } else if (
-              responsesReceived.length === this.workerNodes.length &&
               responsesReceived.some(
                 message => message.taskFunctionOperationStatus === false
               )
@@ -832,8 +762,18 @@ export abstract class AbstractPool<
                 )
               )
             }
+            this.deregisterWorkerMessageListener(
+              this.getWorkerNodeKeyByWorkerId(message.workerId),
+              taskFunctionOperationsListener
+            )
           }
-        })
+        }
+      }
+      for (const [workerNodeKey] of this.workerNodes.entries()) {
+        this.registerWorkerMessageListener(
+          workerNodeKey,
+          taskFunctionOperationsListener
+        )
         this.sendToWorker(workerNodeKey, message)
       }
     })
@@ -1004,6 +944,9 @@ export abstract class AbstractPool<
       })
     )
     this.emitter?.emit(PoolEvents.destroy, this.info)
+    if (this.emitter instanceof EventEmitterAsyncResource) {
+      this.emitter?.emitDestroy()
+    }
     this.started = false
   }
 
@@ -1011,19 +954,21 @@ export abstract class AbstractPool<
     workerNodeKey: number
   ): Promise<void> {
     await new Promise<void>((resolve, reject) => {
-      this.registerWorkerMessageListener(workerNodeKey, message => {
+      const killMessageListener = (message: MessageValue<Response>): void => {
+        this.checkMessageWorkerId(message)
         if (message.kill === 'success') {
           resolve()
         } else if (message.kill === 'failure') {
           reject(
             new Error(
-              `Worker ${
+              `Kill message handling failed on worker ${
                 message.workerId as number
-              } kill message handling failed`
+              }`
             )
           )
         }
-      })
+      }
+      this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
       this.sendToWorker(workerNodeKey, { kill: true })
     })
   }
@@ -1307,6 +1252,7 @@ export abstract class AbstractPool<
   protected createAndSetupDynamicWorkerNode (): number {
     const workerNodeKey = this.createAndSetupWorkerNode()
     this.registerWorkerMessageListener(workerNodeKey, message => {
+      this.checkMessageWorkerId(message)
       const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
         message.workerId
       )
@@ -1365,6 +1311,32 @@ export abstract class AbstractPool<
     listener: (message: MessageValue<Message>) => void
   ): void
 
+  /**
+   * Registers once a listener callback on the worker given its worker node key.
+   *
+   * @param workerNodeKey - The worker node key.
+   * @param listener - The message listener callback.
+   */
+  protected abstract registerOnceWorkerMessageListener<
+    Message extends Data | Response
+  >(
+    workerNodeKey: number,
+    listener: (message: MessageValue<Message>) => void
+  ): void
+
+  /**
+   * Deregisters a listener callback on the worker given its worker node key.
+   *
+   * @param workerNodeKey - The worker node key.
+   * @param listener - The message listener callback.
+   */
+  protected abstract deregisterWorkerMessageListener<
+    Message extends Data | Response
+  >(
+    workerNodeKey: number,
+    listener: (message: MessageValue<Message>) => void
+  ): void
+
   /**
    * Method hooked up after a worker node has been newly created.
    * Can be overridden.