From 1f0766e784d0391b6ed084f2fcbc5b7e09d25fa1 Mon Sep 17 00:00:00 2001
From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= <jerome.benoit@piment-noir.org>
Date: Sat, 26 Aug 2023 16:26:55 +0200
Subject: [PATCH] perf: fine tune continuous task stealing algorithm
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                     |  4 ++++
 src/pools/worker-node.ts         |  6 +++++-
 tests/pools/thread/fixed.test.js | 13 +++++++++++++
 tests/utils.test.js              | 11 ++++-------
 4 files changed, 26 insertions(+), 8 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 020b4b61..941e085e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- Make continuous tasks stealing algorithm less aggressive.
+
 ## [2.6.35] - 2023-08-25
 
 ### Fixed
diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts
index 9ecfd012..fbd00096 100644
--- a/src/pools/worker-node.ts
+++ b/src/pools/worker-node.ts
@@ -178,7 +178,11 @@ implements IWorkerNode<Worker, Data> {
   }
 
   private async startOnEmptyQueue (): Promise<void> {
-    if (this.tasksQueue.size > 0) {
+    if (
+      this.onEmptyQueueCount > 0 &&
+      this.usage.tasks.executing > 0 &&
+      this.tasksQueue.size > 0
+    ) {
       this.onEmptyQueueCount = 0
       return
     }
diff --git a/tests/pools/thread/fixed.test.js b/tests/pools/thread/fixed.test.js
index 8cd337ee..29ccd827 100644
--- a/tests/pools/thread/fixed.test.js
+++ b/tests/pools/thread/fixed.test.js
@@ -124,7 +124,9 @@ describe('Fixed thread pool test suite', () => {
       expect(workerNode.usage.tasks.maxQueued).toBe(
         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
       )
+      expect(workerNode.usage.tasks.stolen).toBe(0)
     }
+    expect(queuePool.info.executedTasks).toBe(0)
     expect(queuePool.info.executingTasks).toBe(
       numberOfThreads * queuePool.opts.tasksQueueOptions.concurrency
     )
@@ -137,6 +139,7 @@ describe('Fixed thread pool test suite', () => {
         (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency)
     )
     expect(queuePool.info.backPressure).toBe(false)
+    expect(queuePool.info.stolenTasks).toBe(0)
     await Promise.all(promises)
     for (const workerNode of queuePool.workerNodes) {
       expect(workerNode.usage.tasks.executing).toBeGreaterThanOrEqual(0)
@@ -148,7 +151,17 @@ describe('Fixed thread pool test suite', () => {
       expect(workerNode.usage.tasks.maxQueued).toBe(
         maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency
       )
+      expect(workerNode.usage.tasks.stolen).toBeGreaterThanOrEqual(0)
+      expect(workerNode.usage.tasks.stolen).toBeLessThanOrEqual(
+        numberOfThreads * maxMultiplier
+      )
     }
+    expect(queuePool.info.executedTasks).toBe(numberOfThreads * maxMultiplier)
+    expect(queuePool.info.backPressure).toBe(false)
+    expect(queuePool.info.stolenTasks).toBeGreaterThanOrEqual(0)
+    expect(queuePool.info.stolenTasks).toBeLessThanOrEqual(
+      numberOfThreads * maxMultiplier
+    )
   })
 
   it('Verify that is possible to have a worker that return undefined', async () => {
diff --git a/tests/utils.test.js b/tests/utils.test.js
index 0f72aa42..6cfe99ae 100644
--- a/tests/utils.test.js
+++ b/tests/utils.test.js
@@ -34,13 +34,10 @@ describe('Utils test suite', () => {
   })
 
   it('Verify exponentialDelay() behavior', () => {
-    expect(typeof exponentialDelay(randomInt(1000)) === 'number').toBe(true)
-    expect(exponentialDelay(randomInt(1000))).toBeGreaterThanOrEqual(
-      Number.MIN_VALUE
-    )
-    expect(exponentialDelay(randomInt(1000))).toBeLessThanOrEqual(
-      Number.MAX_VALUE
-    )
+    const delay = exponentialDelay(randomInt(1000))
+    expect(typeof delay === 'number').toBe(true)
+    expect(delay).toBeGreaterThanOrEqual(Number.MIN_VALUE)
+    expect(delay).toBeLessThanOrEqual(Number.MAX_VALUE)
   })
 
   it('Verify average() computation', () => {
-- 
2.34.1