[SPARK-4454] Properly synchronize accesses to DAGScheduler cacheLocs map

[JoshRosen]

This patch addresses a race condition in DAGScheduler by properly synchronizing accesses to its cacheLocs map.

This map is accessed by the getCacheLocs and clearCacheLocs() methods, which can be called by separate threads, since DAGScheduler's getPreferredLocs() method is called by SparkContext and indirectly calls getCacheLocs(). If this map is cleared by the DAGScheduler event processing thread while a user thread is submitting a job and computing preferred locations, then this can cause the user thread to throw "NoSuchElementException: key not found" errors.

Most accesses to DAGScheduler's internal state do not need synchronization because that state is only accessed from the event processing loop's thread. An alternative approach to fixing this bug would be to refactor this code so that SparkContext sends the DAGScheduler a message in order to get the list of preferred locations. However, this would involve more extensive changes to this code and would be significantly harder to backport to maintenance branches since some of the related code has undergone significant refactoring (e.g. the introduction of EventLoop). Since cacheLocs is the only state that's accessed in this way, adding simple synchronization seems like a better short-term fix.

See #3345 for additional context.

[SparkQA]

Test build #27651 has started for PR 4660 at commit 12d64ba.

• This patch merges cleanly.

[JoshRosen]

/cc @pwendell @markhamstra

[SparkQA]

Test build #27651 has finished for PR 4660 at commit 12d64ba.

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

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/27651/
Test PASSed.

[pwendell]

This LGTM Josh - thanks for giving this such a thorough look.

[JoshRosen]

I pushed a a51fc7e hotfix to roll back a bit of code cleanup that I did which added a getOrElseUpdate call, since this might have a performance impact as this code is called O(num tasks) times.