[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running #16855

GavinGavinNo1 · 2017-02-08T15:10:35Z

What changes were proposed in this pull request?

When function 'executorLost' is invoked in class 'TaskSetManager', it's significant to judge whether variable 'isZombie' is set to true.

This pull request fixes the following hang:

1.Open speculation switch in the application.
2.Run this app and suppose last task of shuffleMapStage 1 finishes. Let's get the record straight, from the eyes of DAG, this stage really finishes, and from the eyes of TaskSetManager, variable 'isZombie' is set to true, but variable runningTasksSet isn't empty because of speculation.
3.Suddenly, executor 3 is lost. TaskScheduler receiving this signal, invokes all executorLost functions of rootPool's taskSetManagers. DAG receiving this signal, removes all this executor's outputLocs.
4.TaskSetManager adds all this executor's tasks to pendingTasks and tells DAG they will be resubmitted (Attention: possibly not on time).
5.DAG starts to submit a new waitingStage, let's say shuffleMapStage 2, and going to find that shuffleMapStage 1 is its missing parent because some outputLocs are removed due to executor lost. Then DAG submits shuffleMapStage 1 again.
6.DAG still receives Task 'Resubmitted' signal from old taskSetManager, and increases the number of pendingTasks of shuffleMapStage 1 each time. However, old taskSetManager won't resolve new task to submit because its variable 'isZombie' is set to true.
7.Finally shuffleMapStage 1 never finishes in DAG together with all stages depending on it.

How was this patch tested?

It's quite difficult to construct test cases.

kayousterhout · 2017-02-08T22:20:42Z

Jenkins, this is OK to test

kayousterhout · 2017-02-08T22:21:19Z

Can you make the PR and JIRA description something more specific? Maybe "[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running"

kayousterhout · 2017-02-08T22:24:12Z

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

@@ -874,7 +874,8 @@ private[spark] class TaskSetManager(
    // and we are not using an external shuffle server which could serve the shuffle outputs.
    // The reason is the next stage wouldn't be able to fetch the data from this dead executor
    // so we would need to rerun these tasks on other executors.
-    if (tasks(0).isInstanceOf[ShuffleMapTask] && !env.blockManager.externalShuffleServiceEnabled) {
+    if (tasks(0).isInstanceOf[ShuffleMapTask] && !env.blockManager.externalShuffleServiceEnabled
+      && !isZombie) {


nit: indentation (add two spaces)

Also I'm concerned that we might need some of the functionality below even when the TSM is a zombie. While the TSM shouldn't tell the DAGScheduler that the task was resubmitted, I think it does need to notify the DAGScheduler that tasks on the executor are finished (otherwise they'll never be marked as finished in the UI, for example), and I also think it needs to properly update the running copies of the task.

@kayousterhout I think, when TSM is a zombie, resubmitted tasks won't be offered or executed, so it's no need to notify the DAGScheduler that tasks are finished.

kayousterhout · 2017-02-08T22:29:24Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+    }
+    sched.setDAGScheduler(dagScheduler)
+
+    val tasks = Array.tabulate[Task[_]](1) { i =>


do you need Array.tabulate here, given that you're only creating one task / task location?

kayousterhout · 2017-02-08T22:36:20Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+    }
+    val taskSet = new TaskSet(tasks, 0, 0, 0, null)
+    val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES)
+    manager.speculatableTasks += tasks.head.partitionId


can you add a comment here about what's going on? I think it would be more clear if you moved this line below task 1. Then before task 1, add a comment saying "Offer host1, which should be accepted as a PROCESS_LOCAL location by the one task in the task set". Then, before this speculatableTasks line, add something like "Mark the task as available for speculation, and then offer another resource, which should be used to launch a speculative copy of the task."

kayousterhout · 2017-02-08T22:36:58Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+    assert(manager.runningTasks == 2)
+    assert(manager.isZombie == false)
+
+    val directTaskResult = new DirectTaskResult[String](null, Seq()) {


here, can you add a comment with something like "Complete one copy of the task, which should result in the task set manager being marked as a zombie, because at least one copy of its only task has completed."

kayousterhout · 2017-02-08T22:37:22Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+    val task1 = manager.resourceOffer("execA", "host1", TaskLocality.PROCESS_LOCAL).get
+    val task2 = manager.resourceOffer("execB", "host2", TaskLocality.ANY).get
+
+    assert(manager.runningTasks == 2)


can you use triple equals here and below? That way Scala Test will print out the expected and actual values automatically.

kayousterhout · 2017-02-08T22:45:08Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+
+    // count for Resubmitted tasks
+    var resubmittedTasks = 0
+    val dagScheduler = new FakeDAGScheduler(sc, sched) {


Rather than defining your own DAGScheduler, can you use the existing FakeDAGSCheduler, and then use the FakeTaskScheduler to make sure that the task was recorded as ended for the correct reason? (i.e., not for the reason of being resubmitted)?

@kayousterhout if I use the existing FakeDAGScheduler, I'll remove variable 'resubmittedTasks', then I can't make this test failed before my code modified.

It doesn't work to check that the TaskEndReason was success (and not Resubmitted), like is done here: https://github.com/GavinGavinNo1/spark/blob/24d8d795d26a5b1477cac01f2748c25fb9b74dc5/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala#L226 ?

I still don't understand. I'm so confused about how to construct a failed test case before code modified, if I modify it below.

test("taskSetManager should not send Resubmitted tasks after being a zombie") { // Regression test for SPARK-13931 val conf = new SparkConf().set("spark.speculation", "true") sc = new SparkContext("local", "test", conf) val sched = new FakeTaskScheduler(sc, ("execA", "host1"), ("execB", "host2")) sched.initialize(new FakeSchedulerBackend() { override def killTask(taskId: Long, executorId: String, interruptThread: Boolean): Unit = {} }) val dagScheduler = new FakeDAGScheduler(sc, sched) sched.setDAGScheduler(dagScheduler) val singleTask = new ShuffleMapTask(0, 0, null, new Partition { override def index: Int = 0 }, Seq(TaskLocation("host1", "execA")), new Properties, null) val taskSet = new TaskSet(Array(singleTask), 0, 0, 0, null) val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES) // Offer host1, which should be accepted as a PROCESS_LOCAL location // by the one task in the task set val task1 = manager.resourceOffer("execA", "host1", TaskLocality.PROCESS_LOCAL).get // Mark the task as available for speculation, and then offer another resource, // which should be used to launch a speculative copy of the task. manager.speculatableTasks += singleTask.partitionId val task2 = manager.resourceOffer("execB", "host2", TaskLocality.ANY).get assert(manager.runningTasks === 2) assert(manager.isZombie === false) val directTaskResult = new DirectTaskResult[String](null, Seq()) { override def value(resultSer: SerializerInstance): String = "" } // Complete one copy of the task, which should result in the task set manager // being marked as a zombie, because at least one copy of its only task has completed. manager.handleSuccessfulTask(task1.taskId, directTaskResult) assert(manager.isZombie === true) assert(sched.endedTasks(0) === Success) assert(manager.runningTasks === 1) manager.executorLost("execB", "host2", new SlaveLost()) assert(manager.runningTasks === 0) assert(sched.endedTasks(0).isInstanceOf[ExecutorLostFailure]) }

@kayousterhout

I see I played around with this a bit and the problem is that the TaskSetManager also sends an ExecutorLost task failure for the task that gets resubmitted, so that failure overrides the saved Resubmitted task end reason. It's fine to leave the existing test, but can you just add a comment that says something like "Keep track of the number of tasks that are resubmitted, so that the test can check that no tasks were resubmitted."

kayousterhout · 2017-02-08T22:45:52Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+    }
+    manager.handleSuccessfulTask(task1.taskId, directTaskResult)
+    assert(manager.isZombie == true)
+    assert(resubmittedTasks == 0)


can you check that manager.runningTasks is 1 here, and 0 below?

kayousterhout · 2017-02-23T23:45:54Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

@@ -664,6 +665,63 @@ class TaskSetManagerSuite extends SparkFunSuite with LocalSparkContext with Logg
    assert(thrown2.getMessage().contains("bigger than spark.driver.maxResultSize"))
  }

+  test("taskSetManager should not send Resubmitted tasks after being a zombie") {


can you make the description here "[SPARK-13931] taskSetManager ..." (and then eliminate the comment below)

kayousterhout · 2017-02-24T01:23:50Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+
+    // count for Resubmitted tasks
+    var resubmittedTasks = 0
+    val dagScheduler = new FakeDAGScheduler(sc, sched) {


I see I played around with this a bit and the problem is that the TaskSetManager also sends an ExecutorLost task failure for the task that gets resubmitted, so that failure overrides the saved Resubmitted task end reason. It's fine to leave the existing test, but can you just add a comment that says something like "Keep track of the number of tasks that are resubmitted, so that the test can check that no tasks were resubmitted."

GavinGavinNo1 · 2017-02-28T08:37:12Z

@kayousterhout OK, I have updated it

kayousterhout

One last tiny style comment -- then this looks good

kayousterhout · 2017-02-28T19:55:16Z

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

+    var resubmittedTasks = 0
+    val dagScheduler = new FakeDAGScheduler(sc, sched) {
+      override def taskEnded(task: Task[_], reason: TaskEndReason, result: Any,
+                             accumUpdates: Seq[AccumulatorV2[_, _]], taskInfo: TaskInfo): Unit = {


Can you fix the indentation here? (the wrapped parameters should be indented 4 extra spaces from "override")

GavinGavinNo1 · 2017-03-01T03:04:59Z

@kayousterhout OK

kayousterhout · 2017-03-02T05:41:41Z

LGTM thanks for fixing this! I've merged this into master.

kayousterhout · 2017-03-02T05:43:16Z

@GavinGavinNo1 are you ZhengYaofeng on JIRA? I want to correctly give you credit on JIRA for fixing this.

Resolve stage hanging up problem in a particular case

e15b2ab

GavinGavinNo1 closed this Feb 8, 2017

GavinGavinNo1 reopened this Feb 8, 2017

GavinGavinNo1 mentioned this pull request Feb 8, 2017

[SPARK-13931] Resolve stage hanging up problem in a particular case #11760

Closed

kayousterhout reviewed Feb 8, 2017

View reviewed changes

GavinGavinNo1 changed the title ~~[SPARK-13931] Resolve stage hanging up problem in a particular case~~ [SPARK-13931] Stage can hang if an executor fails while speculated tasks are running Feb 9, 2017

optimization

24d8d79

kayousterhout suggested changes Feb 24, 2017

View reviewed changes

add comment for defining a DAGScheduler

3b73476

kayousterhout suggested changes Feb 28, 2017

View reviewed changes

format code

1c3a6ff

asfgit closed this in 89990a0 Mar 2, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running #16855

[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running #16855

GavinGavinNo1 commented Feb 8, 2017

kayousterhout commented Feb 8, 2017

kayousterhout commented Feb 8, 2017

kayousterhout Feb 8, 2017

kayousterhout Feb 8, 2017

GavinGavinNo1 Feb 9, 2017

kayousterhout Feb 8, 2017

kayousterhout Feb 8, 2017

kayousterhout Feb 8, 2017

kayousterhout Feb 8, 2017

kayousterhout Feb 8, 2017

GavinGavinNo1 Feb 9, 2017

kayousterhout Feb 9, 2017

GavinGavinNo1 Feb 13, 2017 •

edited

Loading

GavinGavinNo1 Feb 20, 2017

kayousterhout Feb 24, 2017

kayousterhout Feb 8, 2017

kayousterhout Feb 23, 2017

kayousterhout Feb 24, 2017

GavinGavinNo1 commented Feb 28, 2017

kayousterhout left a comment

kayousterhout Feb 28, 2017

GavinGavinNo1 commented Mar 1, 2017

kayousterhout commented Mar 2, 2017

kayousterhout commented Mar 2, 2017

[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running #16855

[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running #16855

Conversation

GavinGavinNo1 commented Feb 8, 2017

What changes were proposed in this pull request?

How was this patch tested?

kayousterhout commented Feb 8, 2017

kayousterhout commented Feb 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GavinGavinNo1 Feb 13, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GavinGavinNo1 commented Feb 28, 2017

kayousterhout left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GavinGavinNo1 commented Mar 1, 2017

kayousterhout commented Mar 2, 2017

kayousterhout commented Mar 2, 2017

GavinGavinNo1 Feb 13, 2017 •

edited

Loading