Github user Ngone51 commented on a diff in the pull request:

    https://github.com/apache/spark/pull/22288#discussion_r214720097
  
    --- Diff: 
core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala ---
    @@ -414,9 +425,54 @@ private[spark] class TaskSchedulerImpl(
                 launchedAnyTask |= launchedTaskAtCurrentMaxLocality
               } while (launchedTaskAtCurrentMaxLocality)
             }
    +
             if (!launchedAnyTask) {
    -          taskSet.abortIfCompletelyBlacklisted(hostToExecutors)
    -        }
    +          taskSet.getCompletelyBlacklistedTaskIfAny(hostToExecutors) match 
{
    +            case taskIndex: Some[Int] => // Returns the taskIndex which 
was unschedulable
    +              if (conf.getBoolean("spark.dynamicAllocation.enabled", 
false)) {
    +                // If the taskSet is unschedulable we kill the existing 
blacklisted executor/s and
    +                // kick off an abortTimer which after waiting will abort 
the taskSet if we were
    +                // unable to get new executors and couldn't schedule a 
task from the taskSet.
    +                // Note: We keep a track of schedulability on a per 
taskSet basis rather than on a
    +                // per task basis.
    +                if (!unschedulableTaskSetToExpiryTime.contains(taskSet)) {
    +                  hostToExecutors.valuesIterator.foreach(executors => 
executors.foreach({
    +                    executor =>
    +                      logDebug("Killing executor because of task 
unschedulability: " + executor)
    +                      blacklistTrackerOpt.foreach(blt => 
blt.killBlacklistedExecutor(executor))
    +                  })
    +                  )
    +                  unschedulableTaskSetToExpiryTime(taskSet) = 
clock.getTimeMillis()
    +                  abortTimer.schedule(new TimerTask() {
    +                    override def run() {
    +                      if 
(unschedulableTaskSetToExpiryTime.contains(taskSet) &&
    +                        (unschedulableTaskSetToExpiryTime(taskSet)
    +                          + UNSCHEDULABLE_TASKSET_TIMEOUT_MS)
    +                          <= clock.getTimeMillis()
    +                      ) {
    +                        logInfo("Cannot schedule any task because of 
complete blacklisting. " +
    +                          "Wait time for scheduling expired. Aborting the 
application.")
    +                        
taskSet.abortSinceCompletelyBlacklisted(taskIndex.get)
    +                      } else {
    +                        this.cancel()
    +                      }
    +                    }
    +                  }, UNSCHEDULABLE_TASKSET_TIMEOUT_MS)
    +                }
    +              } else {
    +                // TODO: try acquiring new executors for static allocation 
before aborting.
    --- End diff --
    
    How ? Waiting for other tasks finish and release resources ?


---

---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to