fix: fix async task statistics accounting

[poolifier.git] / src / worker / abstract-worker.ts

diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts
index 4e92be01ae340da26d4699595ff692566075f006..23739270b3a9863c20ea1a212004b2a94e982335 100644 (file)
--- a/src/worker/abstract-worker.ts
+++ b/src/worker/abstract-worker.ts
@@ -7,7 +7,12 @@ import type {
   TaskPerformance,
   WorkerStatistics
 } from '../utility-types'
-import { EMPTY_FUNCTION, isAsyncFunction, isPlainObject } from '../utils'
+import {
+  DEFAULT_TASK_NAME,
+  EMPTY_FUNCTION,
+  isAsyncFunction,
+  isPlainObject
+} from '../utils'
 import {
   type KillBehavior,
   KillBehaviors,
@@ -20,7 +25,6 @@ import type {
   WorkerSyncFunction
 } from './worker-functions'
 
-const DEFAULT_FUNCTION_NAME = 'default'
 const DEFAULT_MAX_INACTIVE_TIME = 60000
 const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
 
@@ -114,7 +118,7 @@ export abstract class AbstractWorker<
     }
     this.taskFunctions = new Map<string, WorkerFunction<Data, Response>>()
     if (typeof taskFunctions === 'function') {
-      this.taskFunctions.set(DEFAULT_FUNCTION_NAME, taskFunctions.bind(this))
+      this.taskFunctions.set(DEFAULT_TASK_NAME, taskFunctions.bind(this))
     } else if (isPlainObject(taskFunctions)) {
       let firstEntry = true
       for (const [name, fn] of Object.entries(taskFunctions)) {
@@ -125,7 +129,7 @@ export abstract class AbstractWorker<
         }
         this.taskFunctions.set(name, fn.bind(this))
         if (firstEntry) {
-          this.taskFunctions.set(DEFAULT_FUNCTION_NAME, fn.bind(this))
+          this.taskFunctions.set(DEFAULT_TASK_NAME, fn.bind(this))
           firstEntry = false
         }
       }
@@ -145,27 +149,29 @@ export abstract class AbstractWorker<
    * @param message - Message received.
    */
   protected messageListener (message: MessageValue<Data, Data>): void {
-    if (message.ready != null && message.workerId === this.id) {
-      // Startup message received
-      this.workerReady()
-    } else if (message.statistics != null) {
-      // Statistics message received
-      this.statistics = message.statistics
-    } else if (message.checkAlive != null) {
-      // Check alive message received
-      message.checkAlive ? this.startCheckAlive() : this.stopCheckAlive()
-    } else if (message.id != null && message.data != null) {
-      // Task message received
-      const fn = this.getTaskFunction(message.name)
-      if (isAsyncFunction(fn)) {
-        this.runInAsyncScope(this.runAsync.bind(this), this, fn, message)
-      } else {
-        this.runInAsyncScope(this.runSync.bind(this), this, fn, message)
+    if (message.workerId === this.id) {
+      if (message.ready != null) {
+        // Startup message received
+        this.workerReady()
+      } else if (message.statistics != null) {
+        // Statistics message received
+        this.statistics = message.statistics
+      } else if (message.checkAlive != null) {
+        // Check alive message received
+        message.checkAlive ? this.startCheckAlive() : this.stopCheckAlive()
+      } else if (message.id != null && message.data != null) {
+        // Task message received
+        const fn = this.getTaskFunction(message.name)
+        if (isAsyncFunction(fn)) {
+          this.runInAsyncScope(this.runAsync.bind(this), this, fn, message)
+        } else {
+          this.runInAsyncScope(this.runSync.bind(this), this, fn, message)
+        }
+      } else if (message.kill === true) {
+        // Kill message received
+        this.stopCheckAlive()
+        this.emitDestroy()
       }
-    } else if (message.kill === true) {
-      // Kill message received
-      this.stopCheckAlive()
-      this.emitDestroy()
     }
   }
 
@@ -203,7 +209,7 @@ export abstract class AbstractWorker<
       performance.now() - this.lastTaskTimestamp >
       (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
     ) {
-      this.sendToMainWorker({ kill: this.opts.killBehavior })
+      this.sendToMainWorker({ kill: this.opts.killBehavior, workerId: this.id })
     }
   }
 
@@ -249,7 +255,7 @@ export abstract class AbstractWorker<
     message: MessageValue<Data>
   ): void {
     try {
-      let taskPerformance = this.beginTaskPerformance()
+      let taskPerformance = this.beginTaskPerformance(message.name)
       const res = fn(message.data)
       taskPerformance = this.endTaskPerformance(taskPerformance)
       this.sendToMainWorker({
@@ -262,10 +268,11 @@ export abstract class AbstractWorker<
       const errorMessage = this.handleError(e as Error | string)
       this.sendToMainWorker({
         taskError: {
-          workerId: this.id,
+          name: message.name ?? DEFAULT_TASK_NAME,
           message: errorMessage,
           data: message.data
         },
+        workerId: this.id,
         id: message.id
       })
     } finally {
@@ -285,7 +292,7 @@ export abstract class AbstractWorker<
     fn: WorkerAsyncFunction<Data, Response>,
     message: MessageValue<Data>
   ): void {
-    let taskPerformance = this.beginTaskPerformance()
+    let taskPerformance = this.beginTaskPerformance(message.name)
     fn(message.data)
       .then(res => {
         taskPerformance = this.endTaskPerformance(taskPerformance)
@@ -301,10 +308,11 @@ export abstract class AbstractWorker<
         const errorMessage = this.handleError(e as Error | string)
         this.sendToMainWorker({
           taskError: {
-            workerId: this.id,
+            name: message.name ?? DEFAULT_TASK_NAME,
             message: errorMessage,
             data: message.data
           },
+          workerId: this.id,
           id: message.id
         })
       })
@@ -319,10 +327,10 @@ export abstract class AbstractWorker<
   /**
    * Gets the task function in the given scope.
    *
-   * @param name - Name of the function that will be returned.
+   * @param name - Name of the task function that will be returned.
    */
   private getTaskFunction (name?: string): WorkerFunction<Data, Response> {
-    name = name ?? DEFAULT_FUNCTION_NAME
+    name = name ?? DEFAULT_TASK_NAME
     const fn = this.taskFunctions.get(name)
     if (fn == null) {
       throw new Error(`Task function '${name}' not found`)
@@ -330,9 +338,10 @@ export abstract class AbstractWorker<
     return fn
   }
 
-  private beginTaskPerformance (): TaskPerformance {
+  private beginTaskPerformance (name?: string): TaskPerformance {
     this.checkStatistics()
     return {
+      name: name ?? DEFAULT_TASK_NAME,
       timestamp: performance.now(),
       ...(this.statistics.elu && { elu: performance.eventLoopUtilization() })
     }