You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2017/03/02 05:41:26 UTC
spark git commit: [SPARK-13931] Stage can hang if an executor fails
while speculated tasks are running
Repository: spark
Updated Branches:
refs/heads/master 51be63365 -> 89990a010
[SPARK-13931] Stage can hang if an executor fails while speculated tasks are running
## 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.
Author: GavinGavinNo1 <ga...@gmail.com>
Author: 16092929 <16...@cnsuning.com>
Closes #16855 from GavinGavinNo1/resolve-stage-blocked2.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/89990a01
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/89990a01
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/89990a01
Branch: refs/heads/master
Commit: 89990a01099b2d632b65112eb755de648aa54c16
Parents: 51be633
Author: GavinGavinNo1 <ga...@gmail.com>
Authored: Wed Mar 1 21:40:41 2017 -0800
Committer: Kay Ousterhout <ka...@gmail.com>
Committed: Wed Mar 1 21:40:41 2017 -0800
----------------------------------------------------------------------
.../apache/spark/scheduler/TaskSetManager.scala | 3 +-
.../spark/scheduler/TaskSetManagerSuite.scala | 64 +++++++++++++++++++-
2 files changed, 65 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/89990a01/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index e63feb8..19ebaf8 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/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) {
for ((tid, info) <- taskInfos if info.executorId == execId) {
val index = taskInfos(tid).index
if (successful(index)) {
http://git-wip-us.apache.org/repos/asf/spark/blob/89990a01/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
index d03a0c9..2c2cda9 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.scheduler
-import java.util.Random
+import java.util.{Properties, Random}
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -28,6 +28,7 @@ import org.mockito.Mockito.{mock, never, spy, verify, when}
import org.apache.spark._
import org.apache.spark.internal.config
import org.apache.spark.internal.Logging
+import org.apache.spark.serializer.SerializerInstance
import org.apache.spark.storage.BlockManagerId
import org.apache.spark.util.{AccumulatorV2, ManualClock}
@@ -664,6 +665,67 @@ class TaskSetManagerSuite extends SparkFunSuite with LocalSparkContext with Logg
assert(thrown2.getMessage().contains("bigger than spark.driver.maxResultSize"))
}
+ test("[SPARK-13931] taskSetManager should not send Resubmitted tasks after being a zombie") {
+ 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 = {}
+ })
+
+ // Keep track of the number of tasks that are resubmitted,
+ // so that the test can check that no tasks were resubmitted.
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
+ override def taskEnded(
+ task: Task[_],
+ reason: TaskEndReason,
+ result: Any,
+ accumUpdates: Seq[AccumulatorV2[_, _]],
+ taskInfo: TaskInfo): Unit = {
+ super.taskEnded(task, reason, result, accumUpdates, taskInfo)
+ reason match {
+ case Resubmitted => resubmittedTasks += 1
+ case _ =>
+ }
+ }
+ }
+ 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(resubmittedTasks === 0)
+ assert(manager.runningTasks === 1)
+
+ manager.executorLost("execB", "host2", new SlaveLost())
+ assert(manager.runningTasks === 0)
+ assert(resubmittedTasks === 0)
+ }
+
test("speculative and noPref task should be scheduled after node-local") {
sc = new SparkContext("local", "test")
sched = new FakeTaskScheduler(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org