X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=f0119f4ed3c8135e596d171a0f6ec093d7b29935;hb=c3eca146177be73305c08b03e16dccf531855e04;hp=1aa1a13413f807019352907d2dac7a635ab665d1;hpb=16f3605c35d8a2ef293a4382c6e846dc27915b4a;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 1aa1a134..f0119f4e 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -179,7 +179,7 @@ export abstract class AbstractPool<
   protected checkDynamicPoolSize (min: number, max: number): void {
     if (this.type === PoolTypes.dynamic) {
       if (max == null) {
-        throw new Error(
+        throw new TypeError(
           'Cannot instantiate a dynamic pool without specifying the maximum pool size'
         )
       } else if (!Number.isSafeInteger(max)) {
@@ -295,15 +295,31 @@ export abstract class AbstractPool<
       !Number.isSafeInteger(tasksQueueOptions.concurrency)
     ) {
       throw new TypeError(
-        'Invalid worker tasks concurrency: must be an integer'
+        'Invalid worker node tasks concurrency: must be an integer'
       )
     }
     if (
       tasksQueueOptions?.concurrency != null &&
       tasksQueueOptions.concurrency <= 0
     ) {
-      throw new Error(
-        `Invalid worker tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
+      throw new RangeError(
+        `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero`
+      )
+    }
+    if (
+      tasksQueueOptions?.queueMaxSize != null &&
+      !Number.isSafeInteger(tasksQueueOptions.queueMaxSize)
+    ) {
+      throw new TypeError(
+        'Invalid worker node tasks queue max size: must be an integer'
+      )
+    }
+    if (
+      tasksQueueOptions?.queueMaxSize != null &&
+      tasksQueueOptions.queueMaxSize <= 0
+    ) {
+      throw new RangeError(
+        `Invalid worker node tasks queue max size: ${tasksQueueOptions.queueMaxSize} is a negative integer or zero`
       )
     }
   }
@@ -620,16 +636,29 @@ export abstract class AbstractPool<
       this.checkValidTasksQueueOptions(tasksQueueOptions)
       this.opts.tasksQueueOptions =
         this.buildTasksQueueOptions(tasksQueueOptions)
+      this.setTasksQueueMaxSize(
+        this.opts.tasksQueueOptions.queueMaxSize as number
+      )
     } else if (this.opts.tasksQueueOptions != null) {
       delete this.opts.tasksQueueOptions
     }
   }
 
+  private setTasksQueueMaxSize (queueMaxSize: number): void {
+    for (const workerNode of this.workerNodes) {
+      workerNode.tasksQueueBackPressureSize = queueMaxSize
+    }
+  }
+
   private buildTasksQueueOptions (
     tasksQueueOptions: TasksQueueOptions
   ): TasksQueueOptions {
     return {
-      concurrency: tasksQueueOptions?.concurrency ?? 1
+      ...{
+        queueMaxSize: Math.pow(this.maxSize, 2),
+        concurrency: 1
+      },
+      ...tasksQueueOptions
     }
   }
 
@@ -850,7 +879,7 @@ export abstract class AbstractPool<
       const taskFunctionWorkerUsage = this.workerNodes[
         workerNodeKey
       ].getTaskFunctionWorkerUsage(
-        message.taskPerformance?.name ?? DEFAULT_TASK_NAME
+        message.taskPerformance?.name as string
       ) as WorkerUsage
       this.updateTaskStatisticsWorkerUsage(taskFunctionWorkerUsage, message)
       this.updateRunTimeWorkerUsage(taskFunctionWorkerUsage, message)
@@ -1113,6 +1142,10 @@ export abstract class AbstractPool<
     this.sendStartupMessageToWorker(workerNodeKey)
     // Send the statistics message to worker.
     this.sendStatisticsMessageToWorker(workerNodeKey)
+    if (this.opts.enableTasksQueue === true) {
+      this.workerNodes[workerNodeKey].onBackPressure =
+        this.tasksStealingOnBackPressure.bind(this)
+    }
   }
 
   /**
@@ -1146,24 +1179,23 @@ export abstract class AbstractPool<
       let minQueuedTasks = Infinity
       let executeTask = false
       for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
-        const workerInfo = this.getWorkerInfo(workerNodeId) as WorkerInfo
+        if (
+          this.workerNodes[workerNodeId].usage.tasks.executing <
+          (this.opts.tasksQueueOptions?.concurrency as number)
+        ) {
+          executeTask = true
+        }
         if (
           workerNodeId !== workerNodeKey &&
-          workerInfo.ready &&
+          workerNode.info.ready &&
           workerNode.usage.tasks.queued === 0
         ) {
-          if (
-            this.workerNodes[workerNodeId].usage.tasks.executing <
-            (this.opts.tasksQueueOptions?.concurrency as number)
-          ) {
-            executeTask = true
-          }
           targetWorkerNodeKey = workerNodeId
           break
         }
         if (
           workerNodeId !== workerNodeKey &&
-          workerInfo.ready &&
+          workerNode.info.ready &&
           workerNode.usage.tasks.queued < minQueuedTasks
         ) {
           minQueuedTasks = workerNode.usage.tasks.queued
@@ -1173,12 +1205,48 @@ export abstract class AbstractPool<
       if (executeTask) {
         this.executeTask(
           targetWorkerNodeKey,
-          this.dequeueTask(workerNodeKey) as Task<Data>
+          this.popTask(workerNodeKey) as Task<Data>
         )
       } else {
         this.enqueueTask(
           targetWorkerNodeKey,
-          this.dequeueTask(workerNodeKey) as Task<Data>
+          this.popTask(workerNodeKey) as Task<Data>
+        )
+      }
+    }
+  }
+
+  private tasksStealingOnBackPressure (workerId: number): void {
+    const sourceWorkerNode =
+      this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
+    const workerNodes = this.workerNodes
+      .filter((workerNode) => workerNode.info.id !== workerId)
+      .sort(
+        (workerNodeA, workerNodeB) =>
+          workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued
+      )
+    for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
+      if (
+        workerNode.info.ready &&
+        sourceWorkerNode.usage.tasks.queued > 0 &&
+        !workerNode.hasBackPressure() &&
+        workerNode.usage.tasks.executing <
+          (this.opts.tasksQueueOptions?.concurrency as number)
+      ) {
+        this.executeTask(
+          workerNodeKey,
+          sourceWorkerNode.popTask() as Task<Data>
+        )
+      } else if (
+        workerNode.info.ready &&
+        sourceWorkerNode.usage.tasks.queued > 0 &&
+        !workerNode.hasBackPressure() &&
+        workerNode.usage.tasks.executing >=
+          (this.opts.tasksQueueOptions?.concurrency as number)
+      ) {
+        this.enqueueTask(
+          workerNodeKey,
+          sourceWorkerNode.popTask() as Task<Data>
         )
       }
     }
@@ -1292,7 +1360,7 @@ export abstract class AbstractPool<
     const workerNode = new WorkerNode<Worker, Data>(
       worker,
       this.worker,
-      this.maxSize
+      this.opts.tasksQueueOptions?.queueMaxSize ?? Math.pow(this.maxSize, 2)
     )
     // Flag the worker node as ready at pool startup.
     if (this.starting) {
@@ -1358,6 +1426,10 @@ export abstract class AbstractPool<
     return this.workerNodes[workerNodeKey].dequeueTask()
   }
 
+  private popTask (workerNodeKey: number): Task<Data> | undefined {
+    return this.workerNodes[workerNodeKey].popTask()
+  }
+
   private tasksQueueSize (workerNodeKey: number): number {
     return this.workerNodes[workerNodeKey].tasksQueueSize()
   }