def main(args: Array[String]): Unit = {
  // 注册日志
  SignalUtils.registerLogger(log)
  // 实例化 ApplicationMasterArguments对应，用来解析传入的参数
  val amArgs = new ApplicationMasterArguments(args)
  // 是否设置了 --properties-file 参数，如果设置，则将properties文件中的配置加载到系统参数中
  if (amArgs.propertiesFile != null) {
    Utils.getPropertiesFromFile(amArgs.propertiesFile).foreach { case (k, v) =>
      sys.props(k) = v
    }
  }
  // 以指定的用户运行 ApplicationMaster
  SparkHadoopUtil.get.runAsSparkUser { () =>
    master = new ApplicationMaster(amArgs, new YarnRMClient)
    // 运行 application master
    System.exit(master.run())
  }
}

final def run(): Int = {
  try {
    // 获取application的id
    val appAttemptId = client.getAttemptId()
    var attemptID: Option[String] = None
    // 是否是集群模式，则设置集群模式需要的一些属性
    if (isClusterMode) {
      // Set the web ui port to be ephemeral for yarn so we don't conflict with
      // other spark processes running on the same box
      System.setProperty("spark.ui.port", "0")
      // Set the master and deploy mode property to match the requested mode.
      System.setProperty("spark.master", "yarn")
      System.setProperty("spark.submit.deployMode", "cluster")
      // Set this internal configuration if it is running on cluster mode, this
      // configuration will be checked in SparkContext to avoid misuse of yarn cluster mode.
      System.setProperty("spark.yarn.app.id", appAttemptId.getApplicationId().toString())
      attemptID = Option(appAttemptId.getAttemptId.toString)
    }
    // <1>
    // 设置调用上下文，从spark.log.callerContext配置中读取
    new CallerContext(
      "APPMASTER", sparkConf.get(APP_CALLER_CONTEXT),
      Option(appAttemptId.getApplicationId.toString), attemptID).setCurrentContext()
    logInfo("ApplicationAttemptId: " + appAttemptId)
    // This shutdown hook should run *after* the SparkContext is shut down.
    // 设置钩子函数，以便在 SparkContext 之后调用，进行操作
    val priority = ShutdownHookManager.SPARK_CONTEXT_SHUTDOWN_PRIORITY - 1
    //
    ShutdownHookManager.addShutdownHook(priority) { () =>
      val maxAppAttempts = client.getMaxRegAttempts(sparkConf, yarnConf)
      val isLastAttempt = client.getAttemptId().getAttemptId() >= maxAppAttempts
      if (!finished) {
        // The default state of ApplicationMaster is failed if it is invoked by shut down hook.
        // This behavior is different compared to 1.x version.
        // If user application is exited ahead of time by calling System.exit(N), here mark
        // this application as failed with EXIT_EARLY. For a good shutdown, user shouldn't call
        // System.exit(0) to terminate the application.
        finish(finalStatus,
          ApplicationMaster.EXIT_EARLY,
          "Shutdown hook called before final status was reported.")
      }
      if (!unregistered) {
        // we only want to unregister if we don't want the RM to retry
        if (finalStatus == FinalApplicationStatus.SUCCEEDED || isLastAttempt) {
          unregister(finalStatus, finalMsg)
          cleanupStagingDir()
        }
      }
    }
    // <2>
    // Call this to force generation of secret so it gets populated into the
    // Hadoop UGI. This has to happen before the startUserApplication which does a
    // doAs in order for the credentials to be passed on to the executor containers.
    // 根据spark配置生成安全管理器
    val securityMgr = new SecurityManager(sparkConf)
    // If the credentials file config is present, we must periodically renew tokens. So create
    // a new AMDelegationTokenRenewer
    // spark.yarn.credentials.file
    if (sparkConf.contains(CREDENTIALS_FILE_PATH.key)) {
      // If a principal and keytab have been set, use that to create new credentials for executors
      // periodically
      credentialRenewer =
        new ConfigurableCredentialManager(sparkConf, yarnConf).credentialRenewer()
      credentialRenewer.scheduleLoginFromKeytab()
    }
    // <3>
    // 根据不同的集群模式，调用不同的方法
    if (isClusterMode) {
      runDriver(securityMgr)
    } else {
      runExecutorLauncher(securityMgr)
    }
    // <4>
  } catch {
    case e: Exception =>
      // catch everything else if not specifically handled
      logError("Uncaught exception: ", e)
      finish(FinalApplicationStatus.FAILED,
        ApplicationMaster.EXIT_UNCAUGHT_EXCEPTION,
        "Uncaught exception: " + e)
  }
  exitCode
}

