Skip to content

[SPARK-15714][CORE] Fix flaky o.a.s.scheduler.BlacklistIntegrationSuite #13454

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
squito wants to merge 4 commits into from
Closed

[SPARK-15714][CORE] Fix flaky o.a.s.scheduler.BlacklistIntegrationSuite #13454

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
33b3739
do not check if results is empty on a failure -- instead just make su…
squito Jun 1, 2016
ccb6c20
taskScheduler.taskIdToTaskSetManager must be protected by taskSchedul…
squito Jun 1, 2016
6ef06c4
review feedback
squito Jun 3, 2016
08ab5d5
reword
squito Jun 3, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ private[spark] class TaskSchedulerImpl(
// on this class.
private val taskSetsByStageIdAndAttempt = new HashMap[Int, HashMap[Int, TaskSetManager]]

// Protected by `this`
private[scheduler] val taskIdToTaskSetManager = new HashMap[Long, TaskSetManager]
val taskIdToExecutorId = new HashMap[Long, String]

Expand Down
10 changes: 4 additions & 6 deletions core/src/test/scala/org/apache/spark/scheduler/BlacklistIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,12 @@ class BlacklistIntegrationSuite extends SchedulerIntegrationSuite[MultiExecutorM
* all tasks.
*/
def badHostBackend(): Unit = {
val task = backend.beginTask()
val host = backend.executorIdToExecutor(task.executorId).host
val (taskDescription, task) = backend.beginTask()
val host = backend.executorIdToExecutor(taskDescription.executorId).host
if (host == badHost) {
backend.taskFailed(task, new RuntimeException("I'm a bad host!"))
backend.taskFailed(taskDescription, new RuntimeException("I'm a bad host!"))
} else {
backend.taskSuccess(task, 42)
backend.taskSuccess(taskDescription, 42)
}
}

Expand All @@ -48,7 +48,6 @@ class BlacklistIntegrationSuite extends SchedulerIntegrationSuite[MultiExecutorM
val duration = Duration(1, SECONDS)
Await.ready(jobFuture, duration)
}
assert(results.isEmpty)
assertDataStructuresEmpty(noFailure = false)
}

Expand All @@ -68,7 +67,6 @@ class BlacklistIntegrationSuite extends SchedulerIntegrationSuite[MultiExecutorM
val duration = Duration(3, SECONDS)
Await.ready(jobFuture, duration)
}
assert(results.isEmpty)
assertDataStructuresEmpty(noFailure = false)
}

