core/src/main/scala/spark/deploy/master/JobInfo.scala


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

package spark.deploy.master

import spark.deploy.JobDescription
import java.util.Date
import akka.actor.ActorRef
import scala.collection.mutable

class JobInfo(val id: String, val desc: JobDescription, val submitDate: Date, val actor: ActorRef) {
  var state = JobState.WAITING
  var executors = new mutable.HashMap[Int, ExecutorInfo]
  var coresGranted = 0

  private var nextExecutorId = 0

  def newExecutorId(): Int = {
    val id = nextExecutorId
    nextExecutorId += 1
    id
  }

  def addExecutor(worker: WorkerInfo, cores: Int): ExecutorInfo = {
    val exec = new ExecutorInfo(newExecutorId(), this, worker, cores, desc.memoryPerSlave)
    executors(exec.id) = exec
    coresGranted += cores
    exec
  }

  def removeExecutor(exec: ExecutorInfo) {
    executors -= exec.id
    coresGranted -= exec.cores
  }

  def coresLeft: Int = desc.cores - coresGranted
}