Better handling for maxInactiveTime on dynamic pools

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 2ffd292afab9a0b540826820935b4a0febd34057..879e3e4f78f2b3ab60a5918a865fa7c711e3d882 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -83,7 +83,7 @@ export abstract class AbstractPool<
 
   /**
    * - `key`: The `Worker`
-   * - `value`: Number of tasks that has been assigned to that worker since it started
+   * - `value`: Number of tasks currently in progress on the worker.
    */
   public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
 
@@ -120,7 +120,6 @@ export abstract class AbstractPool<
     if (!this.filePath) {
       throw new Error('Please specify a file with a worker implementation')
     }
-
     this.setupHook()
 
     for (let i = 1; i <= this.numberOfWorkers; i++) {
@@ -199,6 +198,20 @@ export abstract class AbstractPool<
     }
   }
 
+  /**
+   * Increase the number of tasks that the given workers has done.
+   *
+   * @param worker Workers whose tasks are increased.
+   */
+  protected decreaseWorkersTasks (worker: Worker): void {
+    const numberOfTasksTheWorkerHas = this.tasks.get(worker)
+    if (numberOfTasksTheWorkerHas !== undefined) {
+      this.tasks.set(worker, numberOfTasksTheWorkerHas - 1)
+    } else {
+      throw Error('Worker could not be found in tasks map')
+    }
+  }
+
   /**
    * Removes the given worker from the pool.
    *
@@ -254,7 +267,7 @@ export abstract class AbstractPool<
       const listener: (message: MessageValue<Response>) => void = message => {
         if (message.id === messageId) {
           this.unregisterWorkerMessageListener(worker, listener)
-          this.increaseWorkersTask(worker)
+          this.decreaseWorkersTasks(worker)
           if (message.error) reject(message.error)
           else resolve(message.data as Response)
         }

diff --git a/src/pools/cluster/dynamic.ts b/src/pools/cluster/dynamic.ts
index b0bb697372cdc3312d4cf78200ecc7ceca214cac..44847b61bebe5b502fec575dfddc3fff7eab33ef 100644 (file)
--- a/src/pools/cluster/dynamic.ts
+++ b/src/pools/cluster/dynamic.ts
@@ -61,6 +61,7 @@ export class DynamicClusterPool<
     // All workers are busy, create a new worker
     const worker = this.createAndSetupWorker()
     this.registerWorkerMessageListener<Data>(worker, message => {
+      const tasksInProgress = this.tasks.get(worker)
       if (message.kill) {
         this.sendToWorker(worker, { kill: 1 })
         void this.destroyWorker(worker)

diff --git a/src/pools/thread/dynamic.ts b/src/pools/thread/dynamic.ts
index fddae4662e8b4a4a2b68baf043015fa498d78b4f..92a5fdf2b20da75e14f08d1c1e98e8637314a3f8 100644 (file)
--- a/src/pools/thread/dynamic.ts
+++ b/src/pools/thread/dynamic.ts
@@ -61,7 +61,11 @@ export class DynamicThreadPool<
     // All workers are busy, create a new worker
     const worker = this.createAndSetupWorker()
     this.registerWorkerMessageListener<Data>(worker, message => {
-      if (message.kill) {
+      const tasksInProgress = this.tasks.get(worker)
+      if (message.kill && !tasksInProgress) {
+        // Kill received from the worker, means that no new tasks are submitted to that worker for a while( > maxInactiveTime)
+        // To handle the case of a long-running task we will check if the there is any active task
+        console.log('Here we are')
         this.sendToWorker(worker, { kill: 1 })
         void this.destroyWorker(worker)
       }

diff --git a/tests/pools/cluster/dynamic.test.js b/tests/pools/cluster/dynamic.test.js
index 6a82b05712a5f5dac09ddedc324f608c78db8b0d..024e26707e66900f3844256334142272888059d3 100644 (file)
--- a/tests/pools/cluster/dynamic.test.js
+++ b/tests/pools/cluster/dynamic.test.js
@@ -87,4 +87,19 @@ describe('Dynamic cluster pool test suite ', () => {
     const res = await pool1.execute({ test: 'test' })
     expect(res).toBeFalsy()
   })
+  it('Verify scale processes up and down is working when long running task is used', async () => {
+    const longRunningPool = new DynamicClusterPool(
+      min,
+      max,
+      './tests/worker/thread/longRunningWorker.js'
+    )
+    expect(longRunningPool.workers.length).toBe(min)
+    for (let i = 0; i < max * 10; i++) {
+      longRunningPool.execute({ test: 'test' })
+    }
+    expect(longRunningPool.workers.length).toBe(max)
+    await new Promise(resolve => setTimeout(resolve, 1000))
+    // Here we expect the workers to be at the max size since that the task is still running
+    expect(longRunningPool.workers.length).toBe(max)
+  })
 })

diff --git a/tests/pools/thread/dynamic.test.js b/tests/pools/thread/dynamic.test.js
index 98a0dd0f0d180076f4a87a17f7e9bbe1ee050310..d59b98587c4f6c225ba9d0f6038b1a54bc6d2f33 100644 (file)
--- a/tests/pools/thread/dynamic.test.js
+++ b/tests/pools/thread/dynamic.test.js
@@ -53,6 +53,7 @@ describe('Dynamic thread pool test suite ', () => {
     await new Promise(resolve => setTimeout(resolve, 1000))
     expect(pool.workers.length).toBe(min)
   })
+
   it('Shutdown test', async () => {
     let closedThreads = 0
     pool.workers.forEach(w => {
@@ -85,4 +86,24 @@ describe('Dynamic thread pool test suite ', () => {
     const res = await pool1.execute({ test: 'test' })
     expect(res).toBeFalsy()
   })
+
+  it('Verify scale thread up and down is working when long running task is used', async () => {
+    const longRunningPool = new DynamicThreadPool(
+      min,
+      max,
+      './tests/worker/thread/longRunningWorker.js',
+      {
+        errorHandler: e => console.error(e),
+        onlineHandler: () => console.log('worker is online')
+      }
+    )
+    expect(longRunningPool.workers.length).toBe(min)
+    for (let i = 0; i < max * 10; i++) {
+      longRunningPool.execute({ test: 'test' })
+    }
+    expect(longRunningPool.workers.length).toBe(max)
+    await new Promise(resolve => setTimeout(resolve, 1000))
+    // Here we expect the workers to be at the max size since that the task is still running
+    expect(longRunningPool.workers.length).toBe(max)
+  })
 })

diff --git a/tests/worker/cluster/longRunningWorker.js b/tests/worker/cluster/longRunningWorker.js
new file mode 100644 (file)
index 0000000..d751d35
--- /dev/null
+++ b/tests/worker/cluster/longRunningWorker.js
@@ -0,0 +1,10 @@
+'use strict'
+const { ClusterWorker } = require('../../../lib/index')
+
+async function sleep (data) {
+  return new Promise((resolve, reject) => {
+    setTimeout(() => resolve(data), 50000)
+  })
+}
+
+module.exports = new ClusterWorker(sleep, { maxInactiveTime: 500, async: true })

diff --git a/tests/worker/thread/longRunningWorker.js b/tests/worker/thread/longRunningWorker.js
new file mode 100644 (file)
index 0000000..8689127
--- /dev/null
+++ b/tests/worker/thread/longRunningWorker.js
@@ -0,0 +1,10 @@
+'use strict'
+const { ThreadWorker } = require('../../../lib/index')
+
+async function sleep (data) {
+  return new Promise((resolve, reject) => {
+    setTimeout(() => resolve(data), 50000)
+  })
+}
+
+module.exports = new ThreadWorker(sleep, { maxInactiveTime: 500, async: true })