Move waitBackendReady to TaskSchedulerImpl.start & some code refactor

Author: li-zhihui

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

@@ -143,11 +143,11 @@ private[spark] class TaskSchedulerImpl(
         Utils.tryOrExit { checkSpeculatableTasks() }
       }
     }
+    waitBackendReady
   }
 
   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)

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

@@ -46,11 +46,12 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: A
 {
   // Use an atomic variable to track total number of cores in the cluster for simplicity and speed
   var totalCoreCount = new AtomicInteger(0)
-  var totalExecutors = new AtomicInteger(0)
+  var totalExpectedExecutors = new AtomicInteger(0)
   val conf = scheduler.sc.conf
   private val timeout = AkkaUtils.askTimeout(conf)
   private val akkaFrameSize = AkkaUtils.maxFrameSizeBytes(conf)
-  // Submit tasks only after (registered executors / total executors) arrived the ratio.
+  // Submit tasks only after (registered executors / total expected executors) 
+  // is equal to at least this value.
   var minRegisteredRatio = conf.getDouble("spark.scheduler.minRegisteredExecutorsRatio", 0)
   // Whatever minRegisteredExecutorsRatio is arrived, submit tasks after the time(milliseconds).
   val maxRegisteredWaitingTime =
@@ -91,7 +92,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: A
           executorAddress(executorId) = sender.path.address
           addressToExecutorId(sender.path.address) = executorId
           totalCoreCount.addAndGet(cores)
-          if (executorActor.size >= totalExecutors.get() * minRegisteredRatio) {
+          if (executorActor.size >= totalExpectedExecutors.get() * minRegisteredRatio) {
             ready = true
           }
           makeOffers()

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

@@ -95,7 +95,7 @@ private[spark] class SparkDeploySchedulerBackend(
 
   override def executorAdded(fullId: String, workerId: String, hostPort: String, cores: Int,
     memory: Int) {
-    totalExecutors.addAndGet(1)
+    totalExpectedExecutors.addAndGet(1)
     logInfo("Granted executor ID %s on hostPort %s with %d cores, %s RAM".format(
       fullId, hostPort, cores, Utils.megabytesToString(memory)))
   }

yarn/common/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala

@@ -77,7 +77,7 @@ private[spark] class YarnClientSchedulerBackend(
 
     logDebug("ClientArguments called with: " + argsArrayBuf)
     val args = new ClientArguments(argsArrayBuf.toArray, conf)
-    totalExecutors.set(args.numExecutors)
+    totalExpectedExecutors.set(args.numExecutors)
     client = new Client(args, conf)
     appId = client.runApp()
     waitForApp()

yarn/common/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala

@@ -17,34 +17,23 @@
 
 package org.apache.spark.scheduler.cluster
 
-import scala.collection.mutable.ArrayBuffer
-
-import org.apache.spark.{Logging, SparkContext}
-import org.apache.spark.deploy.yarn.ApplicationMasterArguments
+import org.apache.spark.SparkContext
 import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.util.IntParam
 
 private[spark] class YarnClusterSchedulerBackend(
     scheduler: TaskSchedulerImpl,
     sc: SparkContext)
-  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.actorSystem)
-  with Logging {
-
-  private def addArg(optionName: String, envVar: String, sysProp: String,
-      arrayBuf: ArrayBuffer[String]) {
-    if (System.getenv(envVar) != null) {
-      arrayBuf += (optionName, System.getenv(envVar))
-    } else if (sc.getConf.contains(sysProp)) {
-      arrayBuf += (optionName, sc.getConf.get(sysProp))
-    }
-  }
+  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.actorSystem) {
 
   override def start() {
     super.start()
-    val argsArrayBuf = new ArrayBuffer[String]()
-    List(("--num-executors", "SPARK_EXECUTOR_INSTANCES", "spark.executor.instances"),
-      ("--num-executors", "SPARK_WORKER_INSTANCES", "spark.worker.instances"))
-      .foreach { case (optName, envVar, sysProp) => addArg(optName, envVar, sysProp, argsArrayBuf) }
-    val args = new ApplicationMasterArguments(argsArrayBuf.toArray)
-    totalExecutors.set(args.numExecutors)
+    var numExecutors = 2
+    if (sc.getConf.contains("spark.executor.instances")) {
+      numExecutors = sc.getConf.getInt("spark.executor.instances", 2)
+    } else if (System.getenv("SPARK_EXECUTOR_INSTANCES") != null) {
+      IntParam.unapply(System.getenv("SPARK_EXECUTOR_INSTANCES")).map(_.toInt).getOrElse(2)
+    }
+    totalExpectedExecutors.set(numExecutors)
   }
 }