[SPARK-23626][CORE] DAGScheduler blocked due to JobSubmitted event

[ajithme]

What changes were proposed in this pull request?

DAGScheduler becomes a bottleneck in cluster when multiple JobSubmitted events has to be processed as DAGSchedulerEventProcessLoop is single threaded and it will block other tasks in queue like TaskCompletion.
The JobSubmitted event is time consuming depending on the nature of the job (Example: calculating parent stage dependencies, shuffle dependencies, partitions) and thus it blocks all the events to be processed.

Similarly in my cluster some jobs partition calculation is time consuming (Similar to stack at SPARK-2647) hence it slows down the spark DAGSchedulerEventProcessLoop which results in user jobs to slowdown, even if its tasks are finished within seconds, as TaskCompletion Events are processed at a slower rate due to blockage.

Move the ResultStage creation to call site thread, which will avoid blocking of DAGScheduler thread for other events

Refer: http://apache-spark-developers-list.1001551.n3.nabble.com/Spark-Scheduler-Spark-DAGScheduler-scheduling-performance-hindered-on-JobSubmitted-Event-td23562.html

How was this patch tested?

Manual test to verify blockage before and after applying patch.

[ajithme]

For UT, i see DAGSchedulerSuite currently does not have same behaviour as DAGScheduler as its events posted to DAGSchedulerEventProcessLoopTester are running in same thread unlike DAGScheduler where posted events are processed in separate thread . Can i modify this.? or write a separate suite.? Pls suggest

[Ngone51]

Do we really need TrieMap for concurrence purpose, since job schedule is FIFO(even if after move createResultStage() ) ?

BTW, out of my curiosity, what's TrieMap advantages compare to ConcurrentHashMap ？

[shivaram]

cc @kayousterhout @markhamstra

[shivaram]

@AjithShetty2489 I'm not sure just changing these two maps is sufficient ? For example createResultStage could in turn create all the parent stages and the parents stages could be ShuffleMapStage which in turn means that the map shuffleIdToMapStage would need to be protected and we'll also need to ensure that the MapOutputTracker registerShuffle is thread safe

[markhamstra]

@squito is the master of DAGSchedulerSuite, and can provide you the best advice on changing or adding to the existing DAGSchedulerSuite. I'll be back from skiing next week and try to find some time to look at this. Hopefully @kayousterhout can find some time too.

[ajithme]

@Ngone51 Thanks for the question. I am not quite sure of actual implementation of Triemap but i see
https://www.scala-lang.org/api/2.12.3/scala/collection/concurrent/TrieMap.html
https://stackoverflow.com/questions/45566633/whats-the-difference-between-scala-triemap-and-java-concurrenthashmap
https://stackoverflow.com/questions/29499381/what-is-a-triemap-and-what-is-its-advantages-disadvantages-compared-to-a-hashmap and conclude trieMap is better for concurrent scenarios.

but i measure the performance of ConcurrentHashMap vs TrieMap with basic put and get operation, it seems TrieMap is slower than ConcurrentHashmap

[ajithme]

@shivaram yes, you are right. Let me recheck

[squito]

took a quick look, agree with shivaram's observations, you've got to handle shuffleIdToMapStage which will not be so easy.

[AmplabJenkins]

Can one of the admins verify this patch?

[HyukjinKwon]

@ajithme so have you got some time to recheck?

[ajithme]

got caught up with something, will relook into this

[ajithme]

Closing this as attempted a new approach, please check #24438