private def runDriver(securityMgr: SecurityManager): Unit = {
  // 添加IP过滤器
  addAmIpFilter()
  // 启动用户Application，就是利用反射机制，运行用户指定的class中的main方法
  userClassThread = startUserApplication()
  // This a bit hacky, but we need to wait until the spark.driver.port property has
  // been set by the Thread executing the user class.
  logInfo("Waiting for spark context initialization...")
  // spark.yarn.am.waitTime
  val totalWaitTime = sparkConf.get(AM_MAX_WAIT_TIME)
  try {
    // 等待 SparkContext的生成，最多等待 totalWaitTime 毫秒
    val sc = ThreadUtils.awaitResult(sparkContextPromise.future,
      Duration(totalWaitTime, TimeUnit.MILLISECONDS))
    if (sc != null) {
      rpcEnv = sc.env.rpcEnv
      // 启动 driver RPC
      val driverRef = runAMEndpoint(
        sc.getConf.get("spark.driver.host"),
        sc.getConf.get("spark.driver.port"),
        isClusterMode = true)
      // 注册Application master
      registerAM(sc.getConf, rpcEnv, driverRef, sc.ui.map(_.webUrl), securityMgr)
    } else {
      // Sanity check; should never happen in normal operation, since sc should only be null
      // if the user app did not create a SparkContext.
      if (!finished) {
        throw new IllegalStateException("SparkContext is null but app is still running!")
      }
    }
    // 等待 用户线程 的完成
    userClassThread.join()
  } catch {
    case e: SparkException if e.getCause().isInstanceOf[TimeoutException] =>
      logError(
        s"SparkContext did not initialize after waiting for $totalWaitTime ms. " +
         "Please check earlier log output for errors. Failing the application.")
      finish(FinalApplicationStatus.FAILED,
        ApplicationMaster.EXIT_SC_NOT_INITED,
        "Timed out waiting for SparkContext.")
  }
}

private def runExecutorLauncher(securityMgr: SecurityManager): Unit = {
  val port = sparkConf.get(AM_PORT)
  rpcEnv = RpcEnv.create("sparkYarnAM", Utils.localHostName, port, sparkConf, securityMgr,
    clientMode = true)
  val driverRef = waitForSparkDriver()
  addAmIpFilter()
  registerAM(sparkConf, rpcEnv, driverRef, sparkConf.getOption("spark.driver.appUIAddress"),
    securityMgr)
  // In client mode the actor will stop the reporter thread.
  reporterThread.join()
}

