def start() {

  // Just launch an actor; it will call back into the listener.

  actor = actorSystem.actorOf(Props(new ClientActor))

}

override def preStart() {

  logInfo("Connecting to master " + masterUrl)

  try {

    master = context.actorFor(Master.toAkkaUrl(masterUrl))

    masterAddress = master.path.address

    master ! RegisterApplication(appDescription) //向master注册application

    context.system.eventStream.subscribe(self, classOf[RemoteClientLifeCycleEvent])

    context.watch(master)  // Doesn't work with remote actors, but useful for testing

  } catch {

    case e: Exception =>

      logError("Failed to connect to master", e)

      markDisconnected()

      context.stop(self)

  }

}

case RegisterApplication(description) => {

  logInfo("Registering app " + description.name)

  val app = addApplication(description, sender)

  logInfo("Registered app " + description.name + " with ID " + app.id)

  waitingApps += app

  context.watch(sender)  // This doesn't work with remote actors but helps for testing

  sender ! RegisteredApplication(app.id)

  schedule()

}

def schedule() {

  if (spreadOutApps) {

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

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

                                 .filter(canUse(app, _)).sortBy(_.coresFree).reverse

      val numUsable = usableWorkers.length

      val assigned = new Array[Int](numUsable) // Number of cores to give on each node

      var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)

      var pos = 0

      while (toAssign > 0) {

        if (usableWorkers(pos).coresFree - assigned(pos) > 0) {

          toAssign -= 1

          assigned(pos) += 1

        }

        pos = (pos + 1) % numUsable

      }

      // Now that we've decided how many cores to give on each node, let's actually give them

      for (pos <- 0 until numUsable) {

        if (assigned(pos) > 0) {

          val exec = app.addExecutor(usableWorkers(pos), assigned(pos))

          launchExecutor(usableWorkers(pos), exec, app.desc.sparkHome)

          app.state = ApplicationState.RUNNING

        }

      }

    }

  } else {

    // Pack each app into as few nodes as possible until we've assigned all its cores

    for (worker <- workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) {

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

        if (canUse(app, worker)) {

          val coresToUse = math.min(worker.coresFree, app.coresLeft)

          if (coresToUse > 0) {

            val exec = app.addExecutor(worker, coresToUse)

            launchExecutor(worker, exec, app.desc.sparkHome)

            app.state = ApplicationState.RUNNING

          }

        }

      }

    }

  }

}

def launchExecutor(worker: WorkerInfo, exec: ExecutorInfo, sparkHome: String) {

  worker.addExecutor(exec)

  worker.actor ! LaunchExecutor(exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory, sparkHome)

  exec.application.driver ! ExecutorAdded(exec.id, worker.id, worker.host, exec.cores, exec.memory)

}

case LaunchExecutor(appId, execId, appDesc, cores_, memory_, execSparkHome_) =>

  val manager = new ExecutorRunner(

    appId, execId, appDesc, cores_, memory_, self, workerId, ip, new File(execSparkHome_), workDir)

  executors(appId + "/" + execId) = manager

  manager.start()

  coresUsed += cores_

  memoryUsed += memory_

  master ! ExecutorStateChanged(appId, execId, ExecutorState.RUNNING, None, None)

def start() {

  workerThread = new Thread("ExecutorRunner for " + fullId) {

    override def run() { fetchAndRunExecutor() }

  }

  workerThread.start()

  // Shutdown hook that kills actors on shutdown.

  ...

}

def fetchAndRunExecutor() {

  try {

    // Create the executor's working directory

    val executorDir = new File(workDir, appId + "/" + execId)

    if (!executorDir.mkdirs()) {

      throw new IOException("Failed to create directory " + executorDir)

    }

    // Launch the process

    val command = buildCommandSeq()

    val builder = new ProcessBuilder(command: _*).directory(executorDir)

    val env = builder.environment()

    for ((key, value) <- appDesc.command.environment) {

      env.put(key, value)

    }

    env.put("SPARK_MEM", memory.toString + "m")

    // In case we are running this from within the Spark Shell, avoid creating a "scala"

    // parent process for the executor command

    env.put("SPARK_LAUNCH_WITH_SCALA", "0")

    process = builder.start()

    // Redirect its stdout and stderr to files

    redirectStream(process.getInputStream, new File(executorDir, "stdout"))

    redirectStream(process.getErrorStream, new File(executorDir, "stderr"))

    // Wait for it to exit; this is actually a bad thing if it happens, because we expect to run

    // long-lived processes only. However, in the future, we might restart the executor a few

    // times on the same machine.

    val exitCode = process.waitFor()

    val message = "Command exited with code " + exitCode

    worker ! ExecutorStateChanged(appId, execId, ExecutorState.FAILED, Some(message),

                                  Some(exitCode))

  } catch {

    case interrupted: InterruptedException =>

      logInfo("Runner thread for executor " + fullId + " interrupted")

    case e: Exception => {

      logError("Error running executor", e)

      if (process != null) {

        process.destroy()

      }

      val message = e.getClass + ": " + e.getMessage

      worker ! ExecutorStateChanged(appId, execId, ExecutorState.FAILED, Some(message), None)

    }

  }

}

def removeApplication(app: ApplicationInfo) {

  if (apps.contains(app)) {

    logInfo("Removing app " + app.id)

    apps -= app

    idToApp -= app.id

    actorToApp -= app.driver

    addressToWorker -= app.driver.path.address

    completedApps += app   // Remember it in our history

    waitingApps -= app

    for (exec <- app.executors.values) {

      exec.worker.removeExecutor(exec)

      exec.worker.actor ! KillExecutor(exec.application.id, exec.id)

    }

    app.markFinished(ApplicationState.FINISHED)  // TODO: Mark it as FAILED if it failed

    schedule()

  }

}

case KillExecutor(appId, execId) =>

  val fullId = appId + "/" + execId

  executors.get(fullId) match {

    case Some(executor) =>

      logInfo("Asked to kill executor " + fullId)

      executor.kill()

    case None =>

      logInfo("Asked to kill unknown executor " + fullId)

  }

def kill() {

  if (workerThread != null) {

    workerThread.interrupt()

    workerThread = null

    if (process != null) {

      logInfo("Killing process!")

      process.destroy()

      process.waitFor()

    }

    worker ! ExecutorStateChanged(appId, execId, ExecutorState.KILLED, None, None)

    Runtime.getRuntime.removeShutdownHook(shutdownHook)

  }

}