override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    case RequestSubmitDriver(description) =>
      if (state != RecoveryState.ALIVE) {
        val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
          "Can only accept driver submissions in ALIVE state."
        context.reply(SubmitDriverResponse(self, false, None, msg))
      } else {
        logInfo("Driver submitted " + description.command.mainClass)
        val driver = createDriver(description)
        // 持久化Driver 用于master recovery 时恢复Driver
        persistenceEngine.addDriver(driver)
        // 注册 Driver
        waitingDrivers += driver
        drivers.add(driver)
        // launch Driver 和 Executor
        schedule()
        context.reply(SubmitDriverResponse(self, true, Some(driver.id),
          s"Driver successfully submitted as ${driver.id}"))
      }
    //忽略
    ...
  }

private def schedule(): Unit = {
  if (state != RecoveryState.ALIVE) {
    return
  }
  // 打乱Worker顺序，避免Driver过度集中
  val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
  val numWorkersAlive = shuffledAliveWorkers.size
  var curPos = 0
  // 变量Worker如果Work节点剩余内存和core足够，启动Driver
  for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
    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) {
        //发送LauncherDriver消息给Worker节点  
        launchDriver(worker, driver)
        waitingDrivers -= driver
        launched = true
      }
      curPos = (curPos + 1) % numWorkersAlive
    }
  }
  // 在work节点启动Executor，该内容在《spark4-启动Executor》讲解
  startExecutorsOnWorkers()
}

private def launchDriver(worker: WorkerInfo, driver: DriverInfo) {
    logInfo("Launching driver " + driver.id + " on worker " + worker.id)
    worker.addDriver(driver)
    driver.worker = Some(worker)
    // 调用RpcEndPointRef，发送LauncherDriver消息给Worker节点
    // 此Work是 schedule 随机选择的
    worker.endpoint.send(LaunchDriver(driver.id, driver.desc))
    driver.state = DriverState.RUNNING
  }

override def receive: PartialFunction[Any, Unit] = synchronized {
   // 忽略
   //...
   case LaunchDriver(driverId, driverDesc) =>
     logInfo(s"Asked to launch driver $driverId")
     //封装Driver信息为DriverRunner
     val driver = new DriverRunner(
       conf,
       driverId,
       workDir,
       sparkHome,
       driverDesc.copy(command = Worker.maybeUpdateSSLSettings(driverDesc.command, conf)),
       self,
       workerUri,
       securityMgr)
     drivers(driverId) = driver
     //启动Driver
     driver.start()

     coresUsed += driverDesc.cores
     memoryUsed += driverDesc.mem
 }

/** 启动一个线程，去运行 这个driver. */
 private[worker] def start() = {
   new Thread("DriverRunner for " + driverId) {
     override def run() {
       var shutdownHook: AnyRef = null
       try {
         shutdownHook = ShutdownHookManager.addShutdownHook { () =>
           logInfo(s"Worker shutting down, killing driver $driverId")
           kill()
         }

         // 从其他环境中下载driver的jars 并在本地运行 driver
         val exitCode = prepareAndRunDriver()

         // 根据exitCode，设置 finalState 
         finalState = if (exitCode == 0) {
           Some(DriverState.FINISHED)
         } else if (killed) {
           Some(DriverState.KILLED)
         } else {
           Some(DriverState.FAILED)
         }
       } catch {
         case e: Exception =>
           kill()
           finalState = Some(DriverState.ERROR)
           finalException = Some(e)
       } finally {
         if (shutdownHook != null) {
           ShutdownHookManager.removeShutdownHook(shutdownHook)
         }
       }

       // 一旦Driver运行完毕, 
       // work会 执行org.apache.spark.deploy.worker.Worker中的 handleDriverStateChanged()
       //  
       // - work会向Master发送消息driverStateChanged
       // - 释放相关资源
       worker.send(DriverStateChanged(driverId, finalState.get, finalException))
     }
   }.start()
 }

