feat: add ELU tasks accounting

Reference: #768

Signed-off-by: Jérôme Benoit <jerome.benoit@sap.com>

diff --git a/.eslintrc.js b/.eslintrc.js
index d7004cc55ff8460161f54f766086dd78d7612eb8..4d8c7ffd0ddd57278e18605e916fb20076accb3a 100644 (file)
--- a/.eslintrc.js
+++ b/.eslintrc.js
@@ -51,6 +51,7 @@ module.exports = defineConfig({
           'dequeue',
           'dequeued',
           'ecma',
+          'elu',
           'enqueue',
           'enum',
           'errored',
@@ -63,6 +64,7 @@ module.exports = defineConfig({
           'mjs',
           'num',
           'os',
+          'perf',
           'piscina',
           'pnpm',
           'poolifier',

diff --git a/rollup.config.mjs b/rollup.config.mjs
index 7454ef722b2fa32c8987187ec7fb12b5faeb0602..b5c0b285b77b8daa31ab63e4cd19f5152e688825 100644 (file)
--- a/rollup.config.mjs
+++ b/rollup.config.mjs
@@ -49,6 +49,7 @@ export default {
     'node:crypto',
     'node:events',
     'node:os',
+    'node:perf_hooks',
     'node:worker_threads'
   ],
   plugins: [

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 5088f3d90957bd276756c120d2496e8ffe01bb5a..065b228ada73e6d21f6fb8238bddf403cc04ac83 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -1,4 +1,5 @@
 import crypto from 'node:crypto'
+import { performance } from 'node:perf_hooks'
 import type { MessageValue, PromiseResponseWrapper } from '../utility-types'
 import {
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
@@ -299,7 +300,8 @@ export abstract class AbstractPool<
         waitTimeHistory: new CircularArray(),
         avgWaitTime: 0,
         medWaitTime: 0,
-        error: 0
+        error: 0,
+        elu: undefined
       })
     }
     this.workerChoiceStrategyContext.setWorkerChoiceStrategy(
@@ -474,6 +476,7 @@ export abstract class AbstractPool<
     }
     this.updateRunTimeTasksUsage(workerTasksUsage, message)
     this.updateWaitTimeTasksUsage(workerTasksUsage, message)
+    this.updateEluTasksUsage(workerTasksUsage, message)
   }
 
   private updateRunTimeTasksUsage (
@@ -522,6 +525,25 @@ export abstract class AbstractPool<
     }
   }
 
+  private updateEluTasksUsage (
+    workerTasksUsage: TasksUsage,
+    message: MessageValue<Response>
+  ): void {
+    if (this.workerChoiceStrategyContext.getRequiredStatistics().elu) {
+      if (workerTasksUsage.elu != null && message.elu != null) {
+        // TODO: cumulative or delta?
+        workerTasksUsage.elu = {
+          idle: workerTasksUsage.elu.idle + message.elu.idle,
+          active: workerTasksUsage.elu.active + message.elu.active,
+          utilization:
+            workerTasksUsage.elu.utilization + message.elu.utilization
+        }
+      } else if (message.elu != null) {
+        workerTasksUsage.elu = message.elu
+      }
+    }
+  }
+
   /**
    * Chooses a worker node for the next task.
    *
@@ -704,7 +726,8 @@ export abstract class AbstractPool<
         waitTimeHistory: new CircularArray(),
         avgWaitTime: 0,
         medWaitTime: 0,
-        error: 0
+        error: 0,
+        elu: undefined
       },
       tasksQueue: new Queue<Task<Data>>()
     })

diff --git a/src/pools/selection-strategies/abstract-worker-choice-strategy.ts b/src/pools/selection-strategies/abstract-worker-choice-strategy.ts
index 6e798fbc4349d98bf5b5761d884013382efa601f..9b14f3c58430dd143d8f84cf79b7ee635428eafc 100644 (file)
--- a/src/pools/selection-strategies/abstract-worker-choice-strategy.ts
+++ b/src/pools/selection-strategies/abstract-worker-choice-strategy.ts
@@ -31,7 +31,8 @@ export abstract class AbstractWorkerChoiceStrategy<
     medRunTime: false,
     waitTime: false,
     avgWaitTime: false,
-    medWaitTime: false
+    medWaitTime: false,
+    elu: false
   }
 
   /**

diff --git a/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts b/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts
index 56d9dc361fe6e70255877ec6c1e6be6f33b70a70..a0e7ee1a35d2620b956eb9e240192754434097f8 100644 (file)
--- a/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts
+++ b/src/pools/selection-strategies/fair-share-worker-choice-strategy.ts
@@ -30,7 +30,8 @@ export class FairShareWorkerChoiceStrategy<
     medRunTime: false,
     waitTime: false,
     avgWaitTime: false,
-    medWaitTime: false
+    medWaitTime: false,
+    elu: false
   }
 
   /**

diff --git a/src/pools/selection-strategies/least-busy-worker-choice-strategy.ts b/src/pools/selection-strategies/least-busy-worker-choice-strategy.ts
index 6e413825d666025022acf6018dfe47c77e8345fe..2dfb6083d4d85e6a4fe9b58d5a2eddca07e55819 100644 (file)
--- a/src/pools/selection-strategies/least-busy-worker-choice-strategy.ts
+++ b/src/pools/selection-strategies/least-busy-worker-choice-strategy.ts
@@ -29,7 +29,8 @@ export class LeastBusyWorkerChoiceStrategy<
     medRunTime: false,
     waitTime: false,
     avgWaitTime: false,
-    medWaitTime: false
+    medWaitTime: false,
+    elu: false
   }
 
   /** @inheritDoc */

diff --git a/src/pools/selection-strategies/selection-strategies-types.ts b/src/pools/selection-strategies/selection-strategies-types.ts
index 18ba40b410d5190271a82d248a12b5e2179485de..c8497c276f07d06adc81dcb76e7255747203cb0e 100644 (file)
--- a/src/pools/selection-strategies/selection-strategies-types.ts
+++ b/src/pools/selection-strategies/selection-strategies-types.ts
@@ -90,6 +90,10 @@ export interface RequiredStatistics {
    * Require tasks median wait time.
    */
   medWaitTime: boolean
+  /**
+   * Event loop utilization.
+   */
+  elu: boolean
 }
 
 /**

diff --git a/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts b/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts
index 17ae66478606990c44244f44c46a9d8e5f8fb943..1ea091c6f712be7be7472534692581fd3b56de0f 100644 (file)
--- a/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts
+++ b/src/pools/selection-strategies/weighted-round-robin-worker-choice-strategy.ts
@@ -30,7 +30,8 @@ export class WeightedRoundRobinWorkerChoiceStrategy<
     medRunTime: false,
     waitTime: false,
     avgWaitTime: false,
-    medWaitTime: false
+    medWaitTime: false,
+    elu: false
   }
 
   /**

diff --git a/src/pools/worker.ts b/src/pools/worker.ts
index ec75c819cfe9be8a5e56dac0e0c2bdad4cfe7d66..d10a27c9171c978ce44e7b9f6a1d2a9fb66131db 100644 (file)
--- a/src/pools/worker.ts
+++ b/src/pools/worker.ts
@@ -1,3 +1,4 @@
+import type { EventLoopUtilization } from 'node:perf_hooks'
 import type { CircularArray } from '../circular-array'
 import type { Queue } from '../queue'
 
@@ -105,6 +106,10 @@ export interface TasksUsage {
    * Number of tasks errored.
    */
   error: number
+  /**
+   * Event loop utilization.
+   */
+  elu: EventLoopUtilization | undefined
 }
 
 /**

diff --git a/src/utility-types.ts b/src/utility-types.ts
index 21a442538d4988eba011f7904b2467cc895c3593..cc29801f9afc81e23d25ce6f9e91e49ed60ae831 100644 (file)
--- a/src/utility-types.ts
+++ b/src/utility-types.ts
@@ -1,5 +1,6 @@
 import type { Worker as ClusterWorker } from 'node:cluster'
 import type { MessagePort } from 'node:worker_threads'
+import type { EventLoopUtilization } from 'node:perf_hooks'
 import type { KillBehavior } from './worker/worker-options'
 import type { IWorker, Task } from './pools/worker'
 
@@ -41,6 +42,10 @@ export interface MessageValue<
    * Wait time.
    */
   readonly waitTime?: number
