fix: fix potential ressource leaks

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index ee3b632bb7f3ffbc46ebcfba8454402885e896ff..3f146f8ae98cf1f3d625fbe762454b4f6166b295 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -2159,29 +2159,37 @@ export abstract class AbstractPool<
   }
 
   private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
-    await new Promise<void>((resolve, reject) => {
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-      if (this.workerNodes[workerNodeKey] == null) {
-        resolve()
-        return
-      }
-      const killMessageListener = (message: MessageValue<Response>): void => {
-        this.checkMessageWorkerId(message)
-        if (message.kill === 'success') {
+    let killMessageListener:
+      | ((message: MessageValue<Response>) => void)
+      | undefined
+    try {
+      await new Promise<void>((resolve, reject) => {
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        if (this.workerNodes[workerNodeKey] == null) {
           resolve()
-        } else if (message.kill === 'failure') {
-          reject(
-            new Error(
-              // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
-              `Kill message handling failed on worker ${message.workerId?.toString()}`
+          return
+        }
+        killMessageListener = (message: MessageValue<Response>): void => {
+          this.checkMessageWorkerId(message)
+          if (message.kill === 'success') {
+            resolve()
+          } else if (message.kill === 'failure') {
+            reject(
+              new Error(
+                // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
+                `Kill message handling failed on worker ${message.workerId?.toString()}`
+              )
             )
-          )
+          }
         }
+        this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
+        this.sendToWorker(workerNodeKey, { kill: true })
+      })
+    } finally {
+      if (killMessageListener != null) {
+        this.deregisterWorkerMessageListener(workerNodeKey, killMessageListener)
       }
-      // FIXME: should be registered only once
-      this.registerWorkerMessageListener(workerNodeKey, killMessageListener)
-      this.sendToWorker(workerNodeKey, { kill: true })
-    })
+    }
   }
 
   /**
@@ -2205,19 +2213,29 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     message: MessageValue<Data>
   ): Promise<boolean> {
-    return await new Promise<boolean>((resolve, reject) => {
-      const taskFunctionOperationListener = (
-        message: MessageValue<Response>
-      ): void => {
-        this.checkMessageWorkerId(message)
-        const workerId = this.getWorkerInfo(workerNodeKey)?.id
-        if (
-          message.taskFunctionOperationStatus != null &&
-          message.workerId === workerId
-        ) {
-          if (message.taskFunctionOperationStatus) {
-            resolve(true)
-          } else {
+    let taskFunctionOperationListener:
+      | ((message: MessageValue<Response>) => void)
+      | undefined
+    try {
+      return await new Promise<boolean>((resolve, reject) => {
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        if (this.workerNodes[workerNodeKey] == null) {
+          resolve(true)
+          return
+        }
+        taskFunctionOperationListener = (
+          message: MessageValue<Response>
+        ): void => {
+          this.checkMessageWorkerId(message)
+          const workerId = this.getWorkerInfo(workerNodeKey)?.id
+          if (
+            message.taskFunctionOperationStatus != null &&
+            message.workerId === workerId
+          ) {
+            if (message.taskFunctionOperationStatus) {
+              resolve(true)
+              return
+            }
             reject(
               new Error(
                 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
@@ -2228,23 +2246,27 @@ export abstract class AbstractPool<
               )
             )
           }
-          this.deregisterWorkerMessageListener(
-            this.getWorkerNodeKeyByWorkerId(message.workerId),
-            taskFunctionOperationListener
-          )
         }
+        this.registerWorkerMessageListener(
+          workerNodeKey,
+          taskFunctionOperationListener
+        )
+        this.sendToWorker(workerNodeKey, message)
+      })
+    } finally {
+      if (taskFunctionOperationListener != null) {
+        this.deregisterWorkerMessageListener(
+          workerNodeKey,
+          taskFunctionOperationListener
+        )
       }
-      this.registerWorkerMessageListener(
-        workerNodeKey,
-        taskFunctionOperationListener
-      )
-      this.sendToWorker(workerNodeKey, message)
-    })
+    }
   }
 
   private async sendTaskFunctionOperationToWorkers (
     message: MessageValue<Data>
   ): Promise<boolean> {
+    const targetWorkerNodeKeys = [...this.workerNodes.keys()]
     const taskFunctionOperationsListener = (
       message: MessageValue<Response>,
       resolve: (value: boolean | PromiseLike<boolean>) => void,
@@ -2252,9 +2274,14 @@ export abstract class AbstractPool<
       responsesReceived: MessageValue<Response>[]
     ): void => {
       this.checkMessageWorkerId(message)
-      if (message.taskFunctionOperationStatus != null) {
+      if (
+        message.taskFunctionOperationStatus != null &&
+        targetWorkerNodeKeys.includes(
+          this.getWorkerNodeKeyByWorkerId(message.workerId)
+        )
+      ) {
         responsesReceived.push(message)
-        if (responsesReceived.length >= this.workerNodes.length) {
+        if (responsesReceived.length >= targetWorkerNodeKeys.length) {
           if (
             responsesReceived.every(
               msg => msg.taskFunctionOperationStatus === true
@@ -2275,9 +2302,13 @@ export abstract class AbstractPool<
         }
       }
     }
-    let listener: (message: MessageValue<Response>) => void
+    let listener: ((message: MessageValue<Response>) => void) | undefined
     try {
       return await new Promise<boolean>((resolve, reject) => {
+        if (targetWorkerNodeKeys.length === 0) {
+          resolve(true)
+          return
+        }
         const responsesReceived: MessageValue<Response>[] = []
         listener = (message: MessageValue<Response>) => {
           taskFunctionOperationsListener(
@@ -2287,15 +2318,16 @@ export abstract class AbstractPool<
             responsesReceived
           )
         }
-        for (const workerNodeKey of this.workerNodes.keys()) {
+        for (const workerNodeKey of targetWorkerNodeKeys) {
           this.registerWorkerMessageListener(workerNodeKey, listener)
           this.sendToWorker(workerNodeKey, message)
         }
       })
     } finally {
-      for (const workerNodeKey of this.workerNodes.keys()) {
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        this.deregisterWorkerMessageListener(workerNodeKey, listener!)
+      if (listener != null) {
+        for (const workerNodeKey of targetWorkerNodeKeys) {
+          this.deregisterWorkerMessageListener(workerNodeKey, listener)
+        }
       }
     }
   }

diff --git a/src/pools/cluster/fixed.ts b/src/pools/cluster/fixed.ts
index 0023aeb8d184c56aa6d7d1bfb83a35dfff1e97c1..d704772660446162fdc973eefa559aa5138a6c16 100644 (file)
--- a/src/pools/cluster/fixed.ts
+++ b/src/pools/cluster/fixed.ts
@@ -73,7 +73,7 @@ export class FixedClusterPool<
     workerNodeKey: number,
     listener: (message: MessageValue<Message>) => void
   ): void {
-    this.workerNodes[workerNodeKey].worker.off('message', listener)
+    this.workerNodes[workerNodeKey]?.worker.off('message', listener)
   }
 
   /** @inheritDoc */

diff --git a/src/pools/thread/fixed.ts b/src/pools/thread/fixed.ts
index 773ab847402f767462e05a3d9001df89b1e8b6ba..5d1ffbabab28c65079abe5c17ec4baf6f3c063e5 100644 (file)
--- a/src/pools/thread/fixed.ts
+++ b/src/pools/thread/fixed.ts
@@ -77,7 +77,7 @@ export class FixedThreadPool<
     workerNodeKey: number,
     listener: (message: MessageValue<Message>) => void
   ): void {
-    this.workerNodes[workerNodeKey].messageChannel?.port1.off(
+    this.workerNodes[workerNodeKey]?.messageChannel?.port1.off(
       'message',
       listener
     )