fix: fix continuous tasks stealing on idle start at worker node idling

[poolifier.git] / src / pools / abstract-pool.ts

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 8c2dcc31d2e849c0fdbf4e0ed9d8bfa297c585de..ab319c627884c25c4851dc34ecb2626ebc528fc3 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -12,12 +12,14 @@ import {
   DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS,
   EMPTY_FUNCTION,
   average,
+  exponentialDelay,
   isKillBehavior,
   isPlainObject,
   max,
   median,
   min,
-  round
+  round,
+  sleep
 } from '../utils'
 import { KillBehaviors } from '../worker/worker-options'
 import type { TaskFunction } from '../worker/task-functions'
@@ -1478,11 +1480,79 @@ export abstract class AbstractPool<
     }
   }
 
+  private updateTaskSequentiallyStolenStatisticsWorkerUsage (
+    workerNodeKey: number,
+    taskName: string
+  ): void {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode?.usage != null) {
+      ++workerNode.usage.tasks.sequentiallyStolen
+    }
+    if (
+      this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+      workerNode.getTaskFunctionWorkerUsage(taskName) != null
+    ) {
+      const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage(
+        taskName
+      ) as WorkerUsage
+      ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
+    }
+  }
+
+  private resetTaskSequentiallyStolenStatisticsWorkerUsage (
+    workerNodeKey: number,
+    taskName: string
+  ): void {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode?.usage != null) {
+      workerNode.usage.tasks.sequentiallyStolen = 0
+    }
+    if (
+      this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
+      workerNode.getTaskFunctionWorkerUsage(taskName) != null
+    ) {
+      const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage(
+        taskName
+      ) as WorkerUsage
+      taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
+    }
+  }
+
   private readonly handleIdleWorkerNodeEvent = (
-    event: CustomEvent<WorkerNodeEventDetail>
+    event: CustomEvent<WorkerNodeEventDetail>,
+    previousStolenTask?: Task<Data>
   ): void => {
-    const { workerId } = event.detail
-    const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId)
+    const { workerNodeKey } = event.detail
+    if (workerNodeKey == null) {
+      throw new Error(
+        'WorkerNode event detail workerNodeKey attribute must be defined'
+      )
+    }
+    const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
+    if (
+      previousStolenTask != null &&
+      workerNodeTasksUsage.sequentiallyStolen > 0 &&
+      (workerNodeTasksUsage.executing > 0 ||
+        this.tasksQueueSize(workerNodeKey) > 0)
+    ) {
+      this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
+        workerNodeKey,
+        previousStolenTask.name as string
+      )
+      return
+    }
+    const stolenTask = this.workerNodeStealTask(workerNodeKey)
+    sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen))
+      .then(() => {
+        this.handleIdleWorkerNodeEvent(event, stolenTask)
+        return undefined
+      })
+      .catch(EMPTY_FUNCTION)
+  }
+
+  private readonly workerNodeStealTask = (
+    workerNodeKey: number
+  ): Task<Data> | undefined => {
     const workerNodes = this.workerNodes
       .slice()
       .sort(
@@ -1490,22 +1560,27 @@ export abstract class AbstractPool<
           workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued
       )
     const sourceWorkerNode = workerNodes.find(
-      workerNode =>
-        workerNode.info.ready &&
-        workerNode.info.id !== workerId &&
-        workerNode.usage.tasks.queued > 0
+      (sourceWorkerNode, sourceWorkerNodeKey) =>
+        sourceWorkerNode.info.ready &&
+        sourceWorkerNodeKey !== workerNodeKey &&
+        sourceWorkerNode.usage.tasks.queued > 0
     )
     if (sourceWorkerNode != null) {
       const task = sourceWorkerNode.popTask() as Task<Data>
-      if (this.shallExecuteTask(destinationWorkerNodeKey)) {
-        this.executeTask(destinationWorkerNodeKey, task)
+      if (this.shallExecuteTask(workerNodeKey)) {
+        this.executeTask(workerNodeKey, task)
       } else {
-        this.enqueueTask(destinationWorkerNodeKey, task)
+        this.enqueueTask(workerNodeKey, task)
       }
+      this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
+        workerNodeKey,
+        task.name as string
+      )
       this.updateTaskStolenStatisticsWorkerUsage(
-        destinationWorkerNodeKey,
+        workerNodeKey,
         task.name as string
       )
+      return task
     }
   }
 
@@ -1567,16 +1642,15 @@ export abstract class AbstractPool<
   }
 
   private handleWorkerReadyResponse (message: MessageValue<Response>): void {
-    if (message.ready === false) {
-      throw new Error(
-        `Worker ${message.workerId as number} failed to initialize`
-      )
+    const { workerId, ready, taskFunctionNames } = message
+    if (ready === false) {
+      throw new Error(`Worker ${workerId as number} failed to initialize`)
     }
     const workerInfo = this.getWorkerInfo(
-      this.getWorkerNodeKeyByWorkerId(message.workerId)
+      this.getWorkerNodeKeyByWorkerId(workerId)
     )
-    workerInfo.ready = message.ready as boolean
-    workerInfo.taskFunctionNames = message.taskFunctionNames
+    workerInfo.ready = ready as boolean
+    workerInfo.taskFunctionNames = taskFunctionNames
     if (!this.readyEventEmitted && this.ready) {
       this.readyEventEmitted = true
       this.emitter?.emit(PoolEvents.ready, this.info)
@@ -1584,7 +1658,7 @@ export abstract class AbstractPool<
   }
 
   private handleTaskExecutionResponse (message: MessageValue<Response>): void {
-    const { taskId, workerError, data } = message
+    const { workerId, taskId, workerError, data } = message
     const promiseResponse = this.promiseResponseMap.get(taskId as string)
     if (promiseResponse != null) {
       const { resolve, reject, workerNodeKey } = promiseResponse
@@ -1597,16 +1671,29 @@ export abstract class AbstractPool<
       this.afterTaskExecutionHook(workerNodeKey, message)
       this.workerChoiceStrategyContext.update(workerNodeKey)
       this.promiseResponseMap.delete(taskId as string)
-      if (
-        this.opts.enableTasksQueue === true &&
-        this.tasksQueueSize(workerNodeKey) > 0 &&
-        this.workerNodes[workerNodeKey].usage.tasks.executing <
-          (this.opts.tasksQueueOptions?.concurrency as number)
-      ) {
-        this.executeTask(
-          workerNodeKey,
-          this.dequeueTask(workerNodeKey) as Task<Data>
-        )
+      if (this.opts.enableTasksQueue === true) {
+        const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
+        if (
+          this.tasksQueueSize(workerNodeKey) > 0 &&
+          workerNodeTasksUsage.executing <
+            (this.opts.tasksQueueOptions?.concurrency as number)
+        ) {
+          this.executeTask(
+            workerNodeKey,
+            this.dequeueTask(workerNodeKey) as Task<Data>
+          )
+        }
+        if (
+          workerNodeTasksUsage.executing === 0 &&
+          this.tasksQueueSize(workerNodeKey) === 0 &&
+          workerNodeTasksUsage.sequentiallyStolen === 0
+        ) {
+          this.workerNodes[workerNodeKey].dispatchEvent(
+            new CustomEvent<WorkerNodeEventDetail>('idleWorkerNode', {
+              detail: { workerId: workerId as number, workerNodeKey }
+            })
+          )
+        }
       }
     }
   }