You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GavinGavinNo1 <gi...@git.apache.org> on 2017/02/08 15:10:37 UTC
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
GitHub user GavinGavinNo1 opened a pull request:
https://github.com/apache/spark/pull/16855
[SPARK-13931] Resolve stage hanging up problem in a particular case
## 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.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/GavinGavinNo1/spark resolve-stage-blocked2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/16855.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #16855
----
commit e15b2abedb6fcaf6bac8775f15bdd246fa22902e
Author: GavinGavinNo1 <ga...@gmail.com>
Date: 2017-02-08T14:51:59Z
Resolve stage hanging up problem in a particular case
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100346247
--- Diff: 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) {
--- End diff --
@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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100190631
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
--- End diff --
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)?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r103538649
--- Diff: 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("[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 = {
--- End diff --
Can you fix the indentation here? (the wrapped parameters should be indented 4 extra spaces from "override")
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Stage can hang if an executor fails while ...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the issue:
https://github.com/apache/spark/pull/16855
@GavinGavinNo1 are you ZhengYaofeng on JIRA? I want to correctly give you credit on JIRA for fixing this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Resolve stage hanging up problem in a part...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16855
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Resolve stage hanging up problem in a part...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the issue:
https://github.com/apache/spark/pull/16855
Jenkins, this is OK to test
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100190769
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ 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 tasks = Array.tabulate[Task[_]](1) { i =>
+ new ShuffleMapTask(i, 0, null, new Partition {
+ override def index: Int = 0
+ }, Seq(TaskLocation("host1", "execA")), new Properties, null)
+ }
+ val taskSet = new TaskSet(tasks, 0, 0, 0, null)
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES)
+ manager.speculatableTasks += tasks.head.partitionId
+ val task1 = manager.resourceOffer("execA", "host1", TaskLocality.PROCESS_LOCAL).get
+ 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 = ""
+ }
+ manager.handleSuccessfulTask(task1.taskId, directTaskResult)
+ assert(manager.isZombie == true)
+ assert(resubmittedTasks == 0)
--- End diff --
can you check that manager.runningTasks is 1 here, and 0 below?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100187114
--- Diff: 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) {
--- End diff --
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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100188875
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ 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 tasks = Array.tabulate[Task[_]](1) { i =>
+ new ShuffleMapTask(i, 0, null, new Partition {
+ override def index: Int = 0
+ }, Seq(TaskLocation("host1", "execA")), new Properties, null)
+ }
+ val taskSet = new TaskSet(tasks, 0, 0, 0, null)
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES)
+ manager.speculatableTasks += tasks.head.partitionId
--- End diff --
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."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Resolve stage hanging up problem in a part...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16855
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 closed the pull request at:
https://github.com/apache/spark/pull/16855
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100356980
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
--- End diff --
@kayousterhout if I use the existing FakeDAGScheduler, I'll remove variable 'resubmittedTasks', then I can't make this test failed before my code modified.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Stage can hang if an executor fails while ...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 commented on the issue:
https://github.com/apache/spark/pull/16855
@kayousterhout OK, I have updated it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100813496
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
--- End diff --
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])
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100187460
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ 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 tasks = Array.tabulate[Task[_]](1) { i =>
--- End diff --
do you need Array.tabulate here, given that you're only creating one task / task location?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
GitHub user GavinGavinNo1 reopened a pull request:
https://github.com/apache/spark/pull/16855
[SPARK-13931] Resolve stage hanging up problem in a particular case
## 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.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/GavinGavinNo1/spark resolve-stage-blocked2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/16855.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #16855
----
commit e15b2abedb6fcaf6bac8775f15bdd246fa22902e
Author: GavinGavinNo1 <ga...@gmail.com>
Date: 2017-02-08T14:51:59Z
Resolve stage hanging up problem in a particular case
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100405998
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
--- End diff --
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 ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r101940439
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
--- End diff --
@kayousterhout
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r102860963
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ var resubmittedTasks = 0
+ val dagScheduler = new FakeDAGScheduler(sc, sched) {
--- End diff --
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."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r102848520
--- Diff: 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") {
--- End diff --
can you make the description here "[SPARK-13931] taskSetManager ..." (and then eliminate the comment below)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Stage can hang if an executor fails while ...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the issue:
https://github.com/apache/spark/pull/16855
LGTM thanks for fixing this! I've merged this into master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100186410
--- Diff: 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) {
--- End diff --
nit: indentation (add two spaces)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Stage can hang if an executor fails...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/16855
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100189015
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ 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 tasks = Array.tabulate[Task[_]](1) { i =>
+ new ShuffleMapTask(i, 0, null, new Partition {
+ override def index: Int = 0
+ }, Seq(TaskLocation("host1", "execA")), new Properties, null)
+ }
+ val taskSet = new TaskSet(tasks, 0, 0, 0, null)
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES)
+ manager.speculatableTasks += tasks.head.partitionId
+ val task1 = manager.resourceOffer("execA", "host1", TaskLocality.PROCESS_LOCAL).get
+ val task2 = manager.resourceOffer("execB", "host2", TaskLocality.ANY).get
+
+ assert(manager.runningTasks == 2)
+ assert(manager.isZombie == false)
+
+ val directTaskResult = new DirectTaskResult[String](null, Seq()) {
--- End diff --
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."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Stage can hang if an executor fails while ...
Posted by GavinGavinNo1 <gi...@git.apache.org>.
Github user GavinGavinNo1 commented on the issue:
https://github.com/apache/spark/pull/16855
@kayousterhout OK
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16855: [SPARK-13931] Resolve stage hanging up problem in...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/16855#discussion_r100189090
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -664,6 +665,55 @@ 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") {
+ // 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 = {}
+ })
+
+ // count for Resubmitted tasks
+ 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 tasks = Array.tabulate[Task[_]](1) { i =>
+ new ShuffleMapTask(i, 0, null, new Partition {
+ override def index: Int = 0
+ }, Seq(TaskLocation("host1", "execA")), new Properties, null)
+ }
+ val taskSet = new TaskSet(tasks, 0, 0, 0, null)
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES)
+ manager.speculatableTasks += tasks.head.partitionId
+ val task1 = manager.resourceOffer("execA", "host1", TaskLocality.PROCESS_LOCAL).get
+ val task2 = manager.resourceOffer("execB", "host2", TaskLocality.ANY).get
+
+ assert(manager.runningTasks == 2)
--- End diff --
can you use triple equals here and below? That way Scala Test will print out the expected and actual values automatically.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16855: [SPARK-13931] Resolve stage hanging up problem in a part...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the issue:
https://github.com/apache/spark/pull/16855
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"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org