+  /**
+   * Event loop utilization.
+   */
+  readonly elu?: EventLoopUtilization
   /**
    * Reference to main worker.
    */

diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts
index 91a45e94b0f36c568326460891332d61d7de0731..a484dce12927c1cd29721e9529aa08a822c37b43 100644 (file)
--- a/src/worker/abstract-worker.ts
+++ b/src/worker/abstract-worker.ts
@@ -1,6 +1,7 @@
 import { AsyncResource } from 'node:async_hooks'
 import type { Worker } from 'node:cluster'
 import type { MessagePort } from 'node:worker_threads'
+import { type EventLoopUtilization, performance } from 'node:perf_hooks'
 import type {
   MessageValue,
   TaskFunctions,
@@ -19,6 +20,13 @@ const DEFAULT_FUNCTION_NAME = 'default'
 const DEFAULT_MAX_INACTIVE_TIME = 60000
 const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
 
+interface TaskPerformance {
+  timestamp: number
+  waitTime: number
+  runTime?: number
+  elu: EventLoopUtilization
+}
+
 /**
  * Base class that implements some shared logic for all poolifier workers.
  *
@@ -209,14 +217,14 @@ export abstract class AbstractWorker<
     message: MessageValue<Data>
   ): void {
     try {
-      const startTimestamp = performance.now()
-      const waitTime = startTimestamp - (message.submissionTimestamp ?? 0)
+      const taskPerformance = this.beforeTaskRunHook(message)
       const res = fn(message.data)
-      const runTime = performance.now() - startTimestamp
+      const { runTime, waitTime, elu } = this.afterTaskRunHook(taskPerformance)
       this.sendToMainWorker({
         data: res,
         runTime,
         waitTime,
+        elu,
         id: message.id
       })
     } catch (e) {
@@ -241,15 +249,16 @@ export abstract class AbstractWorker<
     fn: WorkerAsyncFunction<Data, Response>,
     message: MessageValue<Data>
   ): void {
-    const startTimestamp = performance.now()
-    const waitTime = startTimestamp - (message.submissionTimestamp ?? 0)
+    const taskPerformance = this.beforeTaskRunHook(message)
     fn(message.data)
       .then(res => {
-        const runTime = performance.now() - startTimestamp
+        const { runTime, waitTime, elu } =
+          this.afterTaskRunHook(taskPerformance)
         this.sendToMainWorker({
           data: res,
           runTime,
           waitTime,
+          elu,
           id: message.id
         })
         return null
@@ -281,4 +290,24 @@ export abstract class AbstractWorker<
     }
     return fn
   }
+
+  private beforeTaskRunHook (message: MessageValue<Data>): TaskPerformance {
+    // TODO: conditional accounting
+    const timestamp = performance.now()
+    return {
+      timestamp,
+      waitTime: timestamp - (message.submissionTimestamp ?? 0),
+      elu: performance.eventLoopUtilization()
+    }
+  }
+
+  private afterTaskRunHook (taskPerformance: TaskPerformance): TaskPerformance {
+    return {
+      ...taskPerformance,
+      ...{
+        runTime: performance.now() - taskPerformance.timestamp,
+        elu: performance.eventLoopUtilization(taskPerformance.elu)
+      }
+    }
+  }
 }