diff options
| author | root <root@ip-10-232-19-31.ec2.internal> | 2013-07-14 06:24:29 +0000 |
|---|---|---|
| committer | root <root@ip-10-232-19-31.ec2.internal> | 2013-07-14 06:24:29 +0000 |
| commit | e271fde10b342216e33f4f45af73c5d103215cf2 (patch) | |
| tree | c432c276b2040a8034c978af01e104a6574d968f /core | |
| parent | 668b0dc6a7fac85cde93d96adf966bbf98275018 (diff) | |
| download | spark-e271fde10b342216e33f4f45af73c5d103215cf2.tar.gz spark-e271fde10b342216e33f4f45af73c5d103215cf2.tar.bz2 spark-e271fde10b342216e33f4f45af73c5d103215cf2.zip | |
Fixed a delay scheduling bug in the YARN branch, found by Patrick
Diffstat (limited to 'core')
| -rw-r--r-- | core/src/main/scala/spark/scheduler/cluster/ClusterTaskSetManager.scala | 10 |
1 files changed, 5 insertions, 5 deletions
diff --git a/core/src/main/scala/spark/scheduler/cluster/ClusterTaskSetManager.scala b/core/src/main/scala/spark/scheduler/cluster/ClusterTaskSetManager.scala index fe6420a522..327d6797ae 100644 --- a/core/src/main/scala/spark/scheduler/cluster/ClusterTaskSetManager.scala +++ b/core/src/main/scala/spark/scheduler/cluster/ClusterTaskSetManager.scala @@ -454,10 +454,10 @@ private[spark] class ClusterTaskSetManager( val taskId = sched.newTaskId() // Figure out whether this should count as a preferred launch val taskLocality = - if (isProcessLocalLocation(task, hostPort)) TaskLocality.PROCESS_LOCAL else - if (isHostLocalLocation(task, hostPort)) TaskLocality.NODE_LOCAL else - if (isRackLocalLocation(task, hostPort)) TaskLocality.RACK_LOCAL else - TaskLocality.ANY + if (isProcessLocalLocation(task, hostPort)) TaskLocality.PROCESS_LOCAL + else if (isHostLocalLocation(task, hostPort)) TaskLocality.NODE_LOCAL + else if (isRackLocalLocation(task, hostPort)) TaskLocality.RACK_LOCAL + else TaskLocality.ANY val prefStr = taskLocality.toString logInfo("Starting task %s:%d as TID %s on slave %s: %s (%s)".format( taskSet.id, index, taskId, execId, hostPort, prefStr)) @@ -467,7 +467,7 @@ private[spark] class ClusterTaskSetManager( val info = new TaskInfo(taskId, index, time, execId, hostPort, taskLocality) taskInfos(taskId) = info taskAttempts(index) = info :: taskAttempts(index) - if (TaskLocality.NODE_LOCAL == taskLocality) { + if (taskLocality == TaskLocality.PROCESS_LOCAL || taskLocality == TaskLocality.NODE_LOCAL) { lastPreferredLaunchTime = time } // Serialize and return the task |