SPARK-2294: fix locality inversion bug in TaskManager #1313

CodingCat · 2014-07-07T04:30:05Z

copied from original JIRA (https://issues.apache.org/jira/browse/SPARK-2294):

If an executor E is free, a task may be speculatively assigned to E when there are other tasks in the job that have not been launched (at all) yet. Similarly, a task without any locality preferences may be assigned to E when there was another NODE_LOCAL task that could have been scheduled.
This happens because TaskSchedulerImpl calls TaskSetManager.resourceOffer (which in turn calls TaskSetManager.findTask) with increasing locality levels, beginning with PROCESS_LOCAL, followed by NODE_LOCAL, and so on until the highest currently allowed level. Now, supposed NODE_LOCAL is the highest currently allowed locality level. The first time findTask is called, it will be called with max level PROCESS_LOCAL; if it cannot find any PROCESS_LOCAL tasks, it will try to schedule tasks with no locality preferences or speculative tasks. As a result, speculative tasks or tasks with no preferences may be scheduled instead of NODE_LOCAL tasks.

I added an additional parameter in resourceOffer and findTask, maxLocality, indicating when we should consider the tasks without locality preference

AmplabJenkins · 2014-07-07T04:31:01Z

Merged build triggered.

AmplabJenkins · 2014-07-07T04:31:10Z

Merged build started.

AmplabJenkins · 2014-07-07T06:32:17Z

Merged build finished.

AmplabJenkins · 2014-07-07T06:32:18Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16360/

mridulm · 2014-07-07T09:44:33Z

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

@@ -238,7 +238,7 @@ private[spark] class TaskSetManager(
   */
  private def findTaskFromList(execId: String, list: ArrayBuffer[Int]): Option[Int] = {
    var indexOffset = list.size
-
+    println("findingTaskFromList:" + list)


remove this ? :-)

mridulm · 2014-07-07T09:50:31Z

Might be simpler to move the noprefs block inside the NODE_LOCAL condition [1] - after NODE_LOCAL tasks have been checked.

[1] if (TaskLocality.isAllowed(locality, TaskLocality.NODE_LOCAL)

AmplabJenkins · 2014-07-07T11:01:03Z

Merged build triggered.

AmplabJenkins · 2014-07-07T11:01:13Z

Merged build started.

AmplabJenkins · 2014-07-07T11:16:02Z

Merged build triggered.

AmplabJenkins · 2014-07-07T11:16:07Z

Merged build started.

AmplabJenkins · 2014-07-07T13:02:27Z

Merged build finished.

AmplabJenkins · 2014-07-07T13:02:27Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16370/

AmplabJenkins · 2014-07-07T13:17:12Z

Merged build finished.

AmplabJenkins · 2014-07-07T13:17:13Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16371/

AmplabJenkins · 2014-07-07T14:26:03Z

Merged build triggered.

AmplabJenkins · 2014-07-07T14:26:15Z

Merged build started.

AmplabJenkins · 2014-07-07T16:27:27Z

Merged build finished.

AmplabJenkins · 2014-07-07T16:27:27Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16373/

CodingCat · 2014-07-07T17:26:12Z

close temporarily for local test.....

AmplabJenkins · 2014-07-07T22:11:04Z

Merged build triggered.

AmplabJenkins · 2014-07-07T22:11:11Z

Merged build started.

CodingCat · 2014-07-07T22:51:02Z

will this modification cause streaming test case failed (timeout)?

I cannot reproduce in my local side constantly.....

AmplabJenkins · 2014-07-08T00:12:35Z

Merged build finished.

AmplabJenkins · 2014-07-08T00:12:35Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16388/

AmplabJenkins · 2014-07-08T00:16:04Z

Merged build triggered.

AmplabJenkins · 2014-07-08T00:16:12Z

Merged build started.

AmplabJenkins · 2014-07-08T02:17:20Z

Merged build finished.

CodingCat · 2014-08-05T03:31:49Z

finally,

mateiz · 2014-08-05T18:01:30Z

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

@@ -293,7 +291,7 @@ private[spark] class TaskSetManager(
      for (index <- speculatableTasks if canRunOnHost(index)) {
        val prefs = tasks(index).preferredLocations
        val executors = prefs.flatMap(_.executorId)
-        if (prefs.size == 0 || executors.contains(execId)) {
+        if (executors.contains(execId)) {


In the comment above this, remove "preference-less tasks" since we've moved them below

mateiz · 2014-08-05T18:05:47Z

Thanks, this looks good. I made some more small comments on it and there were a couple left over from before. Please take a look at https://github.com/apache/spark/pull/1313/files and fix them. Then I think it will be good to go!

SparkQA · 2014-08-05T23:29:35Z

QA tests have started for PR 1313. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17957/consoleFull

SparkQA · 2014-08-06T00:27:33Z

QA results for PR 1313:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17957/consoleFull

mateiz · 2014-08-06T01:09:19Z

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

@@ -192,56 +211,59 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
    val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES, clock)

    // An executor that is not NODE_LOCAL should be rejected.
-    assert(manager.resourceOffer("execC", "host2", ANY) === None)
+    assert(manager.resourceOffer("execC", "host2", NODE_LOCAL) === None)


Hey sorry, one other question.. why did you change this? It seems that this breaks what we were testing for before.

mateiz · 2014-08-06T01:18:47Z

Cool, thanks! One final thing, I made some comments on the tests since it seemed that some of the changes broke previous stuff we were testing for. If they didn't, please explain why you changed them.

SparkQA · 2014-08-06T01:39:26Z

QA tests have started for PR 1313. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17970/consoleFull

CodingCat · 2014-08-06T02:37:07Z

that's for fitting some early version of the patches, sorry about that, I just forgot to undo the changes...

mateiz · 2014-08-06T02:47:04Z

Alright, thanks for all your work on this. Just want to make sure we don't miss stuff here.

SparkQA · 2014-08-06T02:51:51Z

QA results for PR 1313:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17970/consoleFull

CodingCat · 2014-08-06T02:52:42Z

thanks for the patient review @mateiz @mridulm @kayousterhout @lirui-intel

mateiz · 2014-08-06T06:03:42Z

Alright, I've merged this in. Thanks Nan!

copied from original JIRA (https://issues.apache.org/jira/browse/SPARK-2294): If an executor E is free, a task may be speculatively assigned to E when there are other tasks in the job that have not been launched (at all) yet. Similarly, a task without any locality preferences may be assigned to E when there was another NODE_LOCAL task that could have been scheduled. This happens because TaskSchedulerImpl calls TaskSetManager.resourceOffer (which in turn calls TaskSetManager.findTask) with increasing locality levels, beginning with PROCESS_LOCAL, followed by NODE_LOCAL, and so on until the highest currently allowed level. Now, supposed NODE_LOCAL is the highest currently allowed locality level. The first time findTask is called, it will be called with max level PROCESS_LOCAL; if it cannot find any PROCESS_LOCAL tasks, it will try to schedule tasks with no locality preferences or speculative tasks. As a result, speculative tasks or tasks with no preferences may be scheduled instead of NODE_LOCAL tasks. ---- I added an additional parameter in resourceOffer and findTask, maxLocality, indicating when we should consider the tasks without locality preference Author: CodingCat <[email protected]> Closes #1313 from CodingCat/SPARK-2294 and squashes the following commits: bf3f13b [CodingCat] rollback some forgotten changes 89f9bc0 [CodingCat] address matei's comments 18cae02 [CodingCat] add test case for node-local tasks 2ba6195 [CodingCat] fix failed test cases 87dd09e [CodingCat] fix style 9b9432f [CodingCat] remove hasNodeLocalOnlyTasks fdd1573 [CodingCat] fix failed test cases 941a4fd [CodingCat] see my shocked face.......... f600085 [CodingCat] remove hasNodeLocalOnlyTasks checking 0b8a46b [CodingCat] test whether hasNodeLocalOnlyTasks affect the results 73ceda8 [CodingCat] style fix b3a430b [CodingCat] remove fine granularity tracking for node-local only tasks f9a2ad8 [CodingCat] simplify the logic in TaskSchedulerImpl c8c1de4 [CodingCat] simplify the patch be652ed [CodingCat] avoid unnecessary delay when we only have nopref tasks dee9e22 [CodingCat] fix locality inversion bug in TaskManager by moving nopref branch (cherry picked from commit 63bdb1f) Signed-off-by: Matei Zaharia <[email protected]>

copied from original JIRA (https://issues.apache.org/jira/browse/SPARK-2294): If an executor E is free, a task may be speculatively assigned to E when there are other tasks in the job that have not been launched (at all) yet. Similarly, a task without any locality preferences may be assigned to E when there was another NODE_LOCAL task that could have been scheduled. This happens because TaskSchedulerImpl calls TaskSetManager.resourceOffer (which in turn calls TaskSetManager.findTask) with increasing locality levels, beginning with PROCESS_LOCAL, followed by NODE_LOCAL, and so on until the highest currently allowed level. Now, supposed NODE_LOCAL is the highest currently allowed locality level. The first time findTask is called, it will be called with max level PROCESS_LOCAL; if it cannot find any PROCESS_LOCAL tasks, it will try to schedule tasks with no locality preferences or speculative tasks. As a result, speculative tasks or tasks with no preferences may be scheduled instead of NODE_LOCAL tasks. ---- I added an additional parameter in resourceOffer and findTask, maxLocality, indicating when we should consider the tasks without locality preference Author: CodingCat <[email protected]> Closes apache#1313 from CodingCat/SPARK-2294 and squashes the following commits: bf3f13b [CodingCat] rollback some forgotten changes 89f9bc0 [CodingCat] address matei's comments 18cae02 [CodingCat] add test case for node-local tasks 2ba6195 [CodingCat] fix failed test cases 87dd09e [CodingCat] fix style 9b9432f [CodingCat] remove hasNodeLocalOnlyTasks fdd1573 [CodingCat] fix failed test cases 941a4fd [CodingCat] see my shocked face.......... f600085 [CodingCat] remove hasNodeLocalOnlyTasks checking 0b8a46b [CodingCat] test whether hasNodeLocalOnlyTasks affect the results 73ceda8 [CodingCat] style fix b3a430b [CodingCat] remove fine granularity tracking for node-local only tasks f9a2ad8 [CodingCat] simplify the logic in TaskSchedulerImpl c8c1de4 [CodingCat] simplify the patch be652ed [CodingCat] avoid unnecessary delay when we only have nopref tasks dee9e22 [CodingCat] fix locality inversion bug in TaskManager by moving nopref branch

mridulm reviewed Jul 7, 2014
View reviewed changes

CodingCat closed this Jul 7, 2014

CodingCat reopened this Jul 7, 2014

mateiz reviewed Aug 5, 2014
View reviewed changes

address matei's comments

89f9bc0

mateiz reviewed Aug 6, 2014
View reviewed changes

rollback some forgotten changes

bf3f13b

asfgit closed this in 63bdb1f Aug 6, 2014

JoshRosen mentioned this pull request Sep 3, 2014

[SPARK-2193] Improve tasks preferrd locality by sorting tasks partial or... #1131

Closed

sryza mentioned this pull request Dec 29, 2014

SPARK-4921. TaskSetManager.dequeueTask returns PROCESS_LOCAL for NO_PREF ... #3816

Closed

squito mentioned this pull request Jul 12, 2019

[SPARK-26755][SCHEDULER] : Optimize Spark Scheduler to dequeue speculative tasks… #23677

Closed

SPARK-2294: fix locality inversion bug in TaskManager #1313

SPARK-2294: fix locality inversion bug in TaskManager #1313

Uh oh!

Conversation

CodingCat commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

mridulm Jul 7, 2014

Choose a reason for hiding this comment

Uh oh!

mridulm commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

CodingCat commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 7, 2014

Uh oh!

CodingCat commented Jul 7, 2014

Uh oh!

AmplabJenkins commented Jul 8, 2014

Uh oh!

AmplabJenkins commented Jul 8, 2014

Uh oh!

AmplabJenkins commented Jul 8, 2014

Uh oh!

AmplabJenkins commented Jul 8, 2014

Uh oh!

AmplabJenkins commented Jul 8, 2014

Uh oh!

CodingCat commented Aug 5, 2014

Uh oh!

mateiz Aug 5, 2014

Choose a reason for hiding this comment

Uh oh!

mateiz commented Aug 5, 2014

Uh oh!

SparkQA commented Aug 5, 2014

Uh oh!

SparkQA commented Aug 6, 2014

Uh oh!

mateiz Aug 6, 2014

Choose a reason for hiding this comment

Uh oh!

mateiz commented Aug 6, 2014

Uh oh!

SparkQA commented Aug 6, 2014

Uh oh!

CodingCat commented Aug 6, 2014