From 4e377863ebacc8cead467297d0365e93dec491c3 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Fri, 21 Jul 2023 21:22:14 +0200 Subject: [PATCH] fix: untangle tasks queuing condition from pool busyness MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- CHANGELOG.md | 1 + src/pools/abstract-pool.ts | 10 +++++----- tests/pools/cluster/fixed.test.js | 30 ++++++++++++++++++++---------- tests/pools/thread/fixed.test.js | 30 ++++++++++++++++++++---------- 4 files changed, 46 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d424256f..d4295cd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Fix queued tasks redistribution on error task execution starvation. +- Ensure task queueing per worker condition is untangled from the pool busyness semantic. ### Changed diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 7c71c855..fa3406a3 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -642,14 +642,14 @@ export abstract class AbstractPool< workerNodeKey }) if ( - this.opts.enableTasksQueue === true && - (this.busy || - this.workerNodes[workerNodeKey].usage.tasks.executing >= + this.opts.enableTasksQueue === false || + (this.opts.enableTasksQueue === true && + this.workerNodes[workerNodeKey].usage.tasks.executing < (this.opts.tasksQueueOptions?.concurrency as number)) ) { - this.enqueueTask(workerNodeKey, task) - } else { this.executeTask(workerNodeKey, task) + } else { + this.enqueueTask(workerNodeKey, task) } this.checkAndEmitEvents() }) diff --git a/tests/pools/cluster/fixed.test.js b/tests/pools/cluster/fixed.test.js index f71529a9..0415d9c1 100644 --- a/tests/pools/cluster/fixed.test.js +++ b/tests/pools/cluster/fixed.test.js @@ -5,6 +5,7 @@ const { waitPoolEvents, waitWorkerEvents } = require('../../test-utils') describe('Fixed cluster pool test suite', () => { const numberOfWorkers = 6 + const tasksConcurrency = 2 const pool = new FixedClusterPool( numberOfWorkers, './tests/worker-files/cluster/testWorker.js', @@ -18,7 +19,7 @@ describe('Fixed cluster pool test suite', () => { { enableTasksQueue: true, tasksQueueOptions: { - concurrency: 2 + concurrency: tasksConcurrency }, errorHandler: e => console.error(e) } @@ -104,7 +105,7 @@ describe('Fixed cluster pool test suite', () => { it('Verify that tasks queuing is working', async () => { const promises = new Set() - const maxMultiplier = 2 + const maxMultiplier = 3 // Must be greater than tasksConcurrency for (let i = 0; i < numberOfWorkers * maxMultiplier; i++) { promises.add(queuePool.execute()) } @@ -114,23 +115,32 @@ describe('Fixed cluster pool test suite', () => { queuePool.opts.tasksQueueOptions.concurrency ) expect(workerNode.usage.tasks.executed).toBe(0) - expect(workerNode.usage.tasks.queued).toBeGreaterThan(0) - expect(workerNode.usage.tasks.maxQueued).toBeGreaterThan(0) + expect(workerNode.usage.tasks.queued).toBe( + maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency + ) + expect(workerNode.usage.tasks.maxQueued).toBe( + maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency + ) } - expect(queuePool.info.executingTasks).toBe(numberOfWorkers) + expect(queuePool.info.executingTasks).toBe( + numberOfWorkers * queuePool.opts.tasksQueueOptions.concurrency + ) expect(queuePool.info.queuedTasks).toBe( - numberOfWorkers * maxMultiplier - numberOfWorkers + numberOfWorkers * + (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency) ) expect(queuePool.info.maxQueuedTasks).toBe( - numberOfWorkers * maxMultiplier - numberOfWorkers + numberOfWorkers * + (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency) ) await Promise.all(promises) for (const workerNode of queuePool.workerNodes) { expect(workerNode.usage.tasks.executing).toBe(0) - expect(workerNode.usage.tasks.executed).toBeGreaterThan(0) - expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(maxMultiplier) + expect(workerNode.usage.tasks.executed).toBe(maxMultiplier) expect(workerNode.usage.tasks.queued).toBe(0) - expect(workerNode.usage.tasks.maxQueued).toBe(1) + expect(workerNode.usage.tasks.maxQueued).toBe( + maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency + ) } }) diff --git a/tests/pools/thread/fixed.test.js b/tests/pools/thread/fixed.test.js index 99b86778..e4b99a63 100644 --- a/tests/pools/thread/fixed.test.js +++ b/tests/pools/thread/fixed.test.js @@ -5,6 +5,7 @@ const { waitPoolEvents, waitWorkerEvents } = require('../../test-utils') describe('Fixed thread pool test suite', () => { const numberOfThreads = 6 + const tasksConcurrency = 2 const pool = new FixedThreadPool( numberOfThreads, './tests/worker-files/thread/testWorker.js', @@ -18,7 +19,7 @@ describe('Fixed thread pool test suite', () => { { enableTasksQueue: true, tasksQueueOptions: { - concurrency: 2 + concurrency: tasksConcurrency }, errorHandler: e => console.error(e) } @@ -104,7 +105,7 @@ describe('Fixed thread pool test suite', () => { it('Verify that tasks queuing is working', async () => { const promises = new Set() - const maxMultiplier = 2 + const maxMultiplier = 3 // Must be greater than tasksConcurrency for (let i = 0; i < numberOfThreads * maxMultiplier; i++) { promises.add(queuePool.execute()) } @@ -114,23 +115,32 @@ describe('Fixed thread pool test suite', () => { queuePool.opts.tasksQueueOptions.concurrency ) expect(workerNode.usage.tasks.executed).toBe(0) - expect(workerNode.usage.tasks.queued).toBeGreaterThan(0) - expect(workerNode.usage.tasks.maxQueued).toBeGreaterThan(0) + expect(workerNode.usage.tasks.queued).toBe( + maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency + ) + expect(workerNode.usage.tasks.maxQueued).toBe( + maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency + ) } - expect(queuePool.info.executingTasks).toBe(numberOfThreads) + expect(queuePool.info.executingTasks).toBe( + numberOfThreads * queuePool.opts.tasksQueueOptions.concurrency + ) expect(queuePool.info.queuedTasks).toBe( - numberOfThreads * maxMultiplier - numberOfThreads + numberOfThreads * + (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency) ) expect(queuePool.info.maxQueuedTasks).toBe( - numberOfThreads * maxMultiplier - numberOfThreads + numberOfThreads * + (maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency) ) await Promise.all(promises) for (const workerNode of queuePool.workerNodes) { expect(workerNode.usage.tasks.executing).toBe(0) - expect(workerNode.usage.tasks.executed).toBeGreaterThan(0) - expect(workerNode.usage.tasks.executed).toBeLessThanOrEqual(maxMultiplier) + expect(workerNode.usage.tasks.executed).toBe(maxMultiplier) expect(workerNode.usage.tasks.queued).toBe(0) - expect(workerNode.usage.tasks.maxQueued).toBe(1) + expect(workerNode.usage.tasks.maxQueued).toBe( + maxMultiplier - queuePool.opts.tasksQueueOptions.concurrency + ) } }) -- 2.34.1