refactor: code cleanups

Signed-off-by: Jérôme Benoit <[email protected]>

Author: jerome-benoit

src/circular-buffer.ts

@@ -4,7 +4,7 @@
 export const defaultBufferSize = 2048
 
 /**
- * Circular buffer designed for positive numbers.
+ * Circular buffer designed for numbers.
  *
  * @internal
  */
@@ -25,7 +25,7 @@ export class CircularBuffer {
     this.writeIdx = 0
     this.maxArrayIdx = size - 1
     this.size = 0
-    this.items = new Float32Array(size).fill(-1)
+    this.items = new Float32Array(size)
   }
 
   /**
@@ -71,7 +71,6 @@ export class CircularBuffer {
       return
     }
     const number = this.items[this.readIdx]
-    this.items[this.readIdx] = -1
     this.readIdx = this.readIdx === this.maxArrayIdx ? 0 : this.readIdx + 1
     --this.size
     return number

src/pools/abstract-pool.ts

@@ -312,7 +312,7 @@ export abstract class AbstractPool<
       ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
             .runTime.aggregate === true &&
         this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
-          .waitTime.aggregate &&
+            .waitTime.aggregate === true &&
         {
           utilization: round(this.utilization),
         }),

src/pools/selection-strategies/interleaved-weighted-round-robin-worker-choice-strategy.ts

@@ -142,16 +142,12 @@ export class InterleavedWeightedRoundRobinWorkerChoiceStrategy<
       return true
     }
     if (
-      this.workerNodeId === workerNodeKey &&
-      this.workerNodeId > this.pool.workerNodes.length - 1
+      this.nextWorkerNodeKey != null &&
+      this.nextWorkerNodeKey >= workerNodeKey
     ) {
-      this.workerNodeId = this.pool.workerNodes.length - 1
-    }
-    if (
-      this.previousWorkerNodeKey === workerNodeKey &&
-      this.previousWorkerNodeKey > this.pool.workerNodes.length - 1
-    ) {
-      this.previousWorkerNodeKey = this.pool.workerNodes.length - 1
+      this.nextWorkerNodeKey =
+        (this.nextWorkerNodeKey - 1 + this.pool.workerNodes.length) %
+        this.pool.workerNodes.length
     }
     return true
   }

src/pools/worker-node.ts

@@ -42,7 +42,6 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
   public readonly tasksQueue: PriorityQueue<Task<Data>>
   /** @inheritdoc */
   public tasksQueueBackPressureSize: number
