ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Revert "[SPARK-13279] Remove O(n^2) operation from scheduler."

Browse source 
This reverts commit 50fa6fd1b3.
This commit is contained in:
Reynold Xin 2016-02-11 13:31:13 -08:00
parent 50fa6fd1b3
commit c86009ceb9
1 changed files with 6 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
View file

@ -114,14 +114,9 @@ private[spark] class TaskSetManager(
// treated as stacks, in which new tasks are added to the end of the
// ArrayBuffer and removed from the end. This makes it faster to detect
// tasks that repeatedly fail because whenever a task failed, it is put
// back at the head of the stack. These collections may contain duplicates
// for two reasons:
// (1): Tasks are only removed lazily; when a task is launched, it remains
// in all the pending lists except the one that it was launched from.
// (2): Tasks may be re-added to these lists multiple times as a result
// of failures.
// Duplicates are handled in dequeueTaskFromList, which ensures that a
// task hasn't already started running before launching it.
// back at the head of the stack. They are also only cleaned up lazily;
// when a task is launched, it remains in all the pending lists except
// the one that it was launched from, but gets removed from them later.
private val pendingTasksForExecutor = new HashMap[String, ArrayBuffer[Int]]
// Set of pending tasks for each host. Similar to pendingTasksForExecutor,
@ -186,7 +181,9 @@ private[spark] class TaskSetManager(
private def addPendingTask(index: Int) {
// Utility method that adds `index` to a list only if it's not already there
def addTo(list: ArrayBuffer[Int]) {
list += index
if (!list.contains(index)) {
list += index
}
}
for (loc <- tasks(index).preferredLocations) {