case RegisterApplication(description, driver) => {

      // TODO Prevent repeated registrations from some driver
      //standby master不调度

      if (state == RecoveryState.STANDBY) {

        // ignore, don't send response

      } else {

        logInfo("Registering app " + description.name)

        val app = createApplication(description, driver)
        //注册app，即将其加入到waitingApps中

        registerApplication(app)

        logInfo("Registered app " + description.name + " with ID " + app.id)
        //将app加入持久化引擎，主要是为了故障恢复

        persistenceEngine.addApplication(app)
        //向driver发送RegisteredApplication消息表明master已经注册了这个app

        driver.send(RegisteredApplication(app.id, self))
        //为waitingApps中的app调度资源

        schedule()

      }

    }

  /**

   * Schedule the currently available resources among waiting apps. This method will be called

   * every time a new app joins or resource availability changes.

   */

  private def schedule(): Unit = {

    if (state != RecoveryState.ALIVE) { return }

    // Drivers take strict precedence over executors
    //为了避免每次schedule，总是在相同的worker上分配资源，所有这里打乱worker顺序。

    val shuffledWorkers = Random.shuffle(workers) // Randomization helps balance drivers
    //下面这个for循环是为driver调度资源，因为这里只将application的调度，所以driver的调度不说了。

    for (worker <- shuffledWorkers if worker.state == WorkerState.ALIVE) {

      for (driver <- waitingDrivers) {

        if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {

          launchDriver(worker, driver)

          waitingDrivers -= driver

        }

      }

    }

    //为application调度资源

    startExecutorsOnWorkers()

  }

  /**

   * 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.
    // 为waitingApps中的app调度资源，app.coresLeft是app还有多少core没有分配

    for (app <- waitingApps if app.coresLeft > 0) {

      val coresPerExecutor: Option[Int] = app.desc.coresPerExecutor

      // Filter out workers that don't have enough resources to launch an executor
      // 筛选出状态为ALIVE并且这个worker剩余内存，剩余core都大于等于app的要求，然后按照coresFree降序排列

      val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)

        .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB &&

          worker.coresFree >= coresPerExecutor.getOrElse(1))

        .sortBy(_.coresFree).reverse
      //在usableWorkers上为app分配Executor

      val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps)

      // Now that we've decided how many cores to allocate on each worker, let's allocate them
      // 在worker上启动Executor进程

      for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {

        allocateWorkerResourceToExecutors(

          app, assignedCores(pos), coresPerExecutor, usableWorkers(pos))

      }

    }

  }

  /**

   * Schedule executors to be launched on the workers.

   * Returns an array containing number of cores assigned to each worker.

   *

   * There are two modes of launching executors. The first attempts to spread out an application's

   * executors on as many workers as possible, while the second does the opposite (i.e. launch them

   * on as few workers as possible). The former is usually better for data locality purposes and is

   * the default.

   *

   * The number of cores assigned to each executor is configurable. When this is explicitly set,

   * multiple executors from the same application may be launched on the same worker if the worker

   * has enough cores and memory. Otherwise, each executor grabs all the cores available on the

   * worker by default, in which case only one executor may be launched on each worker.

   *

   * It is important to allocate coresPerExecutor on each worker at a time (instead of 1 core

   * at a time). Consider the following example: cluster has 4 workers with 16 cores each.

   * User requests 3 executors (spark.cores.max = 48, spark.executor.cores = 16). If 1 core is

   * allocated at a time, 12 cores from each worker would be assigned to each executor.

   * Since 12 < 16, no executors would launch [SPARK-8881].

   */

  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)

    /** Return whether the specified worker can launch an executor for this app. */
    //是否可以在一个worker上分配Executor

    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) {
        //在不里，需要检查worker的空闲core和内存是否够用

        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
        //尤其需要注意的是，oneExecutorPerWorker机制下，不检测内存限制，很重要。

        keepScheduling && enoughCores

      }

    }

    // 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)) {
          //要分配的cores

          coresToAssign -= minCoresPerExecutor
          //已分配的cores

          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.
          //一个worker只启动一个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.
          //如果没有开启spreadOUt算法，就一直在一个worker上分配，直到不能再分配为止。

          if (spreadOutApps) {

            keepScheduling = false

          }

        }

      }

      freeWorkers = freeWorkers.filter(canLaunchExecutor)

    }

    assignedCores

  }

  /**

   * Allocate a worker's resources to one or more executors.

   * @param app the info of the application which the executors belong to

   * @param assignedCores number of cores on this worker for this application

   * @param coresPerExecutor number of cores per executor

   * @param worker the worker info

   */

  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.
    //计算要创建多少个Executor进程，默认值是1.

    val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)

    val coresToAssign = coresPerExecutor.getOrElse(assignedCores)

    for (i <- 1 to numExecutors) {

      val exec = app.addExecutor(worker, coresToAssign)
      //真正的启动Executor进程了。

      launchExecutor(worker, exec)

      app.state = ApplicationState.RUNNING

    }

  }