private[worker] def prepareAndRunDriver(): Int = {
  // 下载Driver jar 到Worker本地，即本示例中的 /path/to/examples.jar
  val driverDir = createWorkingDirectory()
  val localJarFilename = downloadUserJar(driverDir)
  // 替换参数中的workerUrl和localJarFilename
  def substituteVariables(argument: String): String = argument match {
    case "{{WORKER_URL}}" => workerUrl
    case "{{USER_JAR}}" => localJarFilename
    case other => other
  }

  // 将Driver中的参数组织为Linux命令
  // 通过java执行组织好的命令
  val builder = CommandUtils.buildProcessBuilder(driverDesc.command, securityManager,
    driverDesc.mem, sparkHome.getAbsolutePath, substituteVariables)
  // 这一步就是启动Driver，即执行 /parh/to/examples.jar中的main方法.
  runDriver(builder, driverDir, driverDesc.supervise)
}

// 解析执行下面的四个参数: 需要注意的是 childMainClass。
// childMainClass 根据spark的模式 会分为多种情况: 
// - org.apache.spark.deploy.Client 
// - org.apache.spark.deploy.yarn.Client
// - org.apache.spark.deploy.rest.RestSubmissionClient
// - 用户自定义的app类
//
// 在本例子中为 org.apache.spark.deploy.Client 
val (childArgs, childClasspath, sysProps, childMainClass) = prepareSubmitEnvironment(args)
 
//声明doRunMain()方法
def doRunMain(): Unit = {
  if (args.proxyUser != null) {
    val proxyUser = UserGroupInformation.createProxyUser(args.proxyUser,
      UserGroupInformation.getCurrentUser())
    try {
      proxyUser.doAs(new PrivilegedExceptionAction[Unit]() {
        override def run(): Unit = {
          runMain(childArgs, childClasspath, sysProps, childMainClass, args.verbose)
        }
      })
    } catch {
     //忽略
    }
  } else {
    // 一般默认走这里
    runMain(childArgs, childClasspath, sysProps, childMainClass, args.verbose)
  }
}

if (args.isStandaloneCluster && args.useRest) {
  try {
    printStream.println("Running Spark using the REST application submission protocol.")
    doRunMain()
  } catch {
   //忽略
  }
// In all other modes, just run the main class as prepared
} else {
  doRunMain()
}

private def runMain(
      childArgs: Seq[String],
      childClasspath: Seq[String],
      sysProps: Map[String, String],
      childMainClass: String,
      verbose: Boolean): Unit = {
   //...
    val loader =
      if (sysProps.getOrElse("spark.driver.userClassPathFirst", "false").toBoolean) {
        new ChildFirstURLClassLoader(new Array[URL](0),
          Thread.currentThread.getContextClassLoader)
      } else {
        new MutableURLClassLoader(new Array[URL](0),
          Thread.currentThread.getContextClassLoader)
      }
    Thread.currentThread.setContextClassLoader(loader)

    // 将jar 以 URL 的形式加入 loader
    for (jar <- childClasspath) {
      addJarToClasspath(jar, loader)
    }
    // 环境变量
    for ((key, value) <- sysProps) {
      System.setProperty(key, value)
    }
    // 反射 主类 并启动 主类
    var mainClass: Class[_] = null
    try {
      mainClass = Utils.classForName(childMainClass)
    } catch {
      //忽略
    }
    // 忽略
    
    val mainMethod = mainClass.getMethod("main", new Array[String](0).getClass)
    if (!Modifier.isStatic(mainMethod.getModifiers)) {
      throw new IllegalStateException("The main method in the given main class must be static")
    }
    //忽略
    
    // 反射启动主类
    try {
      mainMethod.invoke(null, childArgs.toArray)
    } catch {
      // 忽略
    }
  }

object Client {
  def main(args: Array[String]) {
    //....
    
    val conf = new SparkConf()
    val driverArgs = new ClientArguments(args)
    if (!conf.contains("spark.rpc.askTimeout")) {
      conf.set("spark.rpc.askTimeout", "10s")
    }
    Logger.getRootLogger.setLevel(driverArgs.logLevel)
	//创建PRC_ENV
    val rpcEnv =
      RpcEnv.create("driverClient", Utils.localHostName(), 0, conf, new SecurityManager(conf))
    // 获取与Master通信的EndPoint
    val masterEndpoints = driverArgs.masters.map(RpcAddress.fromSparkURL).
      map(rpcEnv.setupEndpointRef(_, Master.ENDPOINT_NAME))
	//注册ClientEndpoint
    rpcEnv.setupEndpoint("client", new ClientEndpoint(rpcEnv, driverArgs, masterEndpoints, conf))

    rpcEnv.awaitTermination()
  }
}

