You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "turboFei (via GitHub)" <gi...@apache.org> on 2025/03/11 04:46:42 UTC
[PR] cancel running tasks [celeborn]
turboFei opened a new pull request, #3144:
URL: https://github.com/apache/celeborn/pull/3144
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
### Why are the changes needed?
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related task did cause the FetchFailedException eventually(another task running or finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related task did trigger the FetchFailedException eventually(another task running or finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2728268022
ping @RexXiong @FMX
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related tasks did cause the FetchFailedException eventually(another task running or finished).
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related tasks did cause the FetchFailedException eventually.
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
Review Comment:
even we do not know whether the `shuffleFetchFailureTaskIds` related tasks did cause the FetchFailedException eventually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2734982690
> https://github.com/apache/spark/blob/0670e4f7945ae4935261ed1f45db7ede79aca127/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1928-L1936
https://github.com/apache/spark/commit/05b85eb8cb1bec2017b0cd1202b011a185ba6875
In the celeborn stage rerun case, seems the new rerun stage would also launch all the tasks. not sure how much resource can be saved.
how do you think about? @FMX
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2748959890
need to check:
```
val isCelebornShuffleIndeterminate = stage.isInstanceOf[ShuffleMapStage] && CelebornShuffleState.isCelebornSkewedShuffle(stage.asInstanceOf[ShuffleMapStage].shuffleDep.shuffleId)
stage.isIndeterminate || isCelebornShuffleIndeterminate
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related task did trigger the FetchFailedException eventually(depends on whether another task attempt is running or has been finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe to cancel the running tasks after checking the `taskSetManager.runningTasks > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988411692
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
Review Comment:
even we do not know whether the `shuffleFetchFailureTaskIds` related tasks did cause the FetchFailedException eventually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r2000415124
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkUtils.java:
##########
@@ -344,6 +344,20 @@ public static void cancelShuffle(int shuffleId, String reason) {
.hiddenImpl(TaskSetManager.class, "taskInfos")
.defaultAlwaysNull()
.build();
+ private static final DynFields.UnboundField<
+ scala.collection.mutable.HashMap<
+ Integer, scala.collection.mutable.HashMap<Integer, TaskSetManager>>>
+ TASK_SETS_BY_STAGE_ID_AND_ATTEMPT_FIELD =
+ DynFields.builder()
+ .hiddenImpl(TaskSchedulerImpl.class, "taskSetsByStageIdAndAttempt")
Review Comment:
```
// TaskSetManagers are not thread safe, so any access to one should be synchronized
// on this class. Protected by `this`
private val taskSetsByStageIdAndAttempt = new HashMap[Int, HashMap[Int, TaskSetManager]]
```
It looks like this variable is not thread-safe, will this be safe if you get the variable by reflection?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related task did trigger the FetchFailedException eventually(depends on whether another task attempt is running or has been finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe after checking the `taskSetManager.runningTasks > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related tasks did cause the FetchFailedException eventually(another task running or finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei closed pull request #3144: [WIP][CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource
URL: https://github.com/apache/celeborn/pull/3144
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2748887387
@Z1Wu You are correct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2746539744
> > For celeborn shuffle fetch failure, it will rerun the whole stage, so it is fine to cancel all the running tasks.
>
> ```shell
> shuffleMapStage -> FailedStage
> ```
>
> In current implementation, when fetch failure happens without indeterministic shuffle map output, Spark will only rerun the whole shuffleMapStage and tasks that are not finished in the FailedStage. And tasks that finished during rerun of shuffleMapStage will still be counted as successful tasks and will not be rerun in retry of FailedStage.
>
> 
>
> https://github.com/apache/spark/blob/0670e4f7945ae4935261ed1f45db7ede79aca127/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1928-L1936
>
> Maybe we need to add a condition to only cancel still-running tasks which fetch from indeterministic shuffle output. Please correct me if I have any misunderstanding.
This function is for the zombie task case.
But for celeborn stage rerun, it would rerun the whole tasks.
It means that:
1. if we do not cancel the running tasks in the stage, it would occupy the resources, and the tasks in last stage to re-run might wait these tasks finished before launching
2. The tasks might be keep running with the new stage attempts tasks at the same time, it double the computing resource.
So, I think for celeborn stage rerun, cancel all the tasks directly should be fine. @Z1Wu
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r1988412053
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related task did trigger the FetchFailedException eventually(depends on whether another task attempt running or finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/ShuffleFetchFailureReportTaskCleanListener.scala:
##########
@@ -17,12 +17,26 @@
package org.apache.spark.shuffle.celeborn
+import scala.collection.JavaConverters._
+
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted}
class ShuffleFetchFailureReportTaskCleanListener extends SparkListener {
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
- SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(
- stageCompleted.stageInfo.stageId,
- stageCompleted.stageInfo.attemptNumber())
+ val stageId = stageCompleted.stageInfo.stageId
+ val stageAttemptId = stageCompleted.stageInfo.attemptNumber()
+ val shuffleFetchFailureTaskIds =
+ SparkUtils.removeStageReportedShuffleFetchFailureTaskIds(stageId, stageAttemptId)
+ if (shuffleFetchFailureTaskIds != null) {
+ shuffleFetchFailureTaskIds.asScala.headOption.foreach { case taskId =>
+ val taskSetManager = SparkUtils.getTaskSetManager(taskId)
+ if (taskSetManager != null && taskSetManager.runningTasks > 0) {
Review Comment:
even we do not know whether the shuffleFetchFailureTaskIds related task did trigger the FetchFailedException eventually(depends on whether another task attempt is running or has been finished, see https://github.com/apache/celeborn/pull/2921).
But it should be safe after we check the `taskSetManager.runningTasks > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#discussion_r2002154334
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkUtils.java:
##########
@@ -344,6 +344,20 @@ public static void cancelShuffle(int shuffleId, String reason) {
.hiddenImpl(TaskSetManager.class, "taskInfos")
.defaultAlwaysNull()
.build();
+ private static final DynFields.UnboundField<
+ scala.collection.mutable.HashMap<
+ Integer, scala.collection.mutable.HashMap<Integer, TaskSetManager>>>
+ TASK_SETS_BY_STAGE_ID_AND_ATTEMPT_FIELD =
+ DynFields.builder()
+ .hiddenImpl(TaskSchedulerImpl.class, "taskSetsByStageIdAndAttempt")
Review Comment:
Thanks, Have wrapped to code with `synchronized (taskScheduler)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2747046398
ping @RexXiong @FMX
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "Z1Wu (via GitHub)" <gi...@apache.org>.
Z1Wu commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2734252785
> For celeborn shuffle fetch failure, it will rerun the whole stage, so it is fine to cancel all the running tasks.
``` sh
shuffleMapStage -> FailedStage
```
In current implementation, when fetch failure happens without indeterministic shuffle map output, Spark will only rerun the whole shuffleMapStage and tasks that are not finished in the FailedStage. And tasks that finished during rerun of shuffleMapStage will still be counted as successful tasks and will not be rerun in retry of FailedStage.
https://github.com/apache/spark/blob/0670e4f7945ae4935261ed1f45db7ede79aca127/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1928-L1936
Please correct me if I have any misunderstanding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the running tasks if the stage is marked as failed due to shuffle fetch failure [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2715136676
cc @FMX @RexXiong @pan3793
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1904] Cancel the stage running tasks on stage rerun [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2716425363
cc @SteNicholas
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][CELEBORN-1904] Cancel the stage running tasks on stage rerun to save resource [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on PR #3144:
URL: https://github.com/apache/celeborn/pull/3144#issuecomment-2748960607
will reopen it in the future
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org