From 3386e26ca681aa33757903602d6e23c56401a8e4 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Mon, 7 Jul 2025 15:29:31 +0200 Subject: [PATCH] fix: fix potential ressource leaks MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit Signed-off-by: Jérôme Benoit --- src/pools/abstract-pool.ts | 130 +++++++++++++++++++++++-------------- src/pools/cluster/fixed.ts | 2 +- src/pools/thread/fixed.ts | 2 +- 3 files changed, 83 insertions(+), 51 deletions(-) diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index ee3b632bb..3f146f8ae 100644 --- 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 { - await new Promise((resolve, reject) => { - // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (this.workerNodes[workerNodeKey] == null) { - resolve() - return - } - const killMessageListener = (message: MessageValue): void => { - this.checkMessageWorkerId(message) - if (message.kill === 'success') { + let killMessageListener: + | ((message: MessageValue) => void) + | undefined + try { + await new Promise((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): 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 ): Promise { - return await new Promise((resolve, reject) => { - const taskFunctionOperationListener = ( - message: MessageValue - ): 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) => void) + | undefined + try { + return await new Promise((resolve, reject) => { + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (this.workerNodes[workerNodeKey] == null) { + resolve(true) + return + } + taskFunctionOperationListener = ( + message: MessageValue + ): 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 ): Promise { + const targetWorkerNodeKeys = [...this.workerNodes.keys()] const taskFunctionOperationsListener = ( message: MessageValue, resolve: (value: boolean | PromiseLike) => void, @@ -2252,9 +2274,14 @@ export abstract class AbstractPool< responsesReceived: MessageValue[] ): 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) => void + let listener: ((message: MessageValue) => void) | undefined try { return await new Promise((resolve, reject) => { + if (targetWorkerNodeKeys.length === 0) { + resolve(true) + return + } const responsesReceived: MessageValue[] = [] listener = (message: MessageValue) => { 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 0023aeb8d..d70477266 100644 --- a/src/pools/cluster/fixed.ts +++ b/src/pools/cluster/fixed.ts @@ -73,7 +73,7 @@ export class FixedClusterPool< workerNodeKey: number, listener: (message: MessageValue) => 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 773ab8474..5d1ffbaba 100644 --- a/src/pools/thread/fixed.ts +++ b/src/pools/thread/fixed.ts @@ -77,7 +77,7 @@ export class FixedThreadPool< workerNodeKey: number, listener: (message: MessageValue) => void ): void { - this.workerNodes[workerNodeKey].messageChannel?.port1.off( + this.workerNodes[workerNodeKey]?.messageChannel?.port1.off( 'message', listener ) -- 2.53.0