Move waitBackendReady from DAGScheduler.submitStage to TaskSchedulerImpl.submitTasks

Author: li-zhihui

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -710,7 +710,6 @@ class DAGScheduler(
   /** Submits stage, but first recursively submits any missing parents. */
   private def submitStage(stage: Stage) {
     val jobId = activeJobForStage(stage)
-    taskScheduler.waitBackendReady
     if (jobId.isDefined) {
       logDebug("submitStage(" + stage + ")")
       if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {

core/src/main/scala/org/apache/spark/scheduler/TaskScheduler.scala

@@ -54,5 +54,4 @@ private[spark] trait TaskScheduler {
 
   // Get the default level of parallelism to use in the cluster, as a hint for sizing jobs.
   def defaultParallelism(): Int
-  def waitBackendReady(): Unit = { return }
 }

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

@@ -147,6 +147,7 @@ private[spark] class TaskSchedulerImpl(
 
   override def submitTasks(taskSet: TaskSet) {
     val tasks = taskSet.tasks
+    waitBackendReady
     logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
     this.synchronized {
       val manager = new TaskSetManager(this, taskSet, maxTaskFailures)
@@ -432,7 +433,7 @@ private[spark] class TaskSchedulerImpl(
   // By default, rack is unknown
   def getRackForHost(value: String): Option[String] = None
 
-  override def waitBackendReady(): Unit = {
+  private def waitBackendReady(): Unit = {
     if (backend.isReady) {
       return
     }

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -50,9 +50,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: A
   val conf = scheduler.sc.conf
   private val timeout = AkkaUtils.askTimeout(conf)
   private val akkaFrameSize = AkkaUtils.maxFrameSizeBytes(conf)
-  // Submit stage only after (registered executors / total executors) arrived the ratio.
+  // Submit tasks only after (registered executors / total executors) arrived the ratio.
   var minRegisteredRatio = conf.getDouble("spark.scheduler.minRegisteredExecutorsRatio", 0)
-  // Whatever minRegisteredExecutorsRatio is arrived, submit stage after the time(milliseconds).
+  // Whatever minRegisteredExecutorsRatio is arrived, submit tasks after the time(milliseconds).
   val maxRegisteredWaitingTime =
     conf.getInt("spark.scheduler.maxRegisteredExecutorsWaitingTime", 30000)
   val createTime = System.currentTimeMillis()