X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fpools%2Fworker-node.ts;h=f3ec3be3d83f4541da3713cbd46406d19be41ac0;hb=6e9c39d3eedabc9f12808490983d36e551a75a28;hp=e28372efa1b3c73cf9c8904d6077f87fed1dd553;hpb=68cbdc846878bc058323b757a68b4c83eedc6388;p=poolifier.git

diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts
index e28372ef..f3ec3be3 100644
--- a/src/pools/worker-node.ts
+++ b/src/pools/worker-node.ts
@@ -17,6 +17,9 @@ import {
   type WorkerUsage
 } from './worker'
 
+type EmptyQueueCallback = (workerId: number) => void
+type BackPressureCallback = EmptyQueueCallback
+
 /**
  * Worker node.
  *
@@ -30,18 +33,18 @@ implements IWorkerNode<Worker, Data> {
   /** @inheritdoc */
   public readonly info: WorkerInfo
   /** @inheritdoc */
-  public messageChannel?: MessageChannel
-  /** @inheritdoc */
   public usage: WorkerUsage
   /** @inheritdoc */
+  public messageChannel?: MessageChannel
+  /** @inheritdoc */
   public tasksQueueBackPressureSize: number
   /** @inheritdoc */
-  public onBackPressure?: (workerId: number) => void
+  public onBackPressure?: BackPressureCallback
   /** @inheritdoc */
-  public onEmptyQueue?: (workerId: number) => void
-  private readonly taskFunctionsUsage: Map<string, WorkerUsage>
+  public onEmptyQueue?: EmptyQueueCallback
   private readonly tasksQueue: Deque<Task<Data>>
   private onEmptyQueueCount: number
+  private readonly taskFunctionsUsage: Map<string, WorkerUsage>
 
   /**
    * Constructs a new worker node.
@@ -75,14 +78,14 @@ implements IWorkerNode<Worker, Data> {
     }
     this.worker = worker
     this.info = this.initWorkerInfo(worker, workerType)
+    this.usage = this.initWorkerUsage()
     if (workerType === WorkerTypes.thread) {
       this.messageChannel = new MessageChannel()
     }
-    this.usage = this.initWorkerUsage()
-    this.taskFunctionsUsage = new Map<string, WorkerUsage>()
-    this.tasksQueue = new Deque<Task<Data>>()
     this.tasksQueueBackPressureSize = tasksQueueBackPressureSize
+    this.tasksQueue = new Deque<Task<Data>>()
     this.onEmptyQueueCount = 0
+    this.taskFunctionsUsage = new Map<string, WorkerUsage>()
   }
 
   /** @inheritdoc */
@@ -178,16 +181,17 @@ implements IWorkerNode<Worker, Data> {
   }
 
   private async startOnEmptyQueue (): Promise<void> {
-    if (this.onEmptyQueue != null) {
-      if (this.tasksQueue.size > 0) {
-        this.onEmptyQueueCount = 0
-        return
-      }
-      this.onEmptyQueue(this.info.id as number)
-      ++this.onEmptyQueueCount
-      await sleep(exponentialDelay(this.onEmptyQueueCount))
-      await this.startOnEmptyQueue()
+    if (
+      this.onEmptyQueueCount > 0 &&
+      (this.usage.tasks.executing > 0 || this.tasksQueue.size > 0)
+    ) {
+      this.onEmptyQueueCount = 0
+      return
     }
+    (this.onEmptyQueue as EmptyQueueCallback)(this.info.id as number)
+    ++this.onEmptyQueueCount
+    await sleep(exponentialDelay(this.onEmptyQueueCount))
+    await this.startOnEmptyQueue()
   }
 
   private initWorkerInfo (worker: Worker, workerType: WorkerType): WorkerInfo {