fix: fix task stealing related options handling at runtime

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>

diff --git a/.c8rc.json b/.c8rc.json
index 29100bb26408a1226a5c4dc10cf04e3e8e0736a9..aba3e9b0c0cb8e3ea21eb9f372221645bef54aca 100644 (file)
--- a/.c8rc.json
+++ b/.c8rc.json
@@ -3,5 +3,5 @@
   "lines": 94,
   "statements": 94,
   "functions": 95,
-  "branches": 92
+  "branches": 93
 }

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 9aa6ec3dfeab90f8ba462920290f8a7b74b91f80..eb4c415b4a21be72446bb2da22db83ea9891acb7 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -630,6 +630,8 @@ export abstract class AbstractPool<
     tasksQueueOptions?: TasksQueueOptions
   ): void {
     if (this.opts.enableTasksQueue === true && !enable) {
+      this.unsetTaskStealing()
+      this.unsetTasksStealingOnBackPressure()
       this.flushTasksQueues()
     }
     this.opts.enableTasksQueue = enable
@@ -643,6 +645,16 @@ export abstract class AbstractPool<
       this.opts.tasksQueueOptions =
         this.buildTasksQueueOptions(tasksQueueOptions)
       this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number)
+      if (this.opts.tasksQueueOptions.taskStealing === true) {
+        this.setTaskStealing()
+      } else {
+        this.unsetTaskStealing()
+      }
+      if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) {
+        this.setTasksStealingOnBackPressure()
+      } else {
+        this.unsetTasksStealingOnBackPressure()
+      }
     } else if (this.opts.tasksQueueOptions != null) {
       delete this.opts.tasksQueueOptions
     }
@@ -654,6 +666,32 @@ export abstract class AbstractPool<
     }
   }
 
+  private setTaskStealing (): void {
+    for (const [workerNodeKey] of this.workerNodes.entries()) {
+      this.workerNodes[workerNodeKey].onEmptyQueue =
+        this.taskStealingOnEmptyQueue.bind(this)
+    }
+  }
+
+  private unsetTaskStealing (): void {
+    for (const [workerNodeKey] of this.workerNodes.entries()) {
+      delete this.workerNodes[workerNodeKey].onEmptyQueue
+    }
+  }
+
+  private setTasksStealingOnBackPressure (): void {
+    for (const [workerNodeKey] of this.workerNodes.entries()) {
+      this.workerNodes[workerNodeKey].onBackPressure =
+        this.tasksStealingOnBackPressure.bind(this)
+    }
+  }
+
+  private unsetTasksStealingOnBackPressure (): void {
+    for (const [workerNodeKey] of this.workerNodes.entries()) {
+      delete this.workerNodes[workerNodeKey].onBackPressure
+    }
+  }
+
   private buildTasksQueueOptions (
     tasksQueueOptions: TasksQueueOptions
   ): TasksQueueOptions {

diff --git a/tests/pools/abstract/abstract-pool.test.js b/tests/pools/abstract/abstract-pool.test.js
index 719acd766781fed8c6f9c3e46b4cc82d1a3e28a2..aeede25d7149a9c5918626ae9e860e7733cea4b4 100644 (file)
--- a/tests/pools/abstract/abstract-pool.test.js
+++ b/tests/pools/abstract/abstract-pool.test.js
@@ -630,6 +630,10 @@ describe('Abstract pool test suite', () => {
     )
     expect(pool.opts.enableTasksQueue).toBe(false)
     expect(pool.opts.tasksQueueOptions).toBeUndefined()
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
     pool.enableTasksQueue(true)
     expect(pool.opts.enableTasksQueue).toBe(true)
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
@@ -638,6 +642,10 @@ describe('Abstract pool test suite', () => {
       taskStealing: true,
       tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     pool.enableTasksQueue(true, { concurrency: 2 })
     expect(pool.opts.enableTasksQueue).toBe(true)
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
@@ -646,9 +654,17 @@ describe('Abstract pool test suite', () => {
       taskStealing: true,
       tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     pool.enableTasksQueue(false)
     expect(pool.opts.enableTasksQueue).toBe(false)
     expect(pool.opts.tasksQueueOptions).toBeUndefined()
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
     await pool.destroy()
   })
 
@@ -664,13 +680,40 @@ describe('Abstract pool test suite', () => {
       taskStealing: true,
       tasksStealingOnBackPressure: true
     })
-    pool.setTasksQueueOptions({ concurrency: 2 })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
+    pool.setTasksQueueOptions({
+      concurrency: 2,
+      taskStealing: false,
+      tasksStealingOnBackPressure: false
+    })
     expect(pool.opts.tasksQueueOptions).toStrictEqual({
       concurrency: 2,
       size: 4,
+      taskStealing: false,
+      tasksStealingOnBackPressure: false
+    })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeUndefined()
+      expect(workerNode.onBackPressure).toBeUndefined()
+    }
+    pool.setTasksQueueOptions({
+      concurrency: 1,
+      taskStealing: true,
+      tasksStealingOnBackPressure: true
+    })
+    expect(pool.opts.tasksQueueOptions).toStrictEqual({
+      concurrency: 1,
+      size: 4,
       taskStealing: true,
       tasksStealingOnBackPressure: true
     })
+    for (const workerNode of pool.workerNodes) {
+      expect(workerNode.onEmptyQueue).toBeInstanceOf(Function)
+      expect(workerNode.onBackPressure).toBeInstanceOf(Function)
+    }
     expect(() =>
       pool.setTasksQueueOptions('invalidTasksQueueOptions')
     ).toThrowError(