From 2d1d3b2d566ff0b9af3e678c3b0f8de36f3dbbc1 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Thu, 9 May 2024 15:00:49 +0200 Subject: [PATCH] fix: avoid tasks redistribution on the errored worker node 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 | 102 +++++++++++++++---------------------- 1 file changed, 40 insertions(+), 62 deletions(-) diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 0de96a41..73f4e31a 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1619,14 +1619,15 @@ export abstract class AbstractPool< } } - private redistributeQueuedTasks (workerNodeKey: number): void { - if (workerNodeKey === -1 || this.cannotStealTask()) { + private redistributeQueuedTasks (sourceWorkerNodeKey: number): void { + if (sourceWorkerNodeKey === -1 || this.cannotStealTask()) { return } - while (this.tasksQueueSize(workerNodeKey) > 0) { + while (this.tasksQueueSize(sourceWorkerNodeKey) > 0) { const destinationWorkerNodeKey = this.workerNodes.reduce( (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => { - return workerNode.info.ready && + return sourceWorkerNodeKey !== workerNodeKey && + workerNode.info.ready && workerNode.usage.tasks.queued < workerNodes[minWorkerNodeKey].usage.tasks.queued ? workerNodeKey @@ -1637,7 +1638,7 @@ export abstract class AbstractPool< this.handleTask( destinationWorkerNodeKey, // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.dequeueTask(workerNodeKey)! + this.dequeueTask(sourceWorkerNodeKey)! ) } } @@ -1663,46 +1664,42 @@ export abstract class AbstractPool< } private updateTaskSequentiallyStolenStatisticsWorkerUsage ( - workerNodeKey: number + workerNodeKey: number, + taskName: string, + previousStolenTaskName: string ): void { const workerNode = this.workerNodes[workerNodeKey] // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - if (workerNode?.usage != null) { + if (workerNode.usage != null) { ++workerNode.usage.tasks.sequentiallyStolen } - } - - private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage ( - workerNodeKey: number, - taskName: string - ): void { - const workerNode = this.workerNodes[workerNodeKey] + const taskFunctionWorkerUsage = + workerNode.getTaskFunctionWorkerUsage(taskName) if ( this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && - workerNode.getTaskFunctionWorkerUsage(taskName) != null + taskFunctionWorkerUsage != null && + (taskFunctionWorkerUsage.tasks.sequentiallyStolen === 0 || + (previousStolenTaskName === taskName && + taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0)) ) { - const taskFunctionWorkerUsage = - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - workerNode.getTaskFunctionWorkerUsage(taskName)! ++taskFunctionWorkerUsage.tasks.sequentiallyStolen + } else if ( + this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + taskFunctionWorkerUsage != null + ) { + taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0 } } private resetTaskSequentiallyStolenStatisticsWorkerUsage ( - workerNodeKey: number + workerNodeKey: number, + taskName: string ): void { const workerNode = this.workerNodes[workerNodeKey] // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition if (workerNode?.usage != null) { workerNode.usage.tasks.sequentiallyStolen = 0 } - } - - private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage ( - workerNodeKey: number, - taskName: string - ): void { - const workerNode = this.workerNodes[workerNodeKey] if ( this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && workerNode.getTaskFunctionWorkerUsage(taskName) != null @@ -1732,6 +1729,11 @@ export abstract class AbstractPool< ) { if (workerInfo != null && previousStolenTask != null) { workerInfo.stealing = false + this.resetTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + previousStolenTask.name! + ) } return } @@ -1739,19 +1741,15 @@ export abstract class AbstractPool< if ( workerInfo != null && previousStolenTask != null && - workerNodeTasksUsage.sequentiallyStolen > 0 && (workerNodeTasksUsage.executing > 0 || this.tasksQueueSize(workerNodeKey) > 0) ) { workerInfo.stealing = false - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - for (const taskFunctionProperties of workerInfo.taskFunctionsProperties!) { - this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( - workerNodeKey, - taskFunctionProperties.name - ) - } - this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey) + this.resetTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + previousStolenTask.name! + ) return } if (workerInfo == null) { @@ -1761,33 +1759,14 @@ export abstract class AbstractPool< } workerInfo.stealing = true const stolenTask = this.workerNodeStealTask(workerNodeKey) - if ( - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && - stolenTask != null - ) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const taskFunctionTasksWorkerUsage = this.workerNodes[ - workerNodeKey + if (stolenTask != null && previousStolenTask != null) { + this.updateTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - ].getTaskFunctionWorkerUsage(stolenTask.name!)!.tasks - if ( - taskFunctionTasksWorkerUsage.sequentiallyStolen === 0 || - (previousStolenTask != null && - previousStolenTask.name === stolenTask.name && - taskFunctionTasksWorkerUsage.sequentiallyStolen > 0) - ) { - this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( - workerNodeKey, - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - stolenTask.name! - ) - } else { - this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( - workerNodeKey, - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - stolenTask.name! - ) - } + stolenTask.name!, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + previousStolenTask.name! + ) } sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen)) .then(() => { @@ -1819,7 +1798,6 @@ export abstract class AbstractPool< // eslint-disable-next-line @typescript-eslint/no-non-null-assertion const task = sourceWorkerNode.dequeueLastPrioritizedTask()! this.handleTask(workerNodeKey, task) - this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey) // eslint-disable-next-line @typescript-eslint/no-non-null-assertion this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!) return task -- 2.34.1