X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=tests%2Fpools%2Fworker-node.test.mjs;h=f00b8061b2e085e26f6b742ae2059b5da612c0a5;hb=3b594fe1b0f89d6665da2eb2ebdc14eb7628fe70;hp=b77af2fda507fb5401758e479094495dddd805fb;hpb=75f30e744551af87736f998db9ad02be7206e89e;p=poolifier.git diff --git a/tests/pools/worker-node.test.mjs b/tests/pools/worker-node.test.mjs index b77af2fd..f00b8061 100644 --- a/tests/pools/worker-node.test.mjs +++ b/tests/pools/worker-node.test.mjs @@ -3,7 +3,7 @@ import { MessageChannel, Worker as ThreadWorker } from 'node:worker_threads' import { expect } from 'expect' -import { CircularArray } from '../../lib/circular-array.cjs' +import { CircularBuffer } from '../../lib/circular-buffer.cjs' import { WorkerTypes } from '../../lib/index.cjs' import { WorkerNode } from '../../lib/pools/worker-node.cjs' import { PriorityQueue } from '../../lib/priority-queue.cjs' @@ -13,12 +13,20 @@ describe('Worker node test suite', () => { const threadWorkerNode = new WorkerNode( WorkerTypes.thread, './tests/worker-files/thread/testWorker.mjs', - { tasksQueueBackPressureSize: 12 } + { + tasksQueueBackPressureSize: 12, + tasksQueueBucketSize: 6, + tasksQueuePriority: true, + } ) const clusterWorkerNode = new WorkerNode( WorkerTypes.cluster, './tests/worker-files/cluster/testWorker.cjs', - { tasksQueueBackPressureSize: 12 } + { + tasksQueueBackPressureSize: 12, + tasksQueueBucketSize: 6, + tasksQueuePriority: true, + } ) it('Worker node instantiation', () => { @@ -29,8 +37,7 @@ describe('Worker node test suite', () => { () => new WorkerNode( 'invalidWorkerType', - './tests/worker-files/thread/testWorker.mjs', - { tasksQueueBackPressureSize: 12 } + './tests/worker-files/thread/testWorker.mjs' ) ).toThrow( new TypeError( @@ -57,7 +64,7 @@ describe('Worker node test suite', () => { ) ).toThrow( new TypeError( - 'Cannot construct a worker node with invalid options: must be a plain object' + 'Cannot construct a worker node with invalid worker node options: must be a plain object' ) ) expect( @@ -120,6 +127,102 @@ describe('Worker node test suite', () => { 'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer' ) ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { + tasksQueueBackPressureSize: 12, + } + ) + ).toThrow( + new TypeError( + 'Cannot construct a worker node without a tasks queue bucket size option' + ) + ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { + tasksQueueBackPressureSize: 12, + tasksQueueBucketSize: 'invalidTasksQueueBucketSize', + } + ) + ).toThrow( + new TypeError( + 'Cannot construct a worker node with a tasks queue bucket size option that is not an integer' + ) + ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { tasksQueueBackPressureSize: 12, tasksQueueBucketSize: 0.2 } + ) + ).toThrow( + new TypeError( + 'Cannot construct a worker node with a tasks queue bucket size option that is not an integer' + ) + ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { tasksQueueBackPressureSize: 12, tasksQueueBucketSize: 0 } + ) + ).toThrow( + new RangeError( + 'Cannot construct a worker node with a tasks queue bucket size option that is not a positive integer' + ) + ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { tasksQueueBackPressureSize: 12, tasksQueueBucketSize: -1 } + ) + ).toThrow( + new RangeError( + 'Cannot construct a worker node with a tasks queue bucket size option that is not a positive integer' + ) + ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { + tasksQueueBackPressureSize: 12, + tasksQueueBucketSize: 6, + } + ) + ).toThrow( + new RangeError( + 'Cannot construct a worker node without a tasks queue priority option' + ) + ) + expect( + () => + new WorkerNode( + WorkerTypes.thread, + './tests/worker-files/thread/testWorker.mjs', + { + tasksQueueBackPressureSize: 12, + tasksQueueBucketSize: 6, + tasksQueuePriority: 'invalidTasksQueuePriority', + } + ) + ).toThrow( + new RangeError( + 'Cannot construct a worker node with a tasks queue priority option that is not a boolean' + ) + ) expect(threadWorkerNode).toBeInstanceOf(WorkerNode) expect(threadWorkerNode.worker).toBeInstanceOf(ThreadWorker) expect(threadWorkerNode.info).toStrictEqual({ @@ -127,7 +230,8 @@ describe('Worker node test suite', () => { type: WorkerTypes.thread, dynamic: false, ready: false, - stealing: false + stealing: false, + backPressure: false, }) expect(threadWorkerNode.usage).toStrictEqual({ tasks: { @@ -137,31 +241,33 @@ describe('Worker node test suite', () => { maxQueued: 0, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, waitTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, elu: { idle: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, active: { - history: new CircularArray() - } - } + history: expect.any(CircularBuffer), + }, + }, }) expect(threadWorkerNode.messageChannel).toBeInstanceOf(MessageChannel) expect(threadWorkerNode.tasksQueueBackPressureSize).toBe(12) expect(threadWorkerNode.tasksQueue).toBeInstanceOf(PriorityQueue) expect(threadWorkerNode.tasksQueue.size).toBe(0) + expect(threadWorkerNode.tasksQueue.bucketSize).toBe(6) + expect(threadWorkerNode.tasksQueue.enablePriority).toBe(true) expect(threadWorkerNode.tasksQueueSize()).toBe( threadWorkerNode.tasksQueue.size ) - expect(threadWorkerNode.onBackPressureStarted).toBe(false) + expect(threadWorkerNode.setBackPressureFlag).toBe(false) expect(threadWorkerNode.taskFunctionsUsage).toBeInstanceOf(Map) expect(clusterWorkerNode).toBeInstanceOf(WorkerNode) @@ -171,7 +277,8 @@ describe('Worker node test suite', () => { type: WorkerTypes.cluster, dynamic: false, ready: false, - stealing: false + stealing: false, + backPressure: false, }) expect(clusterWorkerNode.usage).toStrictEqual({ tasks: { @@ -181,31 +288,33 @@ describe('Worker node test suite', () => { maxQueued: 0, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, waitTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, elu: { idle: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, active: { - history: new CircularArray() - } - } + history: expect.any(CircularBuffer), + }, + }, }) expect(clusterWorkerNode.messageChannel).toBeUndefined() expect(clusterWorkerNode.tasksQueueBackPressureSize).toBe(12) expect(clusterWorkerNode.tasksQueue).toBeInstanceOf(PriorityQueue) expect(clusterWorkerNode.tasksQueue.size).toBe(0) + expect(clusterWorkerNode.tasksQueue.bucketSize).toBe(6) + expect(clusterWorkerNode.tasksQueue.enablePriority).toBe(true) expect(clusterWorkerNode.tasksQueueSize()).toBe( clusterWorkerNode.tasksQueue.size ) - expect(clusterWorkerNode.onBackPressureStarted).toBe(false) + expect(clusterWorkerNode.setBackPressureFlag).toBe(false) expect(clusterWorkerNode.taskFunctionsUsage).toBeInstanceOf(Map) }) @@ -219,7 +328,7 @@ describe('Worker node test suite', () => { ) threadWorkerNode.info.taskFunctionsProperties = [ { name: DEFAULT_TASK_NAME }, - { name: 'fn1' } + { name: 'fn1' }, ] expect(() => threadWorkerNode.getTaskFunctionWorkerUsage('invalidTaskFunction') @@ -231,7 +340,7 @@ describe('Worker node test suite', () => { threadWorkerNode.info.taskFunctionsProperties = [ { name: DEFAULT_TASK_NAME }, { name: 'fn1' }, - { name: 'fn2' } + { name: 'fn2' }, ] expect( threadWorkerNode.getTaskFunctionWorkerUsage(DEFAULT_TASK_NAME) @@ -242,22 +351,22 @@ describe('Worker node test suite', () => { queued: 0, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, waitTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, elu: { idle: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, active: { - history: new CircularArray() - } - } + history: expect.any(CircularBuffer), + }, + }, }) expect(threadWorkerNode.getTaskFunctionWorkerUsage('fn1')).toStrictEqual({ tasks: { @@ -266,22 +375,22 @@ describe('Worker node test suite', () => { queued: 0, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, waitTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, elu: { idle: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, active: { - history: new CircularArray() - } - } + history: expect.any(CircularBuffer), + }, + }, }) expect(threadWorkerNode.getTaskFunctionWorkerUsage('fn2')).toStrictEqual({ tasks: { @@ -290,22 +399,22 @@ describe('Worker node test suite', () => { queued: 0, sequentiallyStolen: 0, stolen: 0, - failed: 0 + failed: 0, }, runTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, waitTime: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, elu: { idle: { - history: new CircularArray() + history: expect.any(CircularBuffer), }, active: { - history: new CircularArray() - } - } + history: expect.any(CircularBuffer), + }, + }, }) expect(threadWorkerNode.taskFunctionsUsage.size).toBe(2) }) @@ -314,7 +423,7 @@ describe('Worker node test suite', () => { expect(threadWorkerNode.info.taskFunctionsProperties).toStrictEqual([ { name: DEFAULT_TASK_NAME }, { name: 'fn1' }, - { name: 'fn2' } + { name: 'fn2' }, ]) expect(threadWorkerNode.taskFunctionsUsage.size).toBe(2) expect(