From 0bc682672927a4dcc21c191d4ffc721602d9c802 Mon Sep 17 00:00:00 2001
From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= <jerome.benoit@piment-noir.org>
Date: Thu, 24 Aug 2023 10:16:16 +0200
Subject: [PATCH] fix: avoid cascading tasks stealing under back pressure
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>
---
 CHANGELOG.md                                  |  8 +++++
 src/pools/abstract-pool.ts                    | 30 +++++++++++--------
 ...hted-round-robin-worker-choice-strategy.ts |  2 +-
 3 files changed, 27 insertions(+), 13 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ab21a077..2b2a8741 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixes
+
+- Avoid cascading tasks stealing under back pressure.
+
+### Changed
+
+- Add fastpath to queued tasks rescheduling.
+
 ## [2.6.33] - 2023-08-24
 
 ### Fixed
diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 7b0d541d..507fcc10 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1190,7 +1190,7 @@ export abstract class AbstractPool<
 
   private redistributeQueuedTasks (workerNodeKey: number): void {
     while (this.tasksQueueSize(workerNodeKey) > 0) {
-      let destinationWorkerNodeKey: number = workerNodeKey
+      let destinationWorkerNodeKey!: number
       let minQueuedTasks = Infinity
       let executeTask = false
       for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
@@ -1211,15 +1211,17 @@ export abstract class AbstractPool<
           }
         }
       }
-      const task = {
-        ...(this.dequeueTask(workerNodeKey) as Task<Data>),
-        workerId: (this.getWorkerInfo(destinationWorkerNodeKey) as WorkerInfo)
-          .id as number
-      }
-      if (executeTask) {
-        this.executeTask(destinationWorkerNodeKey, task)
-      } else {
-        this.enqueueTask(destinationWorkerNodeKey, task)
+      if (destinationWorkerNodeKey != null) {
+        const task = {
+          ...(this.dequeueTask(workerNodeKey) as Task<Data>),
+          workerId: (this.getWorkerInfo(destinationWorkerNodeKey) as WorkerInfo)
+            .id as number
+        }
+        if (executeTask) {
+          this.executeTask(destinationWorkerNodeKey, task)
+        } else {
+          this.enqueueTask(destinationWorkerNodeKey, task)
+        }
       }
     }
   }
@@ -1234,6 +1236,9 @@ export abstract class AbstractPool<
           workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
       )
     for (const sourceWorkerNode of workerNodes) {
+      if (sourceWorkerNode.usage.tasks.queued === 0) {
+        break
+      }
       if (
         sourceWorkerNode.info.ready &&
         sourceWorkerNode.info.id !== workerId &&
@@ -1267,10 +1272,11 @@ export abstract class AbstractPool<
       )
     for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
       if (
+        sourceWorkerNode.usage.tasks.queued > 0 &&
         workerNode.info.ready &&
         workerNode.info.id !== workerId &&
-        sourceWorkerNode.usage.tasks.queued > 0 &&
-        !workerNode.hasBackPressure()
+        workerNode.usage.tasks.queued <
+          (this.opts.tasksQueueOptions?.size as number) - 1
       ) {
         const task = {
           ...(sourceWorkerNode.popTask() as Task<Data>),
diff --git a/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts b/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts
index 0fc1e0da..86a7f5ca 100644
--- a/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts
+++ b/src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts
@@ -60,7 +60,7 @@ export class InterleavedWeightedRoundRobinWorkerChoiceStrategy<
 
   /** @inheritDoc */
   public choose (): number | undefined {
-    let roundId: number = this.roundId
+    let roundId!: number
     let workerNodeId: number | undefined
     for (
       let roundIndex = this.roundId;
-- 
2.34.1