private def registerAM(
    _sparkConf: SparkConf,
    _rpcEnv: RpcEnv,
    driverRef: RpcEndpointRef,
    uiAddress: Option[String],
    securityMgr: SecurityManager) = {
  val appId = client.getAttemptId().getApplicationId().toString()
  val attemptId = client.getAttemptId().getAttemptId().toString()
  val historyAddress =
    _sparkConf.get(HISTORY_SERVER_ADDRESS)
      .map { text => SparkHadoopUtil.get.substituteHadoopVariables(text, yarnConf) }
      .map { address => s"${address}${HistoryServer.UI_PATH_PREFIX}/${appId}/${attemptId}" }
      .getOrElse("")
  val driverUrl = RpcEndpointAddress(
    _sparkConf.get("spark.driver.host"),
    _sparkConf.get("spark.driver.port").toInt,
    CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
  // Before we initialize the allocator, let's log the information about how executors will
  // be run up front, to avoid printing this out for every single executor being launched.
  // Use placeholders for information that changes such as executor IDs.
  logInfo {
    val executorMemory = sparkConf.get(EXECUTOR_MEMORY).toInt
    val executorCores = sparkConf.get(EXECUTOR_CORES)
    val dummyRunner = new ExecutorRunnable(None, yarnConf, sparkConf, driverUrl, "<executorId>",
      "<hostname>", executorMemory, executorCores, appId, securityMgr, localResources)
    dummyRunner.launchContextDebugInfo()
  }
  allocator = client.register(driverUrl,
    driverRef,
    yarnConf,
    _sparkConf,
    uiAddress,
    historyAddress,
    securityMgr,
    localResources)
  allocator.allocateResources()
  // 在 客户端 模式中 的 runExecutorLauncher 方法中join
  reporterThread = launchReporterThread()
}

private def launchReporterThread(): Thread = {
  // The number of failures in a row until Reporter thread give up
  // 获取配置的 reporter 线程最大失败次数
  val reporterMaxFailures = sparkConf.get(MAX_REPORTER_THREAD_FAILURES)
  val t = new Thread {
    override def run() {
      var failureCount = 0
      while (!finished) {
        try {
          // 如果 allocator 失败的 executor 个数已经超出了设置的最大值，则 终止 application 的运行（终止用户类的运行）
          if (allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
            finish(FinalApplicationStatus.FAILED,
              ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
              s"Max number of executor failures ($maxNumExecutorFailures) reached")
          } else {
            // 向YarnAllocator 申请资源
            logDebug("Sending progress")
            allocator.allocateResources()
          }
          failureCount = 0
        } catch {
          // 不同的异常不同的处理，中断异常，有可能是 finish 方法抛出来的
          case i: InterruptedException => // do nothing
          case e: ApplicationAttemptNotFoundException =>
            failureCount += 1
            logError("Exception from Reporter thread.", e)
            finish(FinalApplicationStatus.FAILED, ApplicationMaster.EXIT_REPORTER_FAILURE,
              e.getMessage)
          // 如果 reporter 线程的尝试次数超过配置的最大值，则终止 用户类的运行
          case e: Throwable =>
            failureCount += 1
            if (!NonFatal(e) || failureCount >= reporterMaxFailures) {
              finish(FinalApplicationStatus.FAILED,
                ApplicationMaster.EXIT_REPORTER_FAILURE, "Exception was thrown " +
                  s"$failureCount time(s) from Reporter thread.")
            } else {
              logWarning(s"Reporter thread fails $failureCount time(s) in a row.", e)
            }
        }
        try {
          val numPendingAllocate = allocator.getPendingAllocate.size
          var sleepStart = 0L
          var sleepInterval = 200L // ms
          allocatorLock.synchronized {
            // 计算 reporter 的休眠时长， 根据
            sleepInterval =
              if (numPendingAllocate > 0 || allocator.getNumPendingLossReasonRequests > 0) {
                // 进入这里，表示有丢失的container， 也有正在添加的container
                val currentAllocationInterval =
                  math.min(heartbeatInterval, nextAllocationInterval)
                nextAllocationInterval = currentAllocationInterval * 2 // avoid overflow
                currentAllocationInterval
              } else {
                nextAllocationInterval = initialAllocationInterval
                heartbeatInterval
              }
            sleepStart = System.currentTimeMillis()
            allocatorLock.wait(sleepInterval)
          }
          val sleepDuration = System.currentTimeMillis() - sleepStart
          // 如果符合这个条件，说明上面的 allocatorLock.wait所等待的时间不够，改用Thread.sleep来休眠
          if (sleepDuration < sleepInterval) {
            // log when sleep is interrupted
            logDebug(s"Number of pending allocations is $numPendingAllocate. " +
                s"Slept for $sleepDuration/$sleepInterval ms.")
            // if sleep was less than the minimum interval, sleep for the rest of it
            val toSleep = math.max(0, initialAllocationInterval - sleepDuration)
            if (toSleep > 0) {
              logDebug(s"Going back to sleep for $toSleep ms")
              // use Thread.sleep instead of allocatorLock.wait. there is no need to be woken up
              // by the methods that signal allocatorLock because this is just finishing the min
              // sleep interval, which should happen even if this is signalled again.
              Thread.sleep(toSleep)
            }
          } else {
            logDebug(s"Number of pending allocations is $numPendingAllocate. " +
                s"Slept for $sleepDuration/$sleepInterval.")
          }
        } catch {
          case e: InterruptedException =>
        }
      }
    }
  }
  // setting to daemon status, though this is usually not a good idea.
  t.setDaemon(true)
  t.setName("Reporter")
  t.start()
  logInfo(s"Started progress reporter thread with (heartbeat : $heartbeatInterval, " +
          s"initial allocation : $initialAllocationInterval) intervals")
  t
}

private def startUserApplication(): Thread = {
  logInfo("Starting the user application in a separate Thread")
  // 获取用户的类路经
  val classpath = Client.getUserClasspath(sparkConf)
  val urls = classpath.map { entry =>
    new URL("file:" + new File(entry.getPath()).getAbsolutePath())
  }
  val userClassLoader =
    if (Client.isUserClassPathFirst(sparkConf, isDriver = true)) {
      new ChildFirstURLClassLoader(urls, Utils.getContextOrSparkClassLoader)
    } else {
      new MutableURLClassLoader(urls, Utils.getContextOrSparkClassLoader)
    }
  var userArgs = args.userArgs
  if (args.primaryPyFile != null && args.primaryPyFile.endsWith(".py")) {
    // When running pyspark, the app is run using PythonRunner. The second argument is the list
    // of files to add to PYTHONPATH, which Client.scala already handles, so it's empty.
    userArgs = Seq(args.primaryPyFile, "") ++ userArgs
  }
  if (args.primaryRFile != null && args.primaryRFile.endsWith(".R")) {
    // TODO(davies): add R dependencies here
  }
  // 用户类的main方法
  val mainMethod = userClassLoader.loadClass(args.userClass)
    .getMethod("main", classOf[Array[String]])
  val userThread = new Thread {
    override def run() {
      try {
        // 在新的线程中运行 用户类的 main方法
        mainMethod.invoke(null, userArgs.toArray)
        // 设置完成状态（用户类执行完成）
        finish(FinalApplicationStatus.SUCCEEDED, ApplicationMaster.EXIT_SUCCESS)
        // 记录日志
        logDebug("Done running users class")
      } catch {
        // 根据异常的原因，设置用户类执行的完成状态（失败状态）
        // 异常的原因分为
        // 中断异常
        // app异常
        // 代码异常
        case e: InvocationTargetException =>
          e.getCause match {
            case _: InterruptedException =>
              // Reporter thread can interrupt to stop user class
            case SparkUserAppException(exitCode) =>
              val msg = s"User application exited with status $exitCode"
              logError(msg)
              finish(FinalApplicationStatus.FAILED, exitCode, msg)
            case cause: Throwable =>
              logError("User class threw exception: " + cause, cause)
              finish(FinalApplicationStatus.FAILED,
                ApplicationMaster.EXIT_EXCEPTION_USER_CLASS,
                "User class threw exception: " + cause)
          }
          sparkContextPromise.tryFailure(e.getCause())
      } finally {
        // Notify the thread waiting for the SparkContext, in case the application did not
        // instantiate one. This will do nothing when the user code instantiates a SparkContext
        // (with the correct master), or when the user code throws an exception (due to the
        // tryFailure above).
        sparkContextPromise.trySuccess(null)
      }
    }
  }
  // 设置线程的 类加载器 、 线程名字， 启动线程
  userThread.setContextClassLoader(userClassLoader)
  userThread.setName("Driver")
  userThread.start()
  userThread
}

private def parseArgs(inputArgs: List[String]): Unit = {
  val userArgsBuffer = new ArrayBuffer[String]()
  var args = inputArgs
  // 从这个匹配可以看出，可以使用的参数列在下面，如果包含了其他参数，系统会异常退出
  // --jar jar包
  // --class 类
  // --primary-py-file  PYTHON语言编写的application
  // --primary-r-file   R语言编写的application
  // --arg  其他参数，多个参数需要使用多个 --arg 1 --arg "name"
  // --properties-file 配置文件
  while (!args.isEmpty) {
    // --num-workers, --worker-memory, and --worker-cores are deprecated since 1.0,
    // the properties with executor in their names are preferred.
    // 开始解析 类参数  case ("--jar") :: value :: tail 就是提取参数和参数名 ，并把剩余的参数放到 tail中
    args match {
      case ("--jar") :: value :: tail =>
        userJar = value
        args = tail
      case ("--class") :: value :: tail =>
        userClass = value
        args = tail
      case ("--primary-py-file") :: value :: tail =>
        primaryPyFile = value
        args = tail
      case ("--primary-r-file") :: value :: tail =>
        primaryRFile = value
        args = tail
      case ("--arg") :: value :: tail =>
        userArgsBuffer += value
        args = tail
      case ("--properties-file") :: value :: tail =>
        propertiesFile = value
        args = tail
      case _ =>
        printUsageAndExit(1, args)
    }
  }
  if (primaryPyFile != null && primaryRFile != null) {
    // scalastyle:off println
    System.err.println("Cannot have primary-py-file and primary-r-file at the same time")
    // scalastyle:on println
    System.exit(-1)
  }
  userArgs = userArgsBuffer.toList
}

args match {
    case ("--jar") :: value :: tail =>
      userJar = value
      args = tail
    case ("--class") :: value :: tail =>
      userClass = value
      args = tail

def register(
      driverUrl: String,
      driverRef: RpcEndpointRef,
      conf: YarnConfiguration,
      sparkConf: SparkConf,
      uiAddress: Option[String],
      uiHistoryAddress: String,
      securityMgr: SecurityManager,
      localResources: Map[String, LocalResource]
    ): YarnAllocator = {
    // 调用AMRMClient自身的方法来生成AMRMClient，再使用 Yarn 配置进行初始化，启动AMRMClient
    amClient = AMRMClient.createAMRMClient()
    amClient.init(conf)
    amClient.start()
    this.uiHistoryAddress = uiHistoryAddress
    val trackingUrl = uiAddress.getOrElse {
      if (sparkConf.get(ALLOW_HISTORY_SERVER_TRACKING_URL)) uiHistoryAddress else ""
    }
    logInfo("Registering the ApplicationMaster")
    // 向 ResourceManager 注册 application master，从代码看出 application master就是本机，TODO 这个本机是啥呢？？？
    synchronized {
      amClient.registerApplicationMaster(Utils.localHostName(), 0, trackingUrl)
      registered = true
    }
    // 生成 YarnAllocator
    // driverUrl和driverRef需要说一下，
    // driverUrl，是driver运行的地址，会传递给Executor，应该是用于Execurot与driver进行交互
    // driverRef 在YarnAllocator中使用，用于同步executor的id，以及 发送删除executor的信息
    new YarnAllocator(driverUrl, driverRef, conf, sparkConf, amClient, getAttemptId(), securityMgr,
      localResources, new SparkRackResolver())
  }

def unregister(status: FinalApplicationStatus, diagnostics: String = ""): Unit = synchronized {
	if (registered) {
	  amClient.unregisterApplicationMaster(status, diagnostics, uiHistoryAddress)
	}
}

/** 获取注册AM的最大尝试次数 分别从spark配置和yarn配置中读取 如果spark配置中设置了，则使用spark和yarn配置中最小那个值 */
def getMaxRegAttempts(sparkConf: SparkConf, yarnConf: YarnConfiguration): Int = {
  val sparkMaxAttempts = sparkConf.get(MAX_APP_ATTEMPTS).map(_.toInt)
  val yarnMaxAttempts = yarnConf.getInt(
    YarnConfiguration.RM_AM_MAX_ATTEMPTS, YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS)
  sparkMaxAttempts match {
    case Some(x) => if (x <= yarnMaxAttempts) x else yarnMaxAttempts
    case None => yarnMaxAttempts
  }
}

def main(args: Array[String]): Unit = {
  
  ...
  
  SparkHadoopUtil.get.runAsSparkUser { () =>
    master = new ApplicationMaster(amArgs, new YarnRMClient)
    System.exit(master.run())
  }
}

private def registerAM(
      _sparkConf: SparkConf,
      _rpcEnv: RpcEnv,
      driverRef: RpcEndpointRef,
      uiAddress: Option[String],
      securityMgr: SecurityManager) = {
    
    ...
    allocator = client.register(driverUrl,
      driverRef,
      yarnConf,
      _sparkConf,
      uiAddress,
      historyAddress,
      securityMgr,
      localResources)
    allocator.allocateResources()
    reporterThread = launchReporterThread()
  }

def allocateResources(): Unit = synchronized {
    updateResourceRequests()
    // 处理指示器
    val progressIndicator = 0.1f
    // Poll the ResourceManager. This doubles as a heartbeat if there are no pending container
    // requests.
    // 调用 AMRMClient 分配资源
    val allocateResponse = amClient.allocate(progressIndicator)
    // 得到已经分配的 container
    val allocatedContainers = allocateResponse.getAllocatedContainers()
    if (allocatedContainers.size > 0) {
      // 输出日志信息，包括 分配的container数量，正在运行的以及启动的executor数量，以及可用的资源信息
      logDebug(("Allocated containers: %d. Current executor count: %d. " +
        "Launching executor count: %d. Cluster resources: %s.")
        .format(
          allocatedContainers.size,
          numExecutorsRunning.get,
          numExecutorsStarting.get,
          allocateResponse.getAvailableResources))
      // 处理已经分配的container
      handleAllocatedContainers(allocatedContainers.asScala)
    }
    // 获取已经执行完成的 container
    val completedContainers = allocateResponse.getCompletedContainersStatuses()
    if (completedContainers.size > 0) {
      logDebug("Completed %d containers".format(completedContainers.size))
      //处理已经完成的container
      processCompletedContainers(completedContainers.asScala)
      logDebug("Finished processing %d completed containers. Current running executor count: %d."
        .format(completedContainers.size, numExecutorsRunning.get))
    }
  }

def updateResourceRequests(): Unit = {
    // 得到正在添加的container 并 得到正在添加 container的数量
    val pendingAllocate = getPendingAllocate
    val numPendingAllocate = pendingAllocate.size
    // 计划要启动的executor数量 - 正在请求启动的executor - 已经启动的executor - 已经运行任务的executor = 缺少多少executor
    // 所以这里是在计算 比预计要启动的executor，还缺少多少个
    val missing = targetNumExecutors - numPendingAllocate -
      numExecutorsStarting.get - numExecutorsRunning.get
    logDebug(s"Updating resource requests, target: $targetNumExecutors, " +
      s"pending: $numPendingAllocate, running: ${numExecutorsRunning.get}, " +
      s"executorsStarting: ${numExecutorsStarting.get}")
    // <1>
    // missing 可以为正数 也可能为负数，负数则说明 动态分配分配多了，但是没有超过最大个数，这个数是通过计划启动个数算出来的
    if (missing > 0) {
      logInfo(s"Will request $missing executor container(s), each with " +
        s"${resource.getVirtualCores} core(s) and " +
        s"${resource.getMemory} MB memory (including $memoryOverhead MB of overhead)")
      // 将要添加的container请求拆分到三个组：位置匹配列表、位置不匹配列表 和 无位置列表。
      // 对于位置匹配的 container 请求，将他们放到可用的地方，等待分配
      // 对于位置不匹配的和无位置的container请求，取消这些container的请求，因为 位置优先权已经变了，
      val (localRequests, staleRequests, anyHostRequests) = splitPendingAllocationsByLocality(
        hostToLocalTaskCounts, pendingAllocate)
      // <2>
      // cancel "stale" requests for locations that are no longer needed
      // 对于位置不匹配的container，进行移除操作，并记录日志， 移除了N个container
      staleRequests.foreach { stale =>
        amClient.removeContainerRequest(stale)
      }
      val cancelledContainers = staleRequests.size
      if (cancelledContainers > 0) {
        logInfo(s"Canceled $cancelledContainers container request(s) (locality no longer needed)")
      }
      // <3>
      // 计算还可以分配的container的数量，就是 缺少的 + 取消的
      // 因为cancelledContainers的个数实际上就是从pendingAllocate 中取消的
      val availableContainers = missing + cancelledContainers
      // 计算潜在的container就是 可以分配的container + 上面那些 不限制位置的的contianer的数量
      val potentialContainers = availableContainers + anyHostRequests.size
      // TODO 这是在弄啥 ？？？ 应该是根据 潜在container的数量，生成对应个的contaner的位置引用
      val containerLocalityPreferences = if (labelExpression.isEmpty) {
        containerPlacementStrategy.localityOfRequestedContainers(
          potentialContainers, numLocalityAwareTasks, hostToLocalTaskCounts,
          allocatedHostToContainersMap, localRequests)
      } else {
        Array.empty[ContainerLocalityPreferences]
      }
      // 根据上面的containerLocalityPreferences 创建container添加请求（注意是请求）
      // newLocalityRequest 用来记录要添加的container的请求
      val newLocalityRequests = new mutable.ArrayBuffer[ContainerRequest]
      containerLocalityPreferences.foreach {
        // createContainerRequest 用来生成container的创建请求
        case ContainerLocalityPreferences(nodes, racks) if nodes != null =>
          newLocalityRequests += createContainerRequest(resource, nodes, racks)
        case _ =>
      }
      // 需要再次判断 总的container的数量，如果availableContainers > newLocalityRequests 表示，还不够
      // 为啥 availableContainers 会大于 newLocalityRequests ？ 因为 labelExpression.isEmpty 为空时，会生成一个空的Array
      if (availableContainers >= newLocalityRequests.size) {
        // more containers are available than needed for locality, fill in requests for any host
        for (i <- 0 until (availableContainers - newLocalityRequests.size)) {
          // createContainerRequest 用来生成container的创建请求
          newLocalityRequests += createContainerRequest(resource, null, null)
        }
      } else {
        // 这里的 newLocalityRequests 实际上对应的个是 potentialContainers = availableContainers + anyHostRequests.size
        // 因此，如果 newLocalityRequests > availableContainers 则表示生成多了，且 anyHostRequests 的多了
        val numToCancel = newLocalityRequests.size - availableContainers
        // 因此释放到一些多余的 anyHostRequests
        // cancel some requests without locality preferences to schedule more local containers
        anyHostRequests.slice(0, numToCancel).foreach { nonLocal =>
          amClient.removeContainerRequest(nonLocal)
        }
        if (numToCancel > 0) {
          logInfo(s"Canceled $numToCancel unlocalized container requests to resubmit with locality")
        }
      }
      // <4>
      // AMRMClient 请求添加container
      newLocalityRequests.foreach { request =>
        amClient.addContainerRequest(request)
      }
      // <5>
      if (log.isInfoEnabled()) {
        val (localized, anyHost) = newLocalityRequests.partition(_.getNodes() != null)
        if (anyHost.nonEmpty) {
          logInfo(s"Submitted ${anyHost.size} unlocalized container requests.")
        }
        localized.foreach { request =>
          logInfo(s"Submitted container request for host ${hostStr(request)}.")
        }
      }
      // <6>
    } else if (numPendingAllocate > 0 && missing < 0) {
      // 如果不缺少executor，并且还有正在添加的executor
      // 计算要取消的contaner的数量，为什么是最小值，个人这样理解：-missing，其实是多出来的，但是在计算missing的时候，
      // 已经减去 numPendingAllocate 了，也就是说 numPendingAllocate 认为是已经使用的数量
      // 因此，如果取最大值，那么当 numPendingAllocate > -missing 时，删除的container就太多了
      val numToCancel = math.min(numPendingAllocate, -missing)
      logInfo(s"Canceling requests for $numToCancel executor container(s) to have a new desired " +
        s"total $targetNumExecutors executors.")
      // 获取匹配的请求，并且有匹配的数据，则移除这些container（而且我猜测，这里的寻找匹配的请求，是在pending的队列中找的）
      val matchingRequests = amClient.getMatchingRequests(RM_REQUEST_PRIORITY, ANY_HOST, resource)
      if (!matchingRequests.isEmpty) {
        matchingRequests.iterator().next().asScala
          .take(numToCancel).foreach(amClient.removeContainerRequest)
      } else {
        logWarning("Expected to find pending requests, but found none.")
      }
      // <7>
    }
  }

def handleAllocatedContainers(allocatedContainers: Seq[Container]): Unit = {
    // 用来保存要使用的Containers
    val containersToUse = new ArrayBuffer[Container](allocatedContainers.size)
    // Match incoming requests by host 根据host进行匹配，匹配不成功的下面继续匹配
    val remainingAfterHostMatches = new ArrayBuffer[Container]
    for (allocatedContainer <- allocatedContainers) {
      matchContainerToRequest(allocatedContainer, allocatedContainer.getNodeId.getHost,
        containersToUse, remainingAfterHostMatches)
    }
    // Match remaining by rack 对上面按照host匹配不成功的进行机架匹配
    val remainingAfterRackMatches = new ArrayBuffer[Container]
    for (allocatedContainer <- remainingAfterHostMatches) {
      val rack = resolver.resolve(conf, allocatedContainer.getNodeId.getHost)
      matchContainerToRequest(allocatedContainer, rack, containersToUse,
        remainingAfterRackMatches)
    }
    // 通过上面的两次匹配可以看到，最优先使用的是 host -> rack -> *
    // Assign remaining that are neither node-local nor rack-local
    val remainingAfterOffRackMatches = new ArrayBuffer[Container]
    for (allocatedContainer <- remainingAfterRackMatches) {
      matchContainerToRequest(allocatedContainer, ANY_HOST, containersToUse,
        remainingAfterOffRackMatches)
    }
    // 三次匹配都没有成功的，就释放掉了
    if (!remainingAfterOffRackMatches.isEmpty) {
      logDebug(s"Releasing ${remainingAfterOffRackMatches.size} unneeded containers that were " +
        s"allocated to us")
      for (container <- remainingAfterOffRackMatches) {
        internalReleaseContainer(container)
      }
    }
    // 启动所有的container，这些container上经过上面过滤后的可以使用container
    runAllocatedContainers(containersToUse)
    logInfo("Received %d containers from YARN, launching executors on %d of them."
      .format(allocatedContainers.size, containersToUse.size))
  }

private def runAllocatedContainers(containersToUse: ArrayBuffer[Container]): Unit = {
    for (container <- containersToUse) {
      // executorIdConter 应该是同步的，用来生成新的executor的id
      executorIdCounter += 1
      val executorHostname = container.getNodeId.getHost
      val containerId = container.getId
      val executorId = executorIdCounter.toString
      assert(container.getResource.getMemory >= resource.getMemory)
      logInfo(s"Launching container $containerId on host $executorHostname " +
        s"for executor with ID $executorId")
      //<1>
      // 主要是为了更新各种计数
      def updateInternalState(): Unit = synchronized {
        // 正在运行的executor加一
        numExecutorsRunning.incrementAndGet()
        // 启动中的executor 减一，因为已经运行了，就不能处于启动中了
        numExecutorsStarting.decrementAndGet()
        //保存 executor -> container的关系 以及 container -> executorId的关系
        executorIdToContainer(executorId) = container
        containerIdToExecutorId(container.getId) = executorId
        // 记录 某个主机上都有哪些 containers
        val containerSet = allocatedHostToContainersMap.getOrElseUpdate(executorHostname,
          new HashSet[ContainerId])
        containerSet += containerId
        // 记录contain运行在那个主机上
        allocatedContainerToHostMap.put(containerId, executorHostname)
      }
      // <2>
      // 如果正在运行的executor的数量小于 想要启动的executor个数
      if (numExecutorsRunning.get < targetNumExecutors) {
        // 将启动的executor的计数加一
        // 这里为啥要加一呢？？？
        // 因为 接下来要启动 container了，没有启动起来之前则认为处于启动中，因此先加一，一旦启动完成，则调用上面的updateInternalState
        // 方法，在这个方法中会对运行中的executor加一，启动中的executor减一。
        numExecutorsStarting.incrementAndGet()
        // spark.yarn.launchContainers 配置的值
        if (launchContainers) {
          launcherPool.execute(new Runnable {
            override def run(): Unit = {
              try {
                // 调用 ExecutorRunnalbe方法来启动container
                new ExecutorRunnable(
                  Some(container),
                  conf,
                  sparkConf,
                  driverUrl,
                  executorId,
                  executorHostname,
                  executorMemory,
                  executorCores,
                  appAttemptId.getApplicationId.toString,
                  securityMgr,
                  localResources
                ).run()
                // 跟新各种计数
                updateInternalState()
              } catch {
                case e: Throwable =>
                  numExecutorsStarting.decrementAndGet()
                  if (NonFatal(e)) {
                    logError(s"Failed to launch executor $executorId on container $containerId", e)
                    // Assigned container should be released immediately
                    // to avoid unnecessary resource occupation.
                    // 启动失败，则释放掉这个container
                    amClient.releaseAssignedContainer(containerId)
                  } else {
                    throw e
                  }
              }
            }
          })
        } else {
          // For test only
          updateInternalState()
        }
      } else {
        logInfo(("Skip launching executorRunnable as runnning Excecutors count: %d " +
          "reached target Executors count: %d.").format(
          numExecutorsRunning.get, targetNumExecutors))
      }
    }
    // <3>
  }

private[yarn] def processCompletedContainers(completedContainers: Seq[ContainerStatus]): Unit = {
    // 循环完成的container
    for (completedContainer <- completedContainers) {
      // 获取containerId, 判断container是否在releasedContainers 中，获取 container所在的host
      val containerId = completedContainer.getContainerId
      val alreadyReleased = releasedContainers.remove(containerId)
      val hostOpt = allocatedContainerToHostMap.get(containerId)
      val onHostStr = hostOpt.map(host => s" on host: $host").getOrElse("")
      // <1>
      // 如果alreadyReleased为false，则表示releasedContainers 中没有这个container，或者是有但是删除失败
      val exitReason = if (!alreadyReleased) {
        // Decrement the number of executors running. The next iteration of
        // the ApplicationMaster's reporting thread will take care of allocating.
        numExecutorsRunning.decrementAndGet()
        logInfo("Completed container %s%s (state: %s, exit status: %s)".format(
          containerId,
          onHostStr,
          completedContainer.getState,
          completedContainer.getExitStatus))
        // 获取container的退出状态
        val exitStatus = completedContainer.getExitStatus
        // 根据退出状态来判断是否是由于Application的原因
        // 以及退出的原因
        // ContainerExitStatus.SUCCESS 表示是因为YARN事件，不是因为运行的job发生error而导致的
        // ContainerExitStatus.PREEMPTED 表示主机上的端口被占用了
        // VMEM_EXCEEDED_EXIT_CODE 表示 虚拟内存的问题
        // PMEM_EXCEEDED_EXIT_CODE 表示 物理内存的问题
        // 否则
        // 除了第一种和第二种，其他的认为都属于 Application的原因
        val (exitCausedByApp, containerExitReason) = exitStatus match {
          case ContainerExitStatus.SUCCESS =>
            (false, s"Executor for container $containerId exited because of a YARN event (e.g., " +
              "pre-emption) and not because of an error in the running job.")
          case ContainerExitStatus.PREEMPTED =>
            // Preemption is not the fault of the running tasks, since YARN preempts containers
            // merely to do resource sharing, and tasks that fail due to preempted executors could
            // just as easily finish on any other executor. See SPARK-8167.
            (false, s"Container ${containerId}${onHostStr} was preempted.")
          // Should probably still count memory exceeded exit codes towards task failures
          case VMEM_EXCEEDED_EXIT_CODE =>
            (true, memLimitExceededLogMessage(
              completedContainer.getDiagnostics,
              VMEM_EXCEEDED_PATTERN))
          case PMEM_EXCEEDED_EXIT_CODE =>
            (true, memLimitExceededLogMessage(
              completedContainer.getDiagnostics,
              PMEM_EXCEEDED_PATTERN))
          case _ =>
            // Enqueue the timestamp of failed executor
            failedExecutorsTimeStamps.enqueue(clock.getTimeMillis())
            (true, "Container marked as failed: " + containerId + onHostStr +
              ". Exit status: " + completedContainer.getExitStatus +
              ". Diagnostics: " + completedContainer.getDiagnostics)
        }
        // 如果是由于Application的原因退出的，则以警告记录日志，否则以info级别记录日志
        if (exitCausedByApp) {
          logWarning(containerExitReason)
        } else {
          logInfo(containerExitReason)
        }
        // 生成一个ExecutorExited 对象并返回给变量
        ExecutorExited(exitStatus, exitCausedByApp, containerExitReason)
      } else {
        // 如果 alreadyReleased 为true，则表示 container已经被kill掉了，在 internalReleaseContainer 方法中会操作
        ExecutorExited(completedContainer.getExitStatus, exitCausedByApp = false,
          s"Container $containerId exited from explicit termination request.")
      }
      // <2>
      for {
        host <- hostOpt
        // 根据host获取上面所运行的container
        containerSet <- allocatedHostToContainersMap.get(host)
      } {
        // 将container从host所包含的container集合中删除，这样host上的container集合就含有这个container了
        containerSet.remove(containerId)
        // 删除完成后，如果 container 集合列表空了，则说明 host上只运行了这一个contianer，则删除host与container的对应关系，否则就更新一下
        if (containerSet.isEmpty) {
          allocatedHostToContainersMap.remove(host)
        } else {
          allocatedHostToContainersMap.update(host, containerSet)
        }
        // 将container 到 host的映射关系也删除
        allocatedContainerToHostMap.remove(containerId)
      }
      // 移除 container到executor的对应关系
      containerIdToExecutorId.remove(containerId).foreach { eid =>
        // 将executor到container的对应关系也删除
        executorIdToContainer.remove(eid)
        // 从 pendingLossReasonRequests 移除 executor
        pendingLossReasonRequests.remove(eid) match {
          case Some(pendingRequests) =>
            // Notify application of executor loss reasons so it can decide whether it should abort
            pendingRequests.foreach(_.reply(exitReason))
          case None =>
            releasedExecutorLossReasons.put(eid, exitReason)
        }
        // <3>
        // 如果没有被kill掉
        if (!alreadyReleased) {
          // The executor could have gone away (like no route to host, node failure, etc)
          // Notify backend about the failure of the executor
          // container非异常释放的计数加一
          numUnexpectedContainerRelease += 1
          // 发送删除 execurot的请求
          driverRef.send(RemoveExecutor(eid, exitReason))
        }
        // <4>
      }
    }
  }

def register(
      driverUrl: String,
      driverRef: RpcEndpointRef,
      conf: YarnConfiguration,
      sparkConf: SparkConf,
      uiAddress: Option[String],
      uiHistoryAddress: String,
      securityMgr: SecurityManager,
      localResources: Map[String, LocalResource]
    ): YarnAllocator = {
    amClient = AMRMClient.createAMRMClient()
    amClient.init(conf)
    amClient.start()
    this.uiHistoryAddress = uiHistoryAddress
    val trackingUrl = uiAddress.getOrElse {
      if (sparkConf.get(ALLOW_HISTORY_SERVER_TRACKING_URL)) uiHistoryAddress else ""
    }
    logInfo("Registering the ApplicationMaster")
    synchronized {
      amClient.registerApplicationMaster(Utils.localHostName(), 0, trackingUrl)
      registered = true
    }
    new YarnAllocator(driverUrl, driverRef, conf, sparkConf, amClient, getAttemptId(), securityMgr,
      localResources, new SparkRackResolver())
  }

abstract class RDD[T: ClassTag](
    @transient private var _sc: SparkContext,
    @transient private var deps: Seq[Dependency[_]]
  ) extends Serializable with Logging {

def this(@transient oneParent: RDD[_]) =
    this(oneParent.context, List(new OneToOneDependency(oneParent)))

def compute(split: Partition, context: TaskContext): Iterator[T] -- 计算给定的分区，返回一个迭代器
protected def getPartitions: Array[Partition] -- 返回RDD的所有分区
protected def getDependencies: Seq[Dependency[_]] = deps -- 返回RDD的到父类RDD的所有依赖

val sc = new SparkContext("master", "testApplication");
sc.hadoopFile("path", 5).map(_ => 1).count()

def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum  
//Utils.getIteratorSize方法，在之后的代码块中展示，其实现就是根据参数的Iterator，计算一下这个迭代器中的数据个数（猜测迭代器最终是RDD分区的迭代器）
//这里的runJob的定义是
def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U): Array[U] = {
    runJob(rdd, func, 0 until rdd.partitions.length)
}
//然后又指向
def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U, partitions: Seq[Int]): Array[U] = {
    val cleanedFunc = clean(func)
    runJob(rdd, (ctx: TaskContext, it: Iterator[T]) => cleanedFunc(it), partitions)
}
//然后继续指向
def runJob[T, U: ClassTag](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int]): Array[U] = {
    val results = new Array[U](partitions.size)
    runJob[T, U](rdd, func, partitions, (index, res) => results(index) = res)
    results
  }
