Merge branch 'master' into combined-prs-branch

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

diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts
index 5d28802c12769c1cbaf0cd519e22708918788016..f7a48d3ba43e06cadf9a5da8c855a0b5ce1547bb 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>
 
   /**
@@ -69,11 +70,19 @@ 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.tasksQueue = new PriorityQueue<Task<Data>>(
+      opts.tasksQueueBucketSize,
+      opts.tasksQueuePriority
+    )
+    this.setBackPressureFlag = false
     this.taskFunctionsUsage = new Map<string, WorkerUsage>()
   }
 
+  /** @inheritdoc */
+  public setTasksQueuePriority (enablePriority: boolean): void {
+    this.tasksQueue.enablePriority = enablePriority
+  }
+
   /** @inheritdoc */
   public tasksQueueSize (): number {
     return this.tasksQueue.size
@@ -82,23 +91,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 +221,8 @@ export class WorkerNode<Worker extends IWorker, Data = unknown>
       type: getWorkerType(worker)!,
       dynamic: false,
       ready: false,
-      stealing: false
+      stealing: false,
+      backPressure: false
     }
   }
 
@@ -223,17 +248,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 +291,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)
         }
       }
     }