Ngone51 opened a new pull request #28260: [SPARK-31487][CORE] Move slots check of barrier job from DAGScheduler to TaskSchedulerImpl URL: https://github.com/apache/spark/pull/28260 <!-- Thanks for sending a pull request! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html 2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html 3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'. 4. Be sure to keep the PR description updated to reflect all changes. 5. Please write your PR title to summarize what this PR proposes. 6. If possible, provide a concise example to reproduce the issue for a faster review. 7. If you want to add a new configuration, please read the guideline first for naming configurations in 'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'. --> ### What changes were proposed in this pull request? <!-- Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below. 1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers. 2. If you fix some SQL features, you can provide some references of other DBMSes. 3. If there is design documentation, please add the link. 4. If there is a discussion in the mailing list, please add the link. --> Move slots check of barrier job from `DAGScheduler` to `TaskSchedulerImpl` and add new configuration `spark.scheduler.barrier.waitForEnoughSlots.timeout` to replace with `spark.scheduler.barrier.maxConcurrentTasksCheck.interval`/ `spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures`. ### Why are the changes needed? <!-- Please clarify why the changes are needed. For instance, 1. If you propose a new API, clarify the use case for a new API. 2. If you fix a bug, you can clarify why it is a bug. --> We do the slots check for a barrier job to ensure the job can be launched finally. We'll continuously resubmit the same job until we pass the check(find enough slots after some executors registered at driver). Thus, it introduces duplicate DAG planning for the same job, which may low down the throughput of the `DAGSchedule`. And it seems DAG planning could always be faster than executor launching, which makes duplicate DAG planning inevitable. Besides, we actually already did the similar slots check inside `TaskSchedulerImpl`. We can combine them as well. One obvious benefit we can get after this improvement is reducing the time of barrier tests, e.g. Before: ``` [info] BarrierTaskContextSuite: [info] - global sync by barrier() call (18 seconds, 972 milliseconds) [info] - share messages with allGather() call (17 seconds, 95 milliseconds) [info] - throw exception if we attempt to synchronize with different blocking calls (16 seconds, 222 milliseconds) [info] - successively sync with allGather and barrier (17 seconds, 532 milliseconds) [info] - support multiple barrier() call within a single task (17 seconds, 792 milliseconds) [info] - throw exception on barrier() call timeout (17 seconds, 85 milliseconds) [info] - throw exception if barrier() call doesn't happen on every task (17 seconds, 166 milliseconds) [info] - throw exception if the number of barrier() calls are not the same on every task (17 seconds, 149 milliseconds) [info] - barrier task killed, no interrupt (16 seconds, 777 milliseconds) [info] - barrier task killed, interrupt (16 seconds, 208 milliseconds) [info] ScalaTest [info] Run completed in 2 minutes, 54 seconds. ``` After: ``` [info] BarrierTaskContextSuite: [info] - global sync by barrier() call (6 seconds, 694 milliseconds) [info] - share messages with allGather() call (5 seconds, 885 milliseconds) [info] - throw exception if we attempt to synchronize with different blocking calls (5 seconds, 384 milliseconds) [info] - successively sync with allGather and barrier (6 seconds, 973 milliseconds) [info] - support multiple barrier() call within a single task (6 seconds, 979 milliseconds) [info] - throw exception on barrier() call timeout (6 seconds, 158 milliseconds) [info] - throw exception if barrier() call doesn't happen on every task (6 seconds, 211 milliseconds) [info] - throw exception if the number of barrier() calls are not the same on every task (6 seconds, 171 milliseconds) [info] - barrier task killed, no interrupt (4 seconds, 776 milliseconds) [info] - barrier task killed, interrupt (5 seconds, 190 milliseconds) [info] ScalaTest [info] Run completed in 1 minute, 3 seconds. ``` ### Does this PR introduce any user-facing change? <!-- If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible. If no, write 'No'. --> Yes, user can not use `spark.scheduler.barrier.maxConcurrentTasksCheck.interval`/ `spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures` any more since Spark 3.0 but should use `spark.scheduler.barrier.waitForEnoughSlots.timeout` instead. ### How was this patch tested? <!-- If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible. If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future. If tests were not added, please describe why they were not added and/or why it was difficult to add. --> Updated existed tests.
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
