From 662c6c101161b8f263e304b461087082edb1c003 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Wed, 20 Aug 2025 20:29:19 +0200 Subject: [PATCH] perf: reduce pool info property internal usage 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 | 55 +++++++++++++++++++++++------------- src/pools/cluster/dynamic.ts | 9 ++++-- src/pools/thread/dynamic.ts | 9 ++++-- 3 files changed, 47 insertions(+), 26 deletions(-) diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 65ad5e3a4..332656a00 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -154,18 +154,8 @@ export abstract class AbstractPool< accumulator + (workerNode.usage.tasks.maxQueued ?? 0), 0 ), - queuedTasks: this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + workerNode.usage.tasks.queued, - 0 - ), - stealingWorkerNodes: this.workerNodes.reduce( - (accumulator, _, workerNodeKey) => - this.isWorkerNodeStealing(workerNodeKey) - ? accumulator + 1 - : accumulator, - 0 - ), + queuedTasks: this.getQueuedTasks(), + stealingWorkerNodes: this.getStealingWorkerNodes(), stolenTasks: this.workerNodes.reduce( (accumulator, workerNode) => accumulator + workerNode.usage.tasks.stolen, @@ -632,7 +622,8 @@ export abstract class AbstractPool< }) ) if (this.emitter != null) { - this.emitter.emit(PoolEvents.destroy, this.info) + this.emitter.listenerCount(PoolEvents.destroy) > 0 && + this.emitter.emit(PoolEvents.destroy, this.info) this.emitter.emitDestroy() this.readyEventEmitted = false } @@ -1425,13 +1416,14 @@ export abstract class AbstractPool< !this.started || this.destroying || this.workerNodes.length <= 1 || - this.info.queuedTasks === 0 + this.getQueuedTasks() === 0 ) } private checkAndEmitReadyEvent (): void { if (this.emitter != null && !this.readyEventEmitted && this.ready) { - this.emitter.emit(PoolEvents.ready, this.info) + this.emitter.listenerCount(PoolEvents.ready) > 0 && + this.emitter.emit(PoolEvents.ready, this.info) this.readyEventEmitted = true } } @@ -1442,21 +1434,24 @@ export abstract class AbstractPool< this.backPressureEventEmitted && !this.backPressure ) { - this.emitter.emit(PoolEvents.backPressureEnd, this.info) + this.emitter.listenerCount(PoolEvents.backPressureEnd) > 0 && + this.emitter.emit(PoolEvents.backPressureEnd, this.info) this.backPressureEventEmitted = false } } private checkAndEmitTaskExecutionEvents (): void { if (this.emitter != null && !this.busyEventEmitted && this.busy) { - this.emitter.emit(PoolEvents.busy, this.info) + this.emitter.listenerCount(PoolEvents.busy) > 0 && + this.emitter.emit(PoolEvents.busy, this.info) this.busyEventEmitted = true } } private checkAndEmitTaskExecutionFinishedEvents (): void { if (this.emitter != null && this.busyEventEmitted && !this.busy) { - this.emitter.emit(PoolEvents.busyEnd, this.info) + this.emitter.listenerCount(PoolEvents.busyEnd) > 0 && + this.emitter.emit(PoolEvents.busyEnd, this.info) this.busyEventEmitted = false } } @@ -1467,7 +1462,8 @@ export abstract class AbstractPool< !this.backPressureEventEmitted && this.backPressure ) { - this.emitter.emit(PoolEvents.backPressure, this.info) + this.emitter.listenerCount(PoolEvents.backPressure) > 0 && + this.emitter.emit(PoolEvents.backPressure, this.info) this.backPressureEventEmitted = true } } @@ -1670,6 +1666,22 @@ export abstract class AbstractPool< : new Error(`Task '${taskName}' id '${taskId}' aborted`) } + private getQueuedTasks (): number { + return this.workerNodes.reduce((accumulator, workerNode) => { + return accumulator + workerNode.usage.tasks.queued + }, 0) + } + + private getStealingWorkerNodes (): number { + return this.workerNodes.reduce( + (accumulator, _, workerNodeKey) => + this.isWorkerNodeStealing(workerNodeKey) + ? accumulator + 1 + : accumulator, + 0 + ) + } + /** * Gets task function worker choice strategy, if any. * @param name - The task function name. @@ -2061,7 +2073,7 @@ export abstract class AbstractPool< private readonly isStealingRatioReached = (): boolean => { return ( this.opts.tasksQueueOptions?.tasksStealingRatio === 0 || - (this.info.stealingWorkerNodes ?? 0) > + this.getStealingWorkerNodes() > Math.ceil( this.workerNodes.length * // eslint-disable-next-line @typescript-eslint/no-non-null-assertion @@ -2308,6 +2320,9 @@ export abstract class AbstractPool< message: MessageValue ): Promise { const targetWorkerNodeKeys = [...this.workerNodes.keys()] + if (targetWorkerNodeKeys.length === 0) { + return true + } const responsesReceived: MessageValue[] = [] const taskFunctionOperationsListener = ( message: MessageValue, diff --git a/src/pools/cluster/dynamic.ts b/src/pools/cluster/dynamic.ts index 1d2900677..48d83e4d2 100644 --- a/src/pools/cluster/dynamic.ts +++ b/src/pools/cluster/dynamic.ts @@ -89,7 +89,8 @@ export class DynamicClusterPool< protected override checkAndEmitDynamicWorkerCreationEvents (): void { if (this.emitter != null) { if (!this.fullEventEmitted && this.full) { - this.emitter.emit(PoolEvents.full, this.info) + this.emitter.listenerCount(PoolEvents.full) > 0 && + this.emitter.emit(PoolEvents.full, this.info) this.fullEventEmitted = true } if (this.emptyEventEmitted && !this.empty) { @@ -102,11 +103,13 @@ export class DynamicClusterPool< protected override checkAndEmitDynamicWorkerDestructionEvents (): void { if (this.emitter != null) { if (this.fullEventEmitted && !this.full) { - this.emitter.emit(PoolEvents.fullEnd, this.info) + this.emitter.listenerCount(PoolEvents.fullEnd) > 0 && + this.emitter.emit(PoolEvents.fullEnd, this.info) this.fullEventEmitted = false } if (!this.emptyEventEmitted && this.empty) { - this.emitter.emit(PoolEvents.empty, this.info) + this.emitter.listenerCount(PoolEvents.empty) > 0 && + this.emitter.emit(PoolEvents.empty, this.info) this.emptyEventEmitted = true } } diff --git a/src/pools/thread/dynamic.ts b/src/pools/thread/dynamic.ts index 32beca8e6..17d759ea9 100644 --- a/src/pools/thread/dynamic.ts +++ b/src/pools/thread/dynamic.ts @@ -89,7 +89,8 @@ export class DynamicThreadPool< protected override checkAndEmitDynamicWorkerCreationEvents (): void { if (this.emitter != null) { if (!this.fullEventEmitted && this.full) { - this.emitter.emit(PoolEvents.full, this.info) + this.emitter.listenerCount(PoolEvents.full) > 0 && + this.emitter.emit(PoolEvents.full, this.info) this.fullEventEmitted = true } if (this.emptyEventEmitted && !this.empty) { @@ -102,11 +103,13 @@ export class DynamicThreadPool< protected override checkAndEmitDynamicWorkerDestructionEvents (): void { if (this.emitter != null) { if (this.fullEventEmitted && !this.full) { - this.emitter.emit(PoolEvents.fullEnd, this.info) + this.emitter.listenerCount(PoolEvents.fullEnd) > 0 && + this.emitter.emit(PoolEvents.fullEnd, this.info) this.fullEventEmitted = false } if (!this.emptyEventEmitted && this.empty) { - this.emitter.emit(PoolEvents.empty, this.info) + this.emitter.listenerCount(PoolEvents.empty) > 0 && + this.emitter.emit(PoolEvents.empty, this.info) this.emptyEventEmitted = true } } -- 2.53.0