Expand Down
47 changes: 28 additions & 19 deletions core/src/test/scala/org/apache/spark/scheduler/SchedulerIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,11 @@ abstract class SchedulerIntegrationSuite[T <: MockBackend: ClassTag] extends Spa
// when the job succeeds
assert(taskScheduler.runningTaskSets.isEmpty)
assert(!backend.hasTasks)
} else {
// Note that we CANNOT check for empty results on a failure -- the resultHandler will

@vanzin vanzin Jun 2, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's hard to correlate this comment with the code around it. It seems like it belongs in the call site, where the results list was actually being checked (and you removed the assert). But without the assert, the comment would be out of place there too.

I just find it very improbable that someone will read this comment when writing a test. It would help if, instead, the expected results were passed to this method, so that the asserts happen here instead of in the body of every test. Then the comment is fine where it is, because this method would handle the asserts (or lack thereof).

Also, the same assert exists in the "job failure after 4 attempts" test in SchedulerIntegrationSuite. Doesn't that need to change too?

@squito squito Jun 3, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I agree this is a little weird, its not really going to help somebody writing a test, only if somebody looks at the entire framework here a little more closely. I thought of passing in expected results or expected failure, but maybe in some test you don't want to check the expectation exactly. I guess somebody writing a test at least needs to understand how to use assertDataStructuresEmpty, results, and failure. I'll put comments there instead, someone writing a test is at least more likely to read those comments.

I'm not totally sure I know what you mean about "job failure after 4 attempts" -- I did remove the assert(results.isEmpty) there. Actually, that assert is fine (all tasks fail, so there are never any results), but I thought it would be better to remove it in case someone copies that code into another test without thinking about it.

@vanzin vanzin Jun 3, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"job failure after 4 attempts" -- I did remove the assert(results.isEmpty)

Ah, my bad, I swear I searched for it in the diff and the browser didn't find it.

// record results from successful tasks, even if the job fails overall. We just check
// that we got a failure.
assert(failure != null)
}
assert(scheduler.activeJobs.isEmpty)
}
Expand Down Expand Up @@ -217,10 +222,10 @@ private[spark] abstract class MockBackend(
* Test backends should call this to get a task that has been assigned to them by the scheduler.
* Each task should be responded to with either [[taskSuccess]] or [[taskFailed]].
*/
def beginTask(): TaskDescription = {
def beginTask(): (TaskDescription, Task[_]) = {
synchronized {
val toRun = assignedTasksWaitingToRun.remove(assignedTasksWaitingToRun.size - 1)
runningTasks += toRun
runningTasks += toRun._1.taskId
toRun
}
}
Expand Down Expand Up @@ -255,7 +260,7 @@ private[spark] abstract class MockBackend(
taskScheduler.statusUpdate(task.taskId, state, resultBytes)
if (TaskState.isFinished(state)) {
synchronized {
runningTasks -= task
runningTasks -= task.taskId
executorIdToExecutor(task.executorId).freeCores += taskScheduler.CPUS_PER_TASK
freeCores += taskScheduler.CPUS_PER_TASK
}
Expand All @@ -264,9 +269,9 @@ private[spark] abstract class MockBackend(
}

// protected by this
private val assignedTasksWaitingToRun = new ArrayBuffer[TaskDescription](10000)
private val assignedTasksWaitingToRun = new ArrayBuffer[(TaskDescription, Task[_])](10000)
// protected by this
private val runningTasks = ArrayBuffer[TaskDescription]()
private val runningTasks = HashSet[Long]()

def hasTasks: Boolean = synchronized {
assignedTasksWaitingToRun.nonEmpty || runningTasks.nonEmpty
Expand Down Expand Up @@ -307,10 +312,19 @@ private[spark] abstract class MockBackend(
*/
override def reviveOffers(): Unit = {
val offers: Seq[WorkerOffer] = generateOffers()
val newTasks = taskScheduler.resourceOffers(offers).flatten
val newTaskDescriptions = taskScheduler.resourceOffers(offers).flatten
// get the task now, since that requires a lock on TaskSchedulerImpl, to prevent individual
// tests for introducing a race if they need it

@vanzin vanzin Jun 2, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s/for/from

val newTasks = taskScheduler.synchronized {
newTaskDescriptions.map { taskDescription =>
val taskSet = taskScheduler.taskIdToTaskSetManager(taskDescription.taskId).taskSet
val task = taskSet.tasks(taskDescription.index)
(taskDescription, task)
}
}
synchronized {
newTasks.foreach { task =>
executorIdToExecutor(task.executorId).freeCores -= taskScheduler.CPUS_PER_TASK
newTasks.foreach { case (taskDescription, task) =>
executorIdToExecutor(taskDescription.executorId).freeCores -= taskScheduler.CPUS_PER_TASK
}
freeCores -= newTasks.size * taskScheduler.CPUS_PER_TASK
assignedTasksWaitingToRun ++= newTasks
Expand Down Expand Up @@ -437,8 +451,8 @@ class BasicSchedulerIntegrationSuite extends SchedulerIntegrationSuite[SingleCor
*/
testScheduler("super simple job") {
def runBackend(): Unit = {
val task = backend.beginTask()
backend.taskSuccess(task, 42)
val (taskDescripition, task) = backend.beginTask()

@vanzin vanzin Jun 2, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

task is unused? (Could be just _.) Also happens in other places.

backend.taskSuccess(taskDescripition, 42)
}
withBackend(runBackend _) {
val jobFuture = submit(new MockRDD(sc, 10, Nil), (0 until 10).toArray)
Expand Down Expand Up @@ -473,9 +487,7 @@ class BasicSchedulerIntegrationSuite extends SchedulerIntegrationSuite[SingleCor
val d = join(30, b, c)

def runBackend(): Unit = {
val taskDescription = backend.beginTask()
val taskSet = taskScheduler.taskIdToTaskSetManager(taskDescription.taskId).taskSet
val task = taskSet.tasks(taskDescription.index)
val (taskDescription, task) = backend.beginTask()

// make sure the required map output is available
task.stageId match {
Expand Down Expand Up @@ -515,9 +527,7 @@ class BasicSchedulerIntegrationSuite extends SchedulerIntegrationSuite[SingleCor
val stageToAttempts = new HashMap[Int, HashSet[Int]]()

def runBackend(): Unit = {
val taskDescription = backend.beginTask()
val taskSet = taskScheduler.taskIdToTaskSetManager(taskDescription.taskId).taskSet
val task = taskSet.tasks(taskDescription.index)
val (taskDescription, task) = backend.beginTask()
stageToAttempts.getOrElseUpdate(task.stageId, new HashSet()) += task.stageAttemptId

// make sure the required map output is available
Expand Down Expand Up @@ -549,16 +559,15 @@ class BasicSchedulerIntegrationSuite extends SchedulerIntegrationSuite[SingleCor

testScheduler("job failure after 4 attempts") {
def runBackend(): Unit = {
val task = backend.beginTask()
backend.taskFailed(task, new RuntimeException("test task failure"))
val (taskDescription, task) = backend.beginTask()
backend.taskFailed(taskDescription, new RuntimeException("test task failure"))
}
withBackend(runBackend _) {
val jobFuture = submit(new MockRDD(sc, 10, Nil), (0 until 10).toArray)
val duration = Duration(1, SECONDS)
Await.ready(jobFuture, duration)
failure.getMessage.contains("test task failure")
}
assert(results.isEmpty)
assertDataStructuresEmpty(noFailure = false)
}
}