[SPARK-17304] Fix perf. issue caused by TaskSetManager.abortIfCompletelyBlacklisted

[JoshRosen]

This patch addresses a minor scheduler performance issue that was introduced in #13603. If you run

sc.parallelize(1 to 100000, 100000).map(identity).count()

then most of the time ends up being spent in TaskSetManager.abortIfCompletelyBlacklisted():

When processing resource offers, the scheduler uses a nested loop which considers every task set at multiple locality levels:

   for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
      do {
        launchedTask = resourceOfferSingleTaskSet(
            taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
      } while (launchedTask)
    }

In order to prevent jobs with globally blacklisted tasks from hanging, #13603 added a taskSet.abortIfCompletelyBlacklisted call inside of resourceOfferSingleTaskSet; if a call to resourceOfferSingleTaskSet fails to schedule any tasks, then abortIfCompletelyBlacklisted checks whether the tasks are completely blacklisted in order to figure out whether they will ever be schedulable. The problem with this placement of the call is that the last call to resourceOfferSingleTaskSet in the while loop will return false, implying that resourceOfferSingleTaskSet will call abortIfCompletelyBlacklisted, so almost every call to resourceOffers will trigger the abortIfCompletelyBlacklisted check for every task set.

Instead, I think that this call should be moved out of the innermost loop and should be called at most once per task set in case none of the task set's tasks can be scheduled at any locality level.

Before this patch's changes, the microbenchmark example that I posted above took 35 seconds to run, but it now only takes 15 seconds after this change.

/cc @squito and @kayousterhout for review.

[kayousterhout]

I know it's super verbose, but to minimize how confusing this code is, what about calling this launchedTaskAtCurrentMaxLocality, and renaming maxLocality below to currentMaxLocality?

[JoshRosen]

Good idea; updated.

[SparkQA]

Test build #64609 has finished for PR 14871 at commit 5d20b44.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #64610 has finished for PR 14871 at commit 321d0c6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[kayousterhout]

LGTM

Josh, how long does your microbenchmark take if you comment out the call to abortIfCompletelyBlacklisted? Wondering how much that continues to affect performance.

[JoshRosen]

@kayousterhout, after this patch's changes commenting out abortIfCompletelyBlacklisted appears to provide no further performance improvement. However, I have identified a few bottlenecks in Accumulator serialization that can be used to obtain an additional ~30% speedup on this microbenchmark; I'll submit those in a separate patch.

[kayousterhout]

Ok cool thanks @JoshRosen -- just wanted to make sure that the blacklisting wasn't still hurting performance. Thanks for fixing this and sorry about the oversight originally!

[JoshRosen]

Merging this to master. Thanks!

[squito]

thanks for finding this and the quick fix @JoshRosen !