You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by davies <gi...@git.apache.org> on 2014/12/23 21:04:15 UTC
[GitHub] spark pull request: [SPARK-4939] move to next locality when no pen...
GitHub user davies opened a pull request:
https://github.com/apache/spark/pull/3779
[SPARK-4939] move to next locality when no pending tasks for executors
Currently, if there are different locality in a task set, the tasks with NODE_LOCAL only get scheduled after all the PROCESS_LOCAL tasks are scheduled and timeout with spark.locality.wait.process (3 seconds by default). In local mode, the LocalScheduler will never call resourceOffer() again once it failed to get a task with same locality, then all the NODE_LOCAL tasks will be never scheduled.
This bug could be reproduced by run example python/streaming/stateful_network_wordcount.py, it will hang after finished a batch with some data.
This patch will try to remove the `execId` from `pendingTasksForExecutor` if no more tasks for this executor, it's cheched once for each launched PROCESS_LOCAL tasks. Then we can check whether there are PROCESS_LOCAL tasks or not by call `pendingTasksForExecutor.isEmpty`, which is cheap. Finally, we can change to next locality level once there is no more PROCESS_LOCAL tasks without waiting for `spark.locality.wait.process` seconds.
The tasks in pendingTasksForHost and pendingTasksForRack are the same set of tasks, so we don't need this logic for them.
cc @tdas
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/davies/spark local_streaming
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/3779.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 #3779
----
commit 7d8c5a5e83fb2516babf4c3fd99e8392edffadb6
Author: Davies Liu <da...@databricks.com>
Date: 2014-12-23T19:49:04Z
jump to next locality if no pending tasks for executors
----
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70702131
[Test build #566 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/566/consoleFull) for PR 3779 at commit [`37a2804`](https://github.com/apache/spark/commit/37a28040750219a4e493d820bdb3e0feeb5398a4).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70587469
**[Test build #25774 timed out](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25774/consoleFull)** for PR 3779 at commit [`49bda82`](https://github.com/apache/spark/commit/49bda82f2e1716c2cc3ace9468d18e3b1e332522) after a configured wait of `120m`.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065269
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
+ case TaskLocality.RACK_LOCAL => hasMoreTasks(pendingTasksForRack)
+ }
+ if (!moreTasks) {
+ // Move to next locality level if there is no task for current level
+ lastLaunchTime = curTime
+ logDebug(s"No tasks for locality level ${myLocalityLevels(currentLocalityIndex)} " +
+ s"move to ${myLocalityLevels(currentLocalityIndex + 1)}")
--- End diff --
nit: "move to..." ==> "so moving to locality level ..."
---
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: [SPARK-4939] move to next locality when no pen...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72586368
> If we have tasks at some high locality level that cannot be scheduled at their preferred locality and there are tasks at a lower locality level that can be scheduled, then I guess we might be concerned about whether scheduling the less-local tasks could rob tasks waiting at a higher locality level of their opportunity to run. This shouldn't happen, though, since those tasks will already have been offered those resources and turned them down.
Actually, on reflection, I guess this gets trickier when you have multiple task sets: in this case, you want to ensure that _all_ unscheduled process-local tasks have turned down a resource before you allow _any_ node-local task to obtain it.
I'm going to try to trace through the interaction between DAGScheduler and TaskSetManager to try to understand how we handle this 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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70350611
Actually I have a slightly different question: should we be doing the same thing for other locality levels?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72584682
_(Stream-of-consciousness ahead, mostly for my own benefit as I think through the implications of this PR)_
I'd like to understand whether this patch has any performance implications for Spark jobs in general. Is there any scenario where this scheduling change might introduce performance regressions?
I guess that a task's locality wait is a sort of hard deadline, where we won't consider scheduling a task at a lower locality level until at least that much time has elapsed. This sounds like a per-task property, where the deadline for one task shouldn't apply / influence other tasks (e.g. treat each task independently), but the bug reported here sounds like we're applying the locality waits to sets of tasks.
It seems like we always want to attempt to schedule tasks in decreasing order of locality, so if there are unscheduled process-local tasks then we should always attempt to schedule them before any tasks at lower locality levels. In addition, if there are free slots in the cluster and there are unscheduled tasks in the scheduler's current locality level, then we should schedule those tasks. If we've exhausted all tasks at a particular locality level, then it makes sense to immediately move onto scheduling at the next lower locality level.
If we have tasks at some high locality level that cannot be scheduled at their preferred locality and there are tasks at a lower locality level that _can_ be scheduled, then I guess we might be concerned about whether scheduling the less-local tasks could rob tasks waiting at a higher locality level of their opportunity to run. This shouldn't happen, though, since those tasks will already have been offered those resources and turned them down.
Therefore, I think that this patch is a good fix: it doesn't make sense to let a single task with strong preferences to delay / block the scheduling of other tasks with weaker preferences / preferences for other resources. I'll take a closer look at the code and tests now.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72953034
Thanks all! I've merged this into 1.3 and 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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70575987
[Test build #25774 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25774/consoleFull) for PR 3779 at commit [`49bda82`](https://github.com/apache/spark/commit/49bda82f2e1716c2cc3ace9468d18e3b1e332522).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23339119
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -495,13 +495,39 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // remove the emptyList from pendingTasks lazily
+ def hasNonEmptyList(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
--- End diff --
I had updated the patch to check whether tasks are scheduled or finished, the above rule also apply to them. It's similar to dequeueTaskFromList()
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23332765
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -313,7 +313,8 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
test("delay scheduling with failed hosts") {
sc = new SparkContext("local", "test")
- val sched = new FakeTaskScheduler(sc, ("exec1", "host1"), ("exec2", "host2"))
+ val sched = new FakeTaskScheduler(sc, ("exec1", "host1"), ("exec2", "host2"),
+ ("exec3", "host3"))
--- End diff --
Why this change?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68318545
[Test build #24870 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/24870/consoleFull) for PR 3779 at commit [`32d363f`](https://github.com/apache/spark/commit/32d363f275c6dea225019a1cb23af4eeeb44781f).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72920317
LGTM (but will wait to merge until @markhamstra and @JoshRosen give the thumb's up, just in case they have any remaining comments)
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72810958
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26733/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24067073
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,63 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def tasksNeedToBeScheduledFrom(pendingTaskIds: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = pendingTaskIds.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = pendingTaskIds(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ pendingTaskIds.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // Walk through the list of tasks that can be scheduled at each location and returns true
+ // if there are any tasks that still need to be scheduled. Lazily cleans up tasks that have
+ // already been scheduled.
+ def noMoreTasksToRunIn(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (tasksNeedToBeScheduledFrom(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => noMoreTasksToRunIn(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => noMoreTasksToRunIn(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.nonEmpty
--- End diff --
Wait, are we going with moreTasks or noMoreTasks? I think this needs to be `val noMoreTasks`, `case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty` and `if (noMoreTasks)` a couple lines down -- or consistently reverse the boolean sense of all that.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24067444
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,63 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def tasksNeedToBeScheduledFrom(pendingTaskIds: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = pendingTaskIds.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = pendingTaskIds(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ pendingTaskIds.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // Walk through the list of tasks that can be scheduled at each location and returns true
+ // if there are any tasks that still need to be scheduled. Lazily cleans up tasks that have
+ // already been scheduled.
+ def noMoreTasksToRunIn(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (tasksNeedToBeScheduledFrom(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => noMoreTasksToRunIn(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => noMoreTasksToRunIn(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.nonEmpty
--- End diff --
noMoreTasksToRunIn => moreTasksToRunIn()
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70586162
**[Test build #25773 timed out](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25773/consoleFull)** for PR 3779 at commit [`d8fb95a`](https://github.com/apache/spark/commit/d8fb95a2d959f9b0e262e29860f44344a855aa38) after a configured wait of `120m`.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68105490
@davies can you add a unit test that fails in the old code but works with your code? It would be helpful to more clearly document the exact bug.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72116552
@pwendell As pointed out in #4147 , the fix in LocalBackend still have several seconds delay in each stage, it still hurt Spark Streaming. Or should we set `locality.wait` to `0` for all streaming jobs?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68318551
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24870/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72803322
[Test build #26732 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26732/consoleFull) for PR 3779 at commit [`1550668`](https://github.com/apache/spark/commit/1550668b047f8702a6c7b19aad56cfd7a56e47c3).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72604521
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26612/
Test PASSed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72597115
[Test build #26612 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26612/consoleFull) for PR 3779 at commit [`6b13824`](https://github.com/apache/spark/commit/6b13824badb6f47b102accd93742e7ce5b36364b).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68324255
[Test build #556 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/556/consoleFull) for PR 3779 at commit [`32d363f`](https://github.com/apache/spark/commit/32d363f275c6dea225019a1cb23af4eeeb44781f).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68319961
[Test build #556 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/556/consoleFull) for PR 3779 at commit [`32d363f`](https://github.com/apache/spark/commit/32d363f275c6dea225019a1cb23af4eeeb44781f).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72805131
[Test build #26733 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26733/consoleFull) for PR 3779 at commit [`2d25fb3`](https://github.com/apache/spark/commit/2d25fb38aab5bac1c4cdb7d36ed2ff75f7d16a57).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70149578
@kayousterhout Could you take a look at this. This is priority for 1.3 :)
---
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: [SPARK-4939] move to next locality when no pen...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23981098
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(list: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = list.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = list(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ list.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (key: String, list: ArrayBuffer[Int]) =>
--- End diff --
Similarly, could you use a more descriptive name for `key` and `list`? Is key supposed to be an executor ID? A host ID?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68000310
[Test build #24746 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/24746/consoleFull) for PR 3779 at commit [`7d8c5a5`](https://github.com/apache/spark/commit/7d8c5a5e83fb2516babf4c3fd99e8392edffadb6).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24067496
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,63 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def tasksNeedToBeScheduledFrom(pendingTaskIds: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = pendingTaskIds.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = pendingTaskIds(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ pendingTaskIds.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // Walk through the list of tasks that can be scheduled at each location and returns true
+ // if there are any tasks that still need to be scheduled. Lazily cleans up tasks that have
+ // already been scheduled.
+ def noMoreTasksToRunIn(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (tasksNeedToBeScheduledFrom(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => noMoreTasksToRunIn(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => noMoreTasksToRunIn(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.nonEmpty
--- End diff --
oops, I just rush it too fast without check it carefully.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065873
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
+ case TaskLocality.RACK_LOCAL => hasMoreTasks(pendingTasksForRack)
+ }
+ if (!moreTasks) {
+ // Move to next locality level if there is no task for current level
--- End diff --
Add something like: "This is a performance optimization: if there are no more tasks that can be scheduled at a particular locality level, there is no point in waiting for the locality wait timeout (SPARK-4939)."
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065900
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
--- End diff --
But shouldn't it be !pendingTasksWithNoPrefs.isEmpty?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70932412
[Test build #25918 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25918/consoleFull) for PR 3779 at commit [`28d1b3c`](https://github.com/apache/spark/commit/28d1b3cbaf88a476dd9711921486708778c450c4).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70714339
[Test build #566 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/566/consoleFull) for PR 3779 at commit [`37a2804`](https://github.com/apache/spark/commit/37a28040750219a4e493d820bdb3e0feeb5398a4).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70955949
[Test build #25931 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25931/consoleFull) for PR 3779 at commit [`414e79e`](https://github.com/apache/spark/commit/414e79eafc7e3b49b63bd7c626459a748c5ced02).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065322
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
+ case TaskLocality.RACK_LOCAL => hasMoreTasks(pendingTasksForRack)
+ }
+ if (!moreTasks) {
+ // Move to next locality level if there is no task for current level
+ lastLaunchTime = curTime
+ logDebug(s"No tasks for locality level ${myLocalityLevels(currentLocalityIndex)} " +
+ s"move to ${myLocalityLevels(currentLocalityIndex + 1)}")
+ currentLocalityIndex += 1
+ } else if (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex)) {
+ // Jump to the next locality level, and remove our waiting time for the current one since
+ // we don't want to count it again on the next one
+ lastLaunchTime += localityWaits(currentLocalityIndex)
+ currentLocalityIndex += 1
+ logDebug(s"Move to ${myLocalityLevels(currentLocalityIndex)} after wait for " +
--- End diff --
s"Moving to ${..} after waiting for ${..}ms"
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72829704
[Test build #578 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/578/consoleFull) for PR 3779 at commit [`2d25fb3`](https://github.com/apache/spark/commit/2d25fb38aab5bac1c4cdb7d36ed2ff75f7d16a57).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70349488
Revive offers periodically can fix the bug, but introduce 1 second delay (by default) for every job, it will be noticeable for streaming workload.
I think it's a bug related to locality, the workaround just hide the bug.
@tdas How do you think?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70574384
[Test build #25773 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25773/consoleFull) for PR 3779 at commit [`d8fb95a`](https://github.com/apache/spark/commit/d8fb95a2d959f9b0e262e29860f44344a855aa38).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23332736
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -495,13 +495,39 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // remove the emptyList from pendingTasks lazily
+ def hasNonEmptyList(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
--- End diff --
In a big cluster, walking through all of these maps each time seems like it could be pretty expensive, right? Why do this lazily?
Also, it occurred to me that we don't remove tasks from these lists when they're scheduled, so this doesn't completely solve the performance issue (because there could still be a task in one of these lists that will never be scheduled).
This is getting fairly complex to change, and I'm wondering whether it's worthwhile. Do you have a sense of whether these issues are a problem beyond just the local case where people run streaming jobs that fail? And if it's only a problem for the local case, should we just fix it in the slightly hacky way by adding a call to revive offers to the local scheduler?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24048516
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
--- End diff --
A comment clarifying why the NO_PREF case seems to have the opposite sense of the other cases (not-empty vs. empty) would be helpful.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70947373
[Test build #25931 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25931/consoleFull) for PR 3779 at commit [`414e79e`](https://github.com/apache/spark/commit/414e79eafc7e3b49b63bd7c626459a748c5ced02).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72803247
@kayousterhout @markhamstra @JoshRosen I should had addressed all your comments, they are very helpful, thanks you all!
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24066698
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
--- End diff --
Ah yes, I see now. So, leave it as `id` and slip in the `key could be executorId, host, or rackId` comment.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23336266
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -313,7 +313,8 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
test("delay scheduling with failed hosts") {
sc = new SparkContext("local", "test")
- val sched = new FakeTaskScheduler(sc, ("exec1", "host1"), ("exec2", "host2"))
+ val sched = new FakeTaskScheduler(sc, ("exec1", "host1"), ("exec2", "host2"),
+ ("exec3", "host3"))
--- End diff --
Because it will change the locality level to NO_PREF if no active executor available.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065502
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
+ case TaskLocality.RACK_LOCAL => hasMoreTasks(pendingTasksForRack)
+ }
+ if (!moreTasks) {
+ // Move to next locality level if there is no task for current level
+ lastLaunchTime = curTime
+ logDebug(s"No tasks for locality level ${myLocalityLevels(currentLocalityIndex)} " +
+ s"move to ${myLocalityLevels(currentLocalityIndex + 1)}")
+ currentLocalityIndex += 1
+ } else if (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex)) {
+ // Jump to the next locality level, and remove our waiting time for the current one since
+ // we don't want to count it again on the next one
--- End diff --
I know this existed before your change, but can you change this comment to: "Jump to the next locality level, and reset lastLaunchTime so that the next locality wait timer doesn't immediately expire"? I think that would make this easier to understand
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72922174
LGTMT
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72808682
[Test build #577 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/577/consoleFull) for PR 3779 at commit [`2d25fb3`](https://github.com/apache/spark/commit/2d25fb38aab5bac1c4cdb7d36ed2ff75f7d16a57).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23981053
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(list: ArrayBuffer[Int]): Boolean = {
--- End diff --
Can you give `list` a more descriptive variable name? Reading top-to-bottom, it's not obvious what it's representing.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24049293
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
+ case TaskLocality.RACK_LOCAL => hasMoreTasks(pendingTasksForRack)
+ }
+ if (!moreTasks) {
+ // Move to next locality level if there is no task for current level
+ lastLaunchTime = curTime
+ logDebug(s"No tasks for locality level ${myLocalityLevels(currentLocalityIndex)} " +
+ s"move to ${myLocalityLevels(currentLocalityIndex + 1)}")
+ currentLocalityIndex += 1
+ } else if (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex)) {
+ // Jump to the next locality level, and remove our waiting time for the current one since
+ // we don't want to count it again on the next one
+ lastLaunchTime += localityWaits(currentLocalityIndex)
+ currentLocalityIndex += 1
+ logDebug(s"Move to ${myLocalityLevels(currentLocalityIndex)} after wait for " +
+ s"${localityWaits(currentLocalityIndex)} ms")
+ } else {
+ return myLocalityLevels(currentLocalityIndex)
+ }
--- End diff --
This looks correct to me, but I'm not finding the naming helpful -- in particular, the excess of "index" in `hasNotScheduledTasks`. I'd suggest something more like this to improve readability (although I'm being a little indecisive about "run" vs. "schedule"):
```scala
def tasksNeedToBeScheduledFrom(pendingTaskIds: ArrayBuffer[Int]): Boolean = {
var idOffset = pendingTaskIds.size
while (idOffset > 0) {
idOffset -= 1
val index = pendingTaskIds(idOffset)
if (copiesRunning(index) == 0 && !successful(index)) {
return true
} else {
pendingTaskIds.remove(idOffset)
}
}
false
}
// It removes the empty lists after check
def noMoreTasksToRunIn(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
val emptyKeys = new ArrayBuffer[String]
val moreTasksToRun = pendingTasks.exists {
case (id: String, pendingTaskIds: ArrayBuffer[Int]) =>
if (tasksNeedToBeScheduledFrom(pendingTaskIds)) {
true
} else {
emptyKeys += id
false
}
}
emptyKeys.foreach(x => pendingTasks.remove(x))
!moreTasksToRun
}
while (currentLocalityIndex < myLocalityLevels.length - 1) {
val noMoreTasksToRun = myLocalityLevels(currentLocalityIndex) match {
case TaskLocality.PROCESS_LOCAL => noMoreTasksToRunIn(pendingTasksForExecutor)
case TaskLocality.NODE_LOCAL => noMoreTasksToRunIn(pendingTasksForHost)
case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.nonEmpty
case TaskLocality.RACK_LOCAL => noMoreTasksToRunIn(pendingTasksForRack)
}
if (noMoreTasksToRun) {
.
.
.
```
`getAllowedLocalityLevel` also doesn't give adequate notice of the side effect that modifies `emptyKeys`, but I haven't got a good solution for that -- other than hoisting and intensifying the comment before `hasMoreTasks`.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23121685
--- Diff: core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala ---
@@ -635,6 +635,28 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
assert(manager.resourceOffer("execA", "host3", NO_PREF).get.index === 2)
}
+ // regression test for SPARK-4939
+ test("node-local tasks should be scheduled right after process-local tasks finished") {
--- End diff --
Can you just change the name of the test to incorporate the JIRA -- so remove the comment and make it:
test("SPARK-4939: node-local...")
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-67992975
[Test build #24746 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/24746/consoleFull) for PR 3779 at commit [`7d8c5a5`](https://github.com/apache/spark/commit/7d8c5a5e83fb2516babf4c3fd99e8392edffadb6).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68396660
[Test build #24903 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/24903/consoleFull) for PR 3779 at commit [`2d6ae73`](https://github.com/apache/spark/commit/2d6ae731debf10178e0675afa1d6f2a8b58edd0e).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68000315
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24746/
Test PASSed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72604515
[Test build #26612 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26612/consoleFull) for PR 3779 at commit [`6b13824`](https://github.com/apache/spark/commit/6b13824badb6f47b102accd93742e7ce5b36364b).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065454
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
--- End diff --
+1 (is this a mistake?)
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72810950
[Test build #26733 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26733/consoleFull) for PR 3779 at commit [`2d25fb3`](https://github.com/apache/spark/commit/2d25fb38aab5bac1c4cdb7d36ed2ff75f7d16a57).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24066639
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
--- End diff --
Also, the key could be executorId, host, or rackId
---
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: [SPARK-4939] move to next locality when no pen...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-71553313
@tdas I noted on the JIRA, but since there is collateral effects of this change beyond local mode I think we decided not to put it in 1.2.1.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/3779
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72808109
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26732/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68314408
[Test build #24870 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/24870/consoleFull) for PR 3779 at commit [`32d363f`](https://github.com/apache/spark/commit/32d363f275c6dea225019a1cb23af4eeeb44781f).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72822525
[Test build #577 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/577/consoleFull) for PR 3779 at commit [`2d25fb3`](https://github.com/apache/spark/commit/2d25fb38aab5bac1c4cdb7d36ed2ff75f7d16a57).
* This patch **passes all tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70940007
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25918/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70939996
[Test build #25918 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25918/consoleFull) for PR 3779 at commit [`28d1b3c`](https://github.com/apache/spark/commit/28d1b3cbaf88a476dd9711921486708778c450c4).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70575375
@kayousterhout I had updated this PR to support all locality levels.
The timeout of locality depends on that the scheduler backend revive offers periodically, so we should still fix the LocalBackend, or it still get a chance to hang forever (because of the false-positive check in this PR).
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70348560
Why not just periodically revive offers, as we do in the cluster scheduler (see the 3 lines here: https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala#L83)?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70575997
Is this definitely still an issue with LocalBackend? It seems like LocalBackend should only ever have PROCESS_LOCAL and NODE_LOCAL tasks. If it only has NODE_LOCAL (no process local), then everything will work fine. If it has some PROCESS_LOCAL tasks (the case for this PR), then you don't need to revive offers, because everything gets updated correctly when the last PROCESS_LOCAL task gets scheduled, right?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72599896
cc scheduler czars also... @kayousterhout @markhamstra
---
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: [SPARK-4939] move to next locality when no pen...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72597256
I took a pass through this, but I'll admit that I'm a little out of my depth w.r.t. scheduler internals here. I think that I understand the goals / invariants of the scheduler at a high level, but I'm probably not comfortable enough with the internals + perf. optimizations to be comfortable signing off on this. (Aside: it seems like a lot of the complexity here comes from what seem like (necessary) perf. optimizations, such as not keeping per-task wait timers, using locality levels to index into parallel arrays, etc.).
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24066157
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
--- End diff --
id ==> executorId
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065817
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
+ case TaskLocality.RACK_LOCAL => hasMoreTasks(pendingTasksForRack)
+ }
+ if (!moreTasks) {
+ // Move to next locality level if there is no task for current level
+ lastLaunchTime = curTime
+ logDebug(s"No tasks for locality level ${myLocalityLevels(currentLocalityIndex)} " +
+ s"move to ${myLocalityLevels(currentLocalityIndex + 1)}")
+ currentLocalityIndex += 1
+ } else if (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex)) {
+ // Jump to the next locality level, and remove our waiting time for the current one since
+ // we don't want to count it again on the next one
+ lastLaunchTime += localityWaits(currentLocalityIndex)
+ currentLocalityIndex += 1
+ logDebug(s"Move to ${myLocalityLevels(currentLocalityIndex)} after wait for " +
+ s"${localityWaits(currentLocalityIndex)} ms")
+ } else {
+ return myLocalityLevels(currentLocalityIndex)
+ }
--- End diff --
It would also help to write a slightly longer / more informative comment in place of "It removes.." and "Remove the scheduled...". What about adding the following before noMoreTasksToRun:
"Walk through the list of tasks that can be scheduled at each location and returns true if there are any tasks that still need to be scheduled. Lazily cleans up tasks that have already been scheduled."
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72800486
Echoing the comments from @markhamstra and @JoshRosen, this change looks correct but is very subtle, so it would be great to improve the commenting throughout to avoid confusion for others who read this code later. I suggested a few comment improvements; if you have time to make these soon, that would be great since I know you wanted this in for 1.3 and I know the merge window for that is closing rapidly. Thanks @davies!
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68387930
[Test build #24903 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/24903/consoleFull) for PR 3779 at commit [`2d6ae73`](https://github.com/apache/spark/commit/2d6ae731debf10178e0675afa1d6f2a8b58edd0e).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72808101
[Test build #26732 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/26732/consoleFull) for PR 3779 at commit [`1550668`](https://github.com/apache/spark/commit/1550668b047f8702a6c7b19aad56cfd7a56e47c3).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065627
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
--- End diff --
The tasks with NO_PREF should not be in other lists, so do not need to remove them lazily, just check the whole list is empty or not.
I will add a comment.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-69666241
@pwendell Can you please take a look?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70582124
[Test build #25778 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25778/consoleFull) for PR 3779 at commit [`37a2804`](https://github.com/apache/spark/commit/37a28040750219a4e493d820bdb3e0feeb5398a4).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24066349
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
--- End diff --
pendingTasksWithNoPrefs.nonEmpty
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70582394
@kayousterhout I had added a test case with PROCESS_LOCAL and NO_PREF, it seems that LocalBackend can work well, because the TaskSchedulerImpl will try multiple locality levels in a row, so it will clear the false-positive state before try next level.
So, you are right, LocalBackend can work well with all cases I can image, without revive offers periodically.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70587346
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25778/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72598196
For any resource, it will try from the highest locality. TaskSetManager checks the current lowest locality level's wait is non-zero to determine the allowed lowest locality level. If it's zero, then TaskSetManager should assign the possible highest task for the resource, without any waiting. I think it make sense.
There are some logic are perf related, such as the index of task is removed from `list` lazily (only scanned once). Others (using locality levels to index into parallel arrays) are out of this PR's scope.
@rxin Could you also help to look at this PR?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by markhamstra <gi...@git.apache.org>.
Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24066137
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
--- End diff --
```scala
emptyKeys.foreach(executorId => pendingTasks.remove(executorId))
```
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23984210
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(list: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = list.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = list(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ list.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (key: String, list: ArrayBuffer[Int]) =>
--- End diff --
The `key` could be execId or host or rackId, so I change it to id.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23335344
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -495,13 +495,39 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // remove the emptyList from pendingTasks lazily
+ def hasNonEmptyList(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
--- End diff --
The for-loop will end once there is a list which is not empty, or it will remove all the empty list (not accessed in next time). So I think it will not be expensive even in a big cluster. We also could do it for tasks, then it will fix the false-positive issue.
In short term, this will fix the bug for streaming examples. In long term, this can reduce the latence for streaming (such as updateStateByKey or others needs union).
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70587474
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25774/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72812543
[Test build #578 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/578/consoleFull) for PR 3779 at commit [`2d25fb3`](https://github.com/apache/spark/commit/2d25fb38aab5bac1c4cdb7d36ed2ff75f7d16a57).
* This patch merges cleanly.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23338266
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -495,13 +495,39 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // remove the emptyList from pendingTasks lazily
+ def hasNonEmptyList(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
--- End diff --
If you have N executors, and they become empty one at a time, then this is a N^2 operation right (because you iterate over all of them -- not just the non-empty ones -- each time)?
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70586165
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25773/
Test FAILed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70587342
[Test build #25778 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/25778/consoleFull) for PR 3779 at commit [`37a2804`](https://github.com/apache/spark/commit/37a28040750219a4e493d820bdb3e0feeb5398a4).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r23338809
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -495,13 +495,39 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // remove the emptyList from pendingTasks lazily
+ def hasNonEmptyList(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
--- End diff --
For the first call, it will be N, then all the execId will be removed from pendingTasksForExecutor, the next call will be 0.
All the empty list will be access once, all the non-empty list, only one of them is accessed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68396667
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24903/
Test PASSed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24065965
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
+ false
+ }
+ }
+ emptyKeys.foreach(x => pendingTasks.remove(x))
+ hasTasks
+ }
+
+ while (currentLocalityIndex < myLocalityLevels.length - 1) {
+ val moreTasks = myLocalityLevels(currentLocalityIndex) match {
+ case TaskLocality.PROCESS_LOCAL => hasMoreTasks(pendingTasksForExecutor)
+ case TaskLocality.NODE_LOCAL => hasMoreTasks(pendingTasksForHost)
+ case TaskLocality.NO_PREF => pendingTasksWithNoPrefs.isEmpty
--- End diff --
Oh, yes, it's really a mistake! will fix 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: [SPARK-4939] move to next locality when no pen...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70955957
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25931/
Test PASSed.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-68387582
@pwendell I had added a regression test that fails in the old code, and added comments for them.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/3779#discussion_r24066617
--- Diff: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala ---
@@ -506,13 +506,59 @@ private[spark] class TaskSetManager(
* Get the level we can launch tasks according to delay scheduling, based on current wait time.
*/
private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
- while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
- currentLocalityIndex < myLocalityLevels.length - 1)
- {
- // Jump to the next locality level, and remove our waiting time for the current one since
- // we don't want to count it again on the next one
- lastLaunchTime += localityWaits(currentLocalityIndex)
- currentLocalityIndex += 1
+ // Remove the scheduled or finished tasks lazily
+ def hasNotScheduledTasks(taskIndexes: ArrayBuffer[Int]): Boolean = {
+ var indexOffset = taskIndexes.size
+ while (indexOffset > 0) {
+ indexOffset -= 1
+ val index = taskIndexes(indexOffset)
+ if (copiesRunning(index) == 0 && !successful(index)) {
+ return true
+ } else {
+ taskIndexes.remove(indexOffset)
+ }
+ }
+ false
+ }
+ // It removes the empty lists after check
+ def hasMoreTasks(pendingTasks: HashMap[String, ArrayBuffer[Int]]): Boolean = {
+ val emptyKeys = new ArrayBuffer[String]
+ val hasTasks = pendingTasks.exists{
+ case (id: String, tasks: ArrayBuffer[Int]) =>
+ if (hasNotScheduledTasks(tasks)) {
+ true
+ } else {
+ emptyKeys += id
--- End diff --
The `id` could be executorId, host or rackId.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by kayousterhout <gi...@git.apache.org>.
Github user kayousterhout commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-70349844
I thought local mode was just used for testing?
I general my preference is to avoid this kind of workaround, which add complexity to the scheduler, make it more brittle, harder to validate that it's working right, harder to maintain, harder to understand, etc. etc.
---
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: [SPARK-4939] move to next locality when no pen...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-72952510
I understood the high-level idea here and was just waiting for @kayousterhout and @markhamstra to review the details, so this seems fine to me. Thanks for reviewing 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 pull request: [SPARK-4939] move to next locality when no pen...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/3779#issuecomment-71000700
@kayousterhout @davies @pwendell can this be merged for 1.2.1?
---
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