refactor: refine error message at pool instantiation inside a worker

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index aec731a5c189903b46230ceceeb831768fa62425..5633613ecebdda8cb763c7922588ebc504be4fc7 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,6 +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 {
   MessageValue,
   PromiseResponseWrapper,
@@ -105,7 +106,9 @@ export abstract class AbstractPool<
     protected readonly opts: PoolOptions<Worker>
   ) {
     if (!this.isMain()) {
-      throw new Error('Cannot start a pool from a worker!')
+      throw new Error(
+        'Cannot start a pool from the same worker type as the current pool one'
+      )
     }
     this.checkNumberOfWorkers(this.numberOfWorkers)
     this.checkFilePath(this.filePath)
@@ -359,14 +362,14 @@ export abstract class AbstractPool<
           minimum: round(
             Math.min(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.runTime?.minimum ?? Infinity
+                (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity
               )
             )
           ),
           maximum: round(
             Math.max(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.runTime?.maximum ?? -Infinity
+                (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity
               )
             )
           ),
@@ -387,7 +390,7 @@ export abstract class AbstractPool<
             median: round(
               median(
                 this.workerNodes.map(
-                  workerNode => workerNode.usage.runTime?.median ?? 0
+                  (workerNode) => workerNode.usage.runTime?.median ?? 0
                 )
               )
             )
@@ -400,14 +403,14 @@ export abstract class AbstractPool<
           minimum: round(
             Math.min(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.waitTime?.minimum ?? Infinity
+                (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity
               )
             )
           ),
           maximum: round(
             Math.max(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity
+                (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity
               )
             )
           ),
@@ -428,7 +431,7 @@ export abstract class AbstractPool<
             median: round(
               median(
                 this.workerNodes.map(
-                  workerNode => workerNode.usage.waitTime?.median ?? 0
+                  (workerNode) => workerNode.usage.waitTime?.median ?? 0
                 )
               )
             )
@@ -521,7 +524,7 @@ export abstract class AbstractPool<
    */
   private getWorkerNodeKeyByWorker (worker: Worker): number {
     return this.workerNodes.findIndex(
-      workerNode => workerNode.worker === worker
+      (workerNode) => workerNode.worker === worker
     )
   }
 
@@ -533,7 +536,7 @@ export abstract class AbstractPool<
    */
   private getWorkerNodeKeyByWorkerId (workerId: number): number {
     return this.workerNodes.findIndex(
-      workerNode => workerNode.info.id === workerId
+      (workerNode) => workerNode.info.id === workerId
     )
   }
 
@@ -615,28 +618,50 @@ export abstract class AbstractPool<
   protected abstract get busy (): boolean
 
   /**
-   * Whether worker nodes are executing at least one task.
+   * Whether worker nodes are executing concurrently their tasks quota or not.
    *
    * @returns Worker nodes busyness boolean status.
    */
   protected internalBusy (): boolean {
-    return (
-      this.workerNodes.findIndex(
-        workerNode =>
-          workerNode.info.ready && workerNode.usage.tasks.executing === 0
-      ) === -1
-    )
+    if (this.opts.enableTasksQueue === true) {
+      return (
+        this.workerNodes.findIndex(
+          (workerNode) =>
+            workerNode.info.ready &&
+            workerNode.usage.tasks.executing <
+              (this.opts.tasksQueueOptions?.concurrency as number)
+        ) === -1
+      )
+    } else {
+      return (
+        this.workerNodes.findIndex(
+          (workerNode) =>
+            workerNode.info.ready && workerNode.usage.tasks.executing === 0
+        ) === -1
+      )
+    }
   }
 
   /** @inheritDoc */
-  public async execute (data?: Data, name?: string): Promise<Response> {
+  public async execute (
+    data?: Data,
+    name?: string,
+    transferList?: TransferListItem[]
+  ): Promise<Response> {
     return await new Promise<Response>((resolve, reject) => {
+      if (name != null && typeof name !== 'string') {
+        reject(new TypeError('name argument must be a string'))
+      }
+      if (transferList != null && !Array.isArray(transferList)) {
+        reject(new TypeError('transferList argument must be an array'))
+      }
       const timestamp = performance.now()
       const workerNodeKey = this.chooseWorkerNode()
       const task: Task<Data> = {
         name: name ?? DEFAULT_TASK_NAME,
         // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
         data: data ?? ({} as Data),
+        transferList,
         timestamp,
         workerId: this.getWorkerInfo(workerNodeKey).id as number,
         taskId: randomUUID()
@@ -839,10 +864,12 @@ export abstract class AbstractPool<
    *
    * @param workerNodeKey - The worker node key.
    * @param message - The message.
+   * @param transferList - The optional array of transferable objects.
    */
   protected abstract sendToWorker (
     workerNodeKey: number,
-    message: MessageValue<Data>
+    message: MessageValue<Data>,
+    transferList?: TransferListItem[]
   ): void
 
   /**
@@ -862,7 +889,7 @@ export abstract class AbstractPool<
 
     worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
     worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
-    worker.on('error', error => {
+    worker.on('error', (error) => {
       const workerNodeKey = this.getWorkerNodeKeyByWorker(worker)
       const workerInfo = this.getWorkerInfo(workerNodeKey)
       workerInfo.ready = false
@@ -899,7 +926,7 @@ export abstract class AbstractPool<
    */
   protected createAndSetupDynamicWorkerNode (): number {
     const workerNodeKey = this.createAndSetupWorkerNode()
-    this.registerWorkerMessageListener(workerNodeKey, message => {
+    this.registerWorkerMessageListener(workerNodeKey, (message) => {
       const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
         message.workerId
       )
@@ -1032,7 +1059,7 @@ export abstract class AbstractPool<
    * @returns The listener function to execute when a message is received from a worker.
    */
   protected workerListener (): (message: MessageValue<Response>) => void {
-    return message => {
+    return (message) => {
       this.checkMessageWorkerId(message)
       if (message.ready != null) {
         // Worker ready response received from worker
@@ -1145,7 +1172,7 @@ export abstract class AbstractPool<
    */
   private executeTask (workerNodeKey: number, task: Task<Data>): void {
     this.beforeTaskExecutionHook(workerNodeKey, task)
-    this.sendToWorker(workerNodeKey, task)
+    this.sendToWorker(workerNodeKey, task, task.transferList)
   }
 
   private enqueueTask (workerNodeKey: number, task: Task<Data>): number {