spark源码学习（二）---Master源码分析(2)-master内组件状态改变机制

本篇学习master里面几个组件状态改变时发生的事情，自己写自己看，不对之处请拍砖。

首先，来看driver的状态改变：

case DriverStateChanged(driverId, state, exception) => {
      state match {
        //如果driver的状态是ERROR，Finished，killed，或者Failed，那么移除该driver信息
        case DriverState.ERROR | DriverState.FINISHED | DriverState.KILLED | DriverState.FAILED =>
          removeDriver(driverId, state, exception)
        case _ =>
          throw new Exception(s"Received unexpected state update for driver $driverId: $state")
      }
    }
           

当driver的状态编程上面的状态的时候，会执行remover driver，remove的操作基本上就是移除内存中的driver，worker，并修改状态信息的操作，remove的代码如下：

private def removeDriver(
      driverId: String,
      finalState: DriverState,
      exception: Option[Exception]) {
    //首先，查看是否存在该driver，如果不存在日志警告；如果存在该driver，将drivers中的该driver移除
    drivers.find(d => d.id == driverId) match {
      case Some(driver) =>
        logInfo(s"Removing driver: $driverId")
        //从set中移除该driver
        drivers -= driver
        
        if (completedDrivers.size >= RETAINED_DRIVERS) {
          val toRemove = math.max(RETAINED_DRIVERS / 10, 1)
          completedDrivers.trimStart(toRemove)
        }
        //将移除的driver加入到完成的driver记录容器中
        completedDrivers += driver
        //移除driver的持久化信息
        persistenceEngine.removeDriver(driver)
        //更改driver的状态
        driver.state = finalState
        driver.exception = exception
        //移除该driver对应的worker
        driver.worker.foreach(w => w.removeDriver(driver))
        schedule()
      case None =>
        logWarning(s"Asked to remove unknown driver: $driverId")
    }
  }
}
           

可以发现master类中RequestKillDriver方法：master收到要求kill掉该driver的时候会触发driver状态改变   

case RequestKillDriver(driverId) => {
      if (state != RecoveryState.ALIVE) {
        val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
          s"Can only kill drivers in ALIVE state."
        context.reply(KillDriverResponse(self, driverId, success = false, msg))
      } else {
        logInfo("Asked to kill driver " + driverId)
        val driver = drivers.find(_.id == driverId)
        driver match {
          case Some(d) =>
            if (waitingDrivers.contains(d)) {
              waitingDrivers -= d
              self.send(DriverStateChanged(driverId, DriverState.KILLED, None))
            } else {
              // We just notify the worker to kill the driver here. The final bookkeeping occurs
              // on the return path when the worker submits a state change back to the master
              // to notify it that the driver was successfully killed.
              d.worker.foreach { w =>
                w.endpoint.send(KillDriver(driverId))
              }
            }
            // TODO: It would be nice for this to be a synchronous response
            val msg = s"Kill request for $driverId submitted"
            logInfo(msg)
            context.reply(KillDriverResponse(self, driverId, success = true, msg))
          case None =>
            val msg = s"Driver $driverId has already finished or does not exist"
            logWarning(msg)
            context.reply(KillDriverResponse(self, driverId, success = false, msg))
        }
      }
    }
           

下面分析executor的状态改变：

case ExecutorStateChanged(appId, execId, state, message, exitStatus) => {
      //找到exexutor对应的application，通过获取executor的信息
      val execOption = idToApp.get(appId).flatMap(app => app.executors.get(execId))
      
      execOption match {
      //如果找到了该executor  
      case Some(exec) => {
          val appInfo = idToApp(appId)
          //修改状态为当前的新状态
          val oldState = exec.state
          exec.state = state
          //如果该executor正在运行，那么，重新设置APP的重试次数
          if (state == ExecutorState.RUNNING) {
            assert(oldState == ExecutorState.LAUNCHING,
              s"executor $execId state transfer from $oldState to RUNNING is illegal")
            appInfo.resetRetryCount()
          }
          //向driver发送executor的更新信息
          exec.application.driver.send(ExecutorUpdated(execId, state, message, exitStatus))
          //如果executor状态是finished，则从worker和APP中移除该driver
          if (ExecutorState.isFinished(state)) {
            // Remove this executor from the worker and app
            logInfo(s"Removing executor ${exec.fullId} because it is $state")
            // If an application has already finished, preserve its
            // state to display its information properly on the UI
            // 如果AP没有完成，需要保持状态以便将信息正确输出到ui，那么不从APP中移除executor，只移除worker中的executor
            if (!appInfo.isFinished) {
              appInfo.removeExecutor(exec)
            }
            exec.worker.removeExecutor(exec)
            
            val normalExit = exitStatus == Some(0)
            // Only retry certain number of times so we don't go into an infinite loop.
            //如果executor退出状态是非正常的
            if (!normalExit) {
              //如果没有达到app重试次数最大值，那么会重新执行调度Schedule方法
              if (appInfo.incrementRetryCount() < ApplicationState.MAX_NUM_RETRY) {
                schedule()
              } else {
                //否则 进行removeApplication操作
                val execs = appInfo.executors.values
                if (!execs.exists(_.state == ExecutorState.RUNNING)) {
                  logError(s"Application ${appInfo.desc.name} with ID ${appInfo.id} failed " +
                    s"${appInfo.retryCount} times; removing it")
                  removeApplication(appInfo, ApplicationState.FAILED)
                }
              }
            }
          }
        }
        case None =>
          logWarning(s"Got status update for unknown executor $appId/$execId")
      }
    }