private def schedule(): Unit = {
    if (state != RecoveryState.ALIVE) {
      return
    }
    // Drivers take strict precedence over executors
    val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
    val numWorkersAlive = shuffledAliveWorkers.size
    var curPos = 0
→   for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
      // We assign workers to each waiting driver in a round-robin fashion. For each driver, we
      // start from the last worker that was assigned a driver, and continue onwards until we have
      // explored all alive workers.
      var launched = false
      var numWorkersVisited = 0
      while (numWorkersVisited < numWorkersAlive && !launched) {
        val worker = shuffledAliveWorkers(curPos)
        numWorkersVisited += 1
        if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
          launchDriver(worker, driver)
          waitingDrivers -= driver
          launched = true
        }
        curPos = (curPos + 1) % numWorkersAlive
      }
    }
→  startExecutorsOnWorkers()
  }

→  private val spreadOutApps = conf.getBoolean("spark.deploy.spreadOut", true)
/**
   * Schedule and launch executors on workers
   */
  private def startExecutorsOnWorkers(): Unit = {
    // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app
    // in the queue, then the second app, etc.
    for (app <- waitingApps) {
→      val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1)
      // If the cores left is less than the coresPerExecutor,the cores left will not be allocated
      if (app.coresLeft >= coresPerExecutor) {
        // Filter out workers that don't have enough resources to launch an executor
        val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
          .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB &&
            worker.coresFree >= coresPerExecutor)
          .sortBy(_.coresFree).reverse
→        val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps)

        // Now that we've decided how many cores to allocate on each worker, let's allocate them
        for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
          allocateWorkerResourceToExecutors(
            app, assignedCores(pos), app.desc.coresPerExecutor, usableWorkers(pos))
        }
      }
    }
  }

private def scheduleExecutorsOnWorkers(
    app: ApplicationInfo,
    usableWorkers: Array[WorkerInfo],
    spreadOutApps: Boolean): Array[Int] = {
  val coresPerExecutor = app.desc.coresPerExecutor
  val minCoresPerExecutor = coresPerExecutor.getOrElse(1)
  val oneExecutorPerWorker = coresPerExecutor.isEmpty
  val memoryPerExecutor = app.desc.memoryPerExecutorMB
  val numUsable = usableWorkers.length
  val assignedCores = new Array[Int](numUsable) // Number of cores to give to each worker
  val assignedExecutors = new Array[Int](numUsable) // Number of new executors on each worker
  var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
  ....
}

   private def scheduleExecutorsOnWorkers(
      app: ApplicationInfo,
      usableWorkers: Array[WorkerInfo],
      spreadOutApps: Boolean): Array[Int] = {
        ....
        var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
        .... 
        /** Return whether the specified worker can launch an executor for this app. */
→          def canLaunchExecutor(pos: Int): Boolean = {
              val keepScheduling = coresToAssign >= minCoresPerExecutor
              val enoughCores = usableWorkers(pos).coresFree - assignedCores(pos) >= minCoresPerExecutor

              // If we allow multiple executors per worker, then we can always launch new executors.
              // Otherwise, if there is already an executor on this worker, just give it more cores.
              val launchingNewExecutor = !oneExecutorPerWorker || assignedExecutors(pos) == 0
              if (launchingNewExecutor) {
                val assignedMemory = assignedExecutors(pos) * memoryPerExecutor
                val enoughMemory = usableWorkers(pos).memoryFree - assignedMemory >= memoryPerExecutor
                val underLimit = assignedExecutors.sum + app.executors.size < app.executorLimit
                keepScheduling && enoughCores && enoughMemory && underLimit
              } else {
                // We're adding cores to an existing executor, so no need
                // to check memory and executor limits
                keepScheduling && enoughCores
              }
            }
        ....
   }

private def scheduleExecutorsOnWorkers(
     app: ApplicationInfo,
     usableWorkers: Array[WorkerInfo],
     spreadOutApps: Boolean): Array[Int] = {
    ....
/** Return whether the specified worker can launch an executor for this app. */
   def canLaunchExecutor(pos: Int): Boolean = {.....}

   // Keep launching executors until no more workers can accommodate any
   // more executors, or if we have reached this application's limits
   var freeWorkers = (0 until numUsable).filter(canLaunchExecutor)
   while (freeWorkers.nonEmpty) {
     freeWorkers.foreach { pos =>
       var keepScheduling = true
       while (keepScheduling && canLaunchExecutor(pos)) {
         coresToAssign -= minCoresPerExecutor
         assignedCores(pos) += minCoresPerExecutor

         // If we are launching one executor per worker, then every iteration assigns 1 core
         // to the executor. Otherwise, every iteration assigns cores to a new executor.
         if (oneExecutorPerWorker) {
           assignedExecutors(pos) = 1
         } else {
           assignedExecutors(pos) += 1
         }

         // Spreading out an application means spreading out its executors across as
         // many workers as possible. If we are not spreading out, then we should keep
         // scheduling executors on this worker until we use all of its resources.
         // Otherwise, just move on to the next worker.
         if (spreadOutApps) {
           keepScheduling = false
         }
       }
     }
     freeWorkers = freeWorkers.filter(canLaunchExecutor)
   }
   assignedCores
 }

val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps)

// Now that we've decided how many cores to allocate on each worker, let's allocate them
for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
  allocateWorkerResourceToExecutors(
    app, assignedCores(pos), app.desc.coresPerExecutor, usableWorkers(pos))
}

  private def allocateWorkerResourceToExecutors(
      app: ApplicationInfo,
      assignedCores: Int,
      coresPerExecutor: Option[Int],
      worker: WorkerInfo): Unit = {
    // If the number of cores per executor is specified, we divide the cores assigned
    // to this worker evenly among the executors with no remainder.
    // Otherwise, we launch a single executor that grabs all the assignedCores on this worker.
→    val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)
→    val coresToAssign = coresPerExecutor.getOrElse(assignedCores)
    for (i <- 1 to numExecutors) {
→      val exec = app.addExecutor(worker, coresToAssign)
      launchExecutor(worker, exec)
      app.state = ApplicationState.RUNNING
    }
  }

private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc): Unit = {
  logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
  worker.addExecutor(exec)
  worker.endpoint.send(LaunchExecutor(masterUrl,
    exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory))
  exec.application.driver.send(
    ExecutorAdded(exec.id, worker.id, worker.hostPort, exec.cores, exec.memory))
}

  override def receive(
      client: TransportClient,
→       message: ByteBuffer,
      callback: RpcResponseCallback): Unit = {
→   val messageToDispatch = internalReceive(client, message)
    dispatcher.postRemoteMessage(messageToDispatch, callback)
  }