//继续指向
def runJob[T, U: ClassTag](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], resultHandler: (Int, U) => Unit): Unit = {
    ...
    dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
    progressBar.foreach(_.finishAll())
    rdd.doCheckpoint()
  }

def runJob[T, U](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], callSite: CallSite, resultHandler: (Int, U) => Unit, properties: Properties): Unit = {
    val start = System.nanoTime
    val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
    ...
  }

def submitJob[T, U](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], callSite: CallSite, resultHandler: (Int, U) => Unit, properties: Properties):JobWaiter[U] = {
    ...
    val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
    eventProcessLoop.post(JobSubmitted(jobId, rdd, func2, partitions.toArray, callSite, waiter, SerializationUtils.clone(properties)))
    waiter
  }

private[scheduler] val eventProcessLoop = new DAGSchedulerEventProcessLoop(this)

private[scheduler] class DAGSchedulerEventProcessLoop(dagScheduler: DAGScheduler) extends EventLoop[DAGSchedulerEvent]("dag-scheduler-event-loop") with Logging

while (!stopped.get) {
  val event = eventQueue.take()
  try {
    onReceive(event)
  } catch {
  }
}

private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
    case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>
      dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)
    ...
}

private[scheduler] def handleJobSubmitted(jobId: Int,
  finalRDD: RDD[_],
  func: (TaskContext, Iterator[_]) => _,
  partitions: Array[Int],
  callSite: CallSite,
  listener: JobListener,
  properties: Properties) {
	var finalStage: ResultStage = null
	try {
	  // New stage creation may throw an exception if, for example, jobs are run on a
	  // HadoopRDD whose underlying HDFS files have been deleted.
	  finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
	} catch {
	  case e: Exception =>
	    logWarning("Creating new stage failed due to exception - job: " + jobId, e)
	    listener.jobFailed(e)
	    return
	}
	val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)
	
	jobIdToActiveJob(jobId) = job
	activeJobs += job
	finalStage.setActiveJob(job)
	val stageIds = jobIdToStageIds(jobId).toArray
	val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo))
	listenerBus.post(
	  SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties))
	submitStage(finalStage)
}