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)
  }
}