perf: optimize circular buffer for numbers

[poolifier.git] / src / pools / worker-node.ts

diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts
index 5d28802c12769c1cbaf0cd519e22708918788016..079557010565cc7a7e129d0f314caf4762eb7700 100644 (file)
--- a/src/pools/worker-node.ts
+++ b/src/pools/worker-node.ts
@@ -1,7 +1,7 @@
 import { EventEmitter } from 'node:events'
 import { MessageChannel } from 'node:worker_threads'
 
-import { CircularArray } from '../circular-array.js'
+import { CircularBuffer } from '../circular-buffer.js'
 import { PriorityQueue } from '../priority-queue.js'
 import type { Task } from '../utility-types.js'
 import { DEFAULT_TASK_NAME } from '../utils.js'
@@ -15,6 +15,7 @@ import {
   type EventHandler,
   type IWorker,
   type IWorkerNode,
+  MeasurementHistorySize,
   type StrategyData,
   type WorkerInfo,
   type WorkerNodeOptions,
@@ -45,7 +46,7 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
   /** @inheritdoc */
   public tasksQueueBackPressureSize: number
   private readonly tasksQueue: PriorityQueue<Task<Data>>
-  private onBackPressureStarted: boolean
+  private setBackPressureFlag: boolean
   private readonly taskFunctionsUsage: Map<string, WorkerUsage>
 
   /**
@@ -70,7 +71,7 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     this.tasksQueueBackPressureSize = opts.tasksQueueBackPressureSize!
     this.tasksQueue = new PriorityQueue<Task<Data>>(opts.tasksQueueBucketSize)
-    this.onBackPressureStarted = false
+    this.setBackPressureFlag = false
     this.taskFunctionsUsage = new Map<string, WorkerUsage>()
   }
 
@@ -82,23 +83,38 @@ 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.hasBackPressure() && !this.onBackPressureStarted) {
-      this.onBackPressureStarted = true
+    if (
+      !this.setBackPressureFlag &&
+      this.hasBackPressure() &&
+      !this.info.backPressure
+    ) {
+      this.setBackPressureFlag = true
+      this.info.backPressure = true
       this.emit('backPressure', { workerId: this.info.id })
-      this.onBackPressureStarted = false
+      this.setBackPressureFlag = false
     }
     return tasksQueueSize
   }
 
   /** @inheritdoc */
   public dequeueTask (bucket?: number): Task<Data> | undefined {
-    return this.tasksQueue.dequeue(bucket)
+    const task = this.tasksQueue.dequeue(bucket)
+    if (
+      !this.setBackPressureFlag &&
+      !this.hasBackPressure() &&
+      this.info.backPressure
+    ) {
+      this.setBackPressureFlag = true
+      this.info.backPressure = false
+      this.setBackPressureFlag = false
+    }
+    return task
   }
 
   /** @inheritdoc */
-  public dequeueLastBucketTask (): Task<Data> | undefined {
+  public dequeueLastPrioritizedTask (): Task<Data> | undefined {
     // Start from the last empty or partially filled bucket
-    return this.tasksQueue.dequeue(this.tasksQueue.buckets + 1)
+    return this.dequeueTask(this.tasksQueue.buckets + 1)
   }
 
   /** @inheritdoc */
@@ -197,7 +213,8 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
       type: getWorkerType(worker)!,
       dynamic: false,
       ready: false,
-      stealing: false
+      stealing: false,
+      backPressure: false
     }
   }
 
@@ -223,17 +240,17 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
         failed: 0
       },
       runTime: {
-        history: new CircularArray<number>()
+        history: new CircularBuffer(MeasurementHistorySize)
       },
       waitTime: {
-        history: new CircularArray<number>()
+        history: new CircularBuffer(MeasurementHistorySize)
       },
       elu: {
         idle: {
-          history: new CircularArray<number>()
+          history: new CircularBuffer(MeasurementHistorySize)
         },
         active: {
-          history: new CircularArray<number>()
+          history: new CircularBuffer(MeasurementHistorySize)
         }
       }
     }
@@ -266,17 +283,17 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
         failed: 0
       },
       runTime: {
-        history: new CircularArray<number>()
+        history: new CircularBuffer(MeasurementHistorySize)
       },
       waitTime: {
-        history: new CircularArray<number>()
+        history: new CircularBuffer(MeasurementHistorySize)
       },
       elu: {
         idle: {
-          history: new CircularArray<number>()
+          history: new CircularBuffer(MeasurementHistorySize)
         },
         active: {
-          history: new CircularArray<number>()
+          history: new CircularBuffer(MeasurementHistorySize)
         }
       }
     }