override def onStart(): Unit = {
    driverArgs.cmd match {
      case "launch" =>
        //Spark使用**DriverWrapper**启动用户APP的main函数，而不是直接启动，
        // 这是为了Driver程序和启动Driver的Worker程序共命运(源码注释中称为**share fate**).
        // 即如果此Worker挂了，对应的Driver也会停止。
        val mainClass = "org.apache.spark.deploy.worker.DriverWrapper"

        val classPathConf = "spark.driver.extraClassPath"
        val classPathEntries = sys.props.get(classPathConf).toSeq.flatMap { cp =>
          cp.split(java.io.File.pathSeparator)
        }

        val libraryPathConf = "spark.driver.extraLibraryPath"
        val libraryPathEntries = sys.props.get(libraryPathConf).toSeq.flatMap { cp =>
          cp.split(java.io.File.pathSeparator)
        }

        val extraJavaOptsConf = "spark.driver.extraJavaOptions"
        val extraJavaOpts = sys.props.get(extraJavaOptsConf)
          .map(Utils.splitCommandString).getOrElse(Seq.empty)
        val sparkJavaOpts = Utils.sparkJavaOpts(conf)
        val javaOpts = sparkJavaOpts ++ extraJavaOpts
        // 运行命令
        //  这里的mainClass为org.apache.spark.deploy.worker.DriverWrapper
        val command = new Command(mainClass,
          Seq("{{WORKER_URL}}", "{{USER_JAR}}", driverArgs.mainClass) ++ driverArgs.driverOptions,
          sys.env, classPathEntries, libraryPathEntries, javaOpts)

        val driverDescription = new DriverDescription(
          driverArgs.jarUrl,
          driverArgs.memory,
          driverArgs.cores,
          driverArgs.supervise,
          command)
        // 给Master发送RequestSubmitDriver消息，注册Driver
        ayncSendToMasterAndForwardReply[SubmitDriverResponse](
          RequestSubmitDriver(driverDescription))

      case "kill" =>
        val driverId = driverArgs.driverId
        ayncSendToMasterAndForwardReply[KillDriverResponse](RequestKillDriver(driverId))
    }
  }

private[spark] case class Command(
    mainClass: String, // DriverWrapper
    arguments: Seq[String], 
    environment: Map[String, String],
    classPathEntries: Seq[String],
    libraryPathEntries: Seq[String],
    javaOpts: Seq[String]) {
}

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    case RegisterWorker(
        id, workerHost, workerPort, workerRef, cores, memory, workerWebUiUrl) =>

    case RequestSubmitDriver(description) =>
      if (state != RecoveryState.ALIVE) {
        // 判断 Master 的运行状态，非 ALIVE 则 报错。
        val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
          "Can only accept driver submissions in ALIVE state."
        context.reply(SubmitDriverResponse(self, false, None, msg))
      } else {
        logInfo("Driver submitted " + description.command.mainClass)
        val driver = createDriver(description)
        // 持久化Driver 用于master recovery 时恢复Driver
        persistenceEngine.addDriver(driver)
        // 注册 Driver
        waitingDrivers += driver
        drivers.add(driver)
        // launch Driver 和 Executor，这个下一章分析
        schedule()
        context.reply(SubmitDriverResponse(self, true, Some(driver.id),
          s"Driver successfully submitted as ${driver.id}"))
      }

    case RequestKillDriver(driverId) =>
     //...

    case RequestDriverStatus(driverId) =>
     //...

    case RequestMasterState =>
     //...

    case BoundPortsRequest =>
     //...

    case RequestExecutors(appId, requestedTotal) =>
     //...

    case KillExecutors(appId, executorIds) =>
     //...
  }