Skip to content

Commit

Permalink
fix code style
Browse files Browse the repository at this point in the history
  • Loading branch information
Rui Li committed Jun 23, 2014
1 parent fafd57f commit 8444d7c
Show file tree
Hide file tree
Showing 2 changed files with 7 additions and 7 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -724,11 +724,11 @@ private[spark] class TaskSetManager(
import TaskLocality.{PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL, ANY}
val levels = new ArrayBuffer[TaskLocality.TaskLocality]
if (!pendingTasksForExecutor.isEmpty && getLocalityWait(PROCESS_LOCAL) != 0 &&
pendingTasksForExecutor.keySet.exists(sched.isExecutorAlive(_))) {
pendingTasksForExecutor.keySet.exists(sched.isExecutorAlive(_))) {
levels += PROCESS_LOCAL
}
if (!pendingTasksForHost.isEmpty && getLocalityWait(NODE_LOCAL) != 0 &&
pendingTasksForHost.keySet.exists(sched.hasExecutorsAliveOnHost(_))) {
pendingTasksForHost.keySet.exists(sched.hasExecutorsAliveOnHost(_))) {
levels += NODE_LOCAL
}
if (!pendingTasksForRack.isEmpty && getLocalityWait(RACK_LOCAL) != 0) {
Expand All @@ -744,7 +744,7 @@ private[spark] class TaskSetManager(
def newLocAvail(index: Int): Boolean = {
for (loc <- tasks(index).preferredLocations) {
if (sched.hasExecutorsAliveOnHost(loc.host) ||
sched.getRackForHost(loc.host).isDefined) {
sched.getRackForHost(loc.host).isDefined) {
return true
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,8 @@ class FakeTaskScheduler(sc: SparkContext, liveExecutors: (String, String)* /* ex

override def hasExecutorsAliveOnHost(host: String): Boolean = executors.values.exists(_ == host)

def addExecutor(newExecutors: (String, String)*) {
executors ++= newExecutors
def addExecutor(execId: String, host: String) {
executors.put(execId, host)
}
}

Expand Down Expand Up @@ -403,14 +403,14 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
// Only ANY is valid
assert(manager.myLocalityLevels.sameElements(Array(ANY)))
// Add a new executor
sched.addExecutor(("execD", "host1"))
sched.addExecutor("execD", "host1")
manager.executorAdded()
// Task 0 and 1 should be removed from no-pref list
assert(manager.pendingTasksWithNoPrefs.size === 2)
// Valid locality should contain NODE_LOCAL and ANY
assert(manager.myLocalityLevels.sameElements(Array(NODE_LOCAL, ANY)))
// Add another executor
sched.addExecutor(("execC", "host2"))
sched.addExecutor("execC", "host2")
manager.executorAdded()
// No-pref list now only contains task 3
assert(manager.pendingTasksWithNoPrefs.size === 1)
Expand Down

0 comments on commit 8444d7c

Please sign in to comment.