-  private setBackPressureFlag: boolean
   private readonly taskFunctionsUsage: Map<string, WorkerUsage>
 
   /**
@@ -84,7 +83,6 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
       opts.tasksQueueBucketSize,
       opts.tasksQueuePriority,
     )
-    this.setBackPressureFlag = false
     this.taskFunctionsUsage = new Map<string, WorkerUsage>()
   }
 
@@ -101,34 +99,22 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
   /** @inheritdoc */
   public enqueueTask(task: Task<Data>): number {
     const tasksQueueSize = this.tasksQueue.enqueue(task, task.priority)
-    if (
-      !this.setBackPressureFlag &&
-      this.hasBackPressure() &&
-      !this.info.backPressure
-    ) {
-      this.setBackPressureFlag = true
+    if (this.hasBackPressure() && !this.info.backPressure) {
       this.info.backPressure = true
       this.dispatchEvent(
         new CustomEvent<WorkerNodeEventDetail>('backPressure', {
           detail: { workerId: this.info.id },
         }),
       )
-      this.setBackPressureFlag = false
     }
     return tasksQueueSize
   }
 
   /** @inheritdoc */
   public dequeueTask(bucket?: number): Task<Data> | undefined {
     const task = this.tasksQueue.dequeue(bucket)
-    if (
-      !this.setBackPressureFlag &&
-      !this.hasBackPressure() &&
-      this.info.backPressure
-    ) {
-      this.setBackPressureFlag = true
+    if (!this.hasBackPressure() && this.info.backPressure) {
       this.info.backPressure = false
-      this.setBackPressureFlag = false
     }
     return task
   }

src/queues/abstract-fixed-queue.ts

@@ -72,23 +72,19 @@ export abstract class AbstractFixedQueue<T> implements IFixedQueue<T> {
       }
     }
     if (logicalIndex !== -1) {
-      let toShiftIndex = this.start + logicalIndex
-      if (toShiftIndex >= this.capacity) {
-        toShiftIndex -= this.capacity
+      let physicalShiftIndex = this.start + logicalIndex
+      if (physicalShiftIndex >= this.capacity) {
+        physicalShiftIndex -= this.capacity
       }
       for (let i = logicalIndex; i < this.size - 1; i++) {
-        let nextIndex = toShiftIndex + 1
-        if (nextIndex === this.capacity) {
-          nextIndex = 0
+        let nextPhysicalIndex = physicalShiftIndex + 1
+        if (nextPhysicalIndex === this.capacity) {
+          nextPhysicalIndex = 0
         }
-        this.nodeArray[toShiftIndex] = this.nodeArray[nextIndex]
-        toShiftIndex = nextIndex
+        this.nodeArray[physicalShiftIndex] = this.nodeArray[nextPhysicalIndex]
+        physicalShiftIndex = nextPhysicalIndex
       }
-      let end = this.start + this.size - 1
-      if (end >= this.capacity) {
-        end -= this.capacity
-      }
-      this.nodeArray[end] = undefined
+      this.nodeArray[physicalShiftIndex] = undefined
       --this.size
       return true
     }
@@ -101,11 +97,11 @@ export abstract class AbstractFixedQueue<T> implements IFixedQueue<T> {
       return undefined
     }
     const index = this.start
-    --this.size
     ++this.start
     if (this.start === this.capacity) {
       this.start = 0
     }
+    --this.size
     return this.nodeArray[index]!.data
   }
 

src/queues/priority-queue.ts

@@ -146,20 +146,19 @@ export class PriorityQueue<T> {
     let prev: PriorityQueueNode<T> | undefined
     if (bucket != null && bucket > 0) {
       let currentBucket = 1
-      let searchNode: PriorityQueueNode<T> | undefined = this.tail
-      while (searchNode != null) {
-        if (currentBucket === bucket) {
-          targetNode = searchNode
-          break
-        }
+      while (targetNode?.next != null && currentBucket < bucket) {
+        prev = targetNode
+        targetNode = targetNode.next
         ++currentBucket
-        prev = searchNode
-        searchNode = searchNode.next
       }
-    }
-    while (targetNode?.empty() === true && targetNode !== this.head) {
-      prev = targetNode
-      targetNode = targetNode.next
+      if (currentBucket < bucket || targetNode?.empty() === true) {
+        return undefined
+      }
+    } else {
+      while (targetNode?.empty() === true && targetNode !== this.head) {
+        prev = targetNode
+        targetNode = targetNode.next
+      }
     }
     if (targetNode?.empty() === true) {
       return undefined
@@ -212,11 +211,11 @@ export class PriorityQueue<T> {
 
       const value = node.get(index)
       if (value == null) {
-        index++
+        ++index
         return getNextValue()
       }
 
-      index++
+      ++index
       return { done: false, value }
     }
     return {

src/worker/utils.ts

@@ -22,11 +22,8 @@ export const checkValidWorkerOptions = (
   }
   if (
     opts?.maxInactiveTime != null &&
-    !Number.isSafeInteger(opts.maxInactiveTime)
+    (!Number.isSafeInteger(opts.maxInactiveTime) || opts.maxInactiveTime < 5)
   ) {
-    throw new TypeError('maxInactiveTime option is not an integer')
-  }
-  if (opts?.maxInactiveTime != null && opts.maxInactiveTime < 5) {
     throw new TypeError(
       'maxInactiveTime option is not a positive integer greater or equal than 5',
     )
@@ -43,14 +40,7 @@ export const checkValidTaskFunctionObjectEntry = <
   name: string,
   fnObj: TaskFunctionObject<Data, Response>,
 ): void => {
-  if (typeof name !== 'string') {
-    throw new TypeError('A taskFunctions parameter object key is not a string')
-  }
-  if (typeof name === 'string' && name.trim().length === 0) {
-    throw new TypeError(
-      'A taskFunctions parameter object key is an empty string',
-    )
-  }
+  checkTaskFunctionName(name)
   if (typeof fnObj.taskFunction !== 'function') {
     throw new TypeError(
       `taskFunction object 'taskFunction' property '${fnObj.taskFunction}' is not a function`,
@@ -64,7 +54,7 @@ export const checkTaskFunctionName = (name: string): void => {
   if (typeof name !== 'string') {
     throw new TypeError('name parameter is not a string')
   }
-  if (typeof name === 'string' && name.trim().length === 0) {
+  if (name.trim().length === 0) {
     throw new TypeError('name parameter is an empty string')
   }
 }

tests/circular-buffer.test.mjs

@@ -41,19 +41,17 @@ describe('Circular buffer test suite', () => {
   it('Verify that circular buffer put() works as intended', () => {
     const circularBuffer = new CircularBuffer(4)
     circularBuffer.put(1)
-    expect(circularBuffer.items).toStrictEqual(
-      new Float32Array([1, -1, -1, -1]),
-    )
+    expect(circularBuffer.items).toStrictEqual(new Float32Array([1, 0, 0, 0]))
     expect(circularBuffer.readIdx).toBe(0)
     expect(circularBuffer.writeIdx).toBe(1)
     expect(circularBuffer.size).toBe(1)
     circularBuffer.put(2)
-    expect(circularBuffer.items).toStrictEqual(new Float32Array([1, 2, -1, -1]))
+    expect(circularBuffer.items).toStrictEqual(new Float32Array([1, 2, 0, 0]))
     expect(circularBuffer.readIdx).toBe(0)
     expect(circularBuffer.writeIdx).toBe(2)
     expect(circularBuffer.size).toBe(2)
     circularBuffer.put(3)
-    expect(circularBuffer.items).toStrictEqual(new Float32Array([1, 2, 3, -1]))
+    expect(circularBuffer.items).toStrictEqual(new Float32Array([1, 2, 3, 0]))
     expect(circularBuffer.readIdx).toBe(0)
     expect(circularBuffer.writeIdx).toBe(3)
     expect(circularBuffer.size).toBe(3)
@@ -155,10 +153,15 @@ describe('Circular buffer test suite', () => {
   it('Verify that circular buffer toArray() works as intended', () => {
     const circularBuffer = new CircularBuffer(4)
     circularBuffer.put(1)
+    expect(circularBuffer.toArray()).toStrictEqual([1])
     circularBuffer.put(2)
+    expect(circularBuffer.toArray()).toStrictEqual([1, 2])
     circularBuffer.put(3)
+    expect(circularBuffer.toArray()).toStrictEqual([1, 2, 3])
     circularBuffer.put(4)
+    expect(circularBuffer.toArray()).toStrictEqual([1, 2, 3, 4])
     circularBuffer.put(5)
+    expect(circularBuffer.toArray()).toStrictEqual([2, 3, 4, 5])
     circularBuffer.put(6)
     expect(circularBuffer.toArray()).toStrictEqual([3, 4, 5, 6])
   })

tests/pools/worker-node.test.mjs

@@ -286,7 +286,6 @@ describe('Worker node test suite', () => {
     expect(threadWorkerNode.tasksQueueSize()).toBe(
       threadWorkerNode.tasksQueue.size,
     )
-    expect(threadWorkerNode.setBackPressureFlag).toBe(false)
     expect(threadWorkerNode.taskFunctionsUsage).toBeInstanceOf(Map)
   })
 

tests/worker/abstract-worker.test.mjs

@@ -29,10 +29,14 @@ describe('Abstract worker test suite', () => {
       new TypeError("killBehavior option '0' is not valid"),
     )
     expect(() => new ThreadWorker(() => {}, { maxInactiveTime: '' })).toThrow(
-      new TypeError('maxInactiveTime option is not an integer'),
+      new TypeError(
+        'maxInactiveTime option is not a positive integer greater or equal than 5',
+      ),
     )
     expect(() => new ThreadWorker(() => {}, { maxInactiveTime: 0.5 })).toThrow(
-      new TypeError('maxInactiveTime option is not an integer'),
+      new TypeError(
+        'maxInactiveTime option is not a positive integer greater or equal than 5',
+      ),
     )
     expect(() => new ThreadWorker(() => {}, { maxInactiveTime: 0 })).toThrow(
       new TypeError(
@@ -143,7 +147,7 @@ describe('Abstract worker test suite', () => {
     }
     const fn2 = ''
     expect(() => new ThreadWorker({ '': fn1 })).toThrow(
-      new TypeError('A taskFunctions parameter object key is an empty string'),
+      new TypeError('name parameter is an empty string'),
     )
     expect(() => new ThreadWorker({ fn1, fn2 })).toThrow(
       new TypeError(