Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/21758#discussion_r204236759
--- Diff:
core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala ---
@@ -359,20 +366,55 @@ private[spark] class TaskSchedulerImpl(
// of locality levels so that it gets a chance to launch local tasks
on all of them.
// NOTE: the preferredLocality order: PROCESS_LOCAL, NODE_LOCAL,
NO_PREF, RACK_LOCAL, ANY
for (taskSet <- sortedTaskSets) {
- var launchedAnyTask = false
- var launchedTaskAtCurrentMaxLocality = false
- for (currentMaxLocality <- taskSet.myLocalityLevels) {
- do {
- launchedTaskAtCurrentMaxLocality = resourceOfferSingleTaskSet(
- taskSet, currentMaxLocality, shuffledOffers, availableCpus,
tasks)
- launchedAnyTask |= launchedTaskAtCurrentMaxLocality
- } while (launchedTaskAtCurrentMaxLocality)
- }
- if (!launchedAnyTask) {
- taskSet.abortIfCompletelyBlacklisted(hostToExecutors)
+ // Skip the barrier taskSet if the available slots are less than the
number of pending tasks.
+ if (taskSet.isBarrier && availableSlots < taskSet.numTasks) {
+ // Skip the launch process.
+ // TODO SPARK-24819 If the job requires more slots than available
(both busy and free
+ // slots), fail the job on submit.
+ logInfo(s"Skip current round of resource offers for barrier stage
${taskSet.stageId} " +
+ s"because the barrier taskSet requires ${taskSet.numTasks}
slots, while the total " +
+ s"number of available slots is ${availableSlots}.")
+ } else {
+ var launchedAnyTask = false
+ var launchedTaskAtCurrentMaxLocality = false
--- End diff --
nit: can we move this line before line 383 to reduce its scope?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
