[Spark源码剖析]Task的调度与执行源码剖析

//< DAGScheduler调用该taskScheduler.submitTasks提交一个stage对应的taskSet，一个taskSet包含多个task
TaskSchedulerImpl.submitTasks(taskSet: TaskSet)
    //< TaskScheduler（实际上是TaskSchedulerImpl）为DAGScheduler提交的每个taskSet创建一个对应的TaskSetManager对象，TaskSetManager用于调度同一个taskSet中的task
    val manager = TaskSchedulerImpl.createTaskSetManager(taskSet, maxTaskFailures)
    //< 将新创建的manager加入到调度树中，调度树由SchedulableBulider维护。有FIFO、Fair两种实现
    SchedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
        //< 触发调用CoarseGrainedSchedulerBackend.reviveOffers()，它将通过发送事件触发makeOffers方法调用
        CoarseGrainedSchedulerBackend.reviveOffers()
            //< 此处为发送ReviveOffers事件
            driverEndpoint.send(ReviveOffers)
        //< 此处为接收事件并处理
        CoarseGrainedSchedulerBackend.receive 
            CoarseGrainedSchedulerBackend.makeOffers
                //< 查找各个节点空闲资源（这里是cores），并返回要在哪些节点上启动哪些tasks的对应关系，用Seq[Seq[TaskDescription]]表示
                TaskSchedulerImpl.resourceOffers
            //< 启动对应task
            CoarseGrainedSchedulerBackend.launchTasks
                executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))

override def submitTasks(taskSet: TaskSet) {
  val tasks = taskSet.tasks
  logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
  this.synchronized {
    //< 为stage对应的taskSet创建TaskSetManager对象
    val manager = createTaskSetManager(taskSet, maxTaskFailures)
    //< 建立taskset与TaskSetManager的对应关系
    activeTaskSets(taskSet.id) = manager

    //< TaskSetManager会被放入调度池（Pool）当中。
      schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)

    //< 设置定时器，若task还没启动，则一直输出未分配到资源报警（输出警告日志）
    if (!isLocal && !hasReceivedTask) { 
      starvationTimer.scheduleAtFixedRate(new TimerTask() {
        override def run() {
          if (!hasLaunchedTask) {
            logWarning("Initial job has not accepted any resources; " +
              "check your cluster UI to ensure that workers are registered " +
              "and have sufficient resources")
          } else {
            this.cancel()
          }
        }
      }, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)
    }
    hasReceivedTask = true
  }

  //< 将处触发调用SchedulerBackend.makeOffers来为tasks分配资源，调度任务
  backend.reviveOffers()
}

  override def reviveOffers() {
    driverEndpoint.send(ReviveOffers)
  }
  
  override def receive: PartialFunction[Any, Unit] = {
  //< 此处省略n行代码
   
  case ReviveOffers =>
    makeOffers()

  //< 此处省略n行代码
  }

    def makeOffers() {
      launchTasks(scheduler.resourceOffers(executorDataMap.map { case (id, executorData) =>
        new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
      }.toSeq))
    }

//< 代表一个executor上的可用资源（这里仅可用cores）
private[spark]
case class WorkerOffer(executorId: String, host: String, cores: Int)

def resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
    //< 标记每个slave为alive并记录它们的hostname
    var newExecAvail = false
    //< 此处省略更新executor，host，rack信息代码；这里会根据是否有新的executor更新newExecAvail的值
  
    //< 为了负载均衡，打乱offers顺序，Random.shuffle用于将一个集合中的元素打乱
    val shuffledOffers = Random.shuffle(offers)
    //< 事先创建好用于存放要在各个worker上launch的 List[workerId, ArrayBuffer[TaskDescription]]。
    //< 由于task要使用的cores并不一定为1，所以每个worker上要launch得task并不一定等于可用的cores数
    val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))
    //< 每个executor上可用的cores
    val availableCpus = shuffledOffers.map(o => o.cores).toArray

    //< 返回排序过的TaskSet队列，有FIFO及Fair两种排序规则，默认为FIFO，可通过配置修改
    val sortedTaskSets = rootPool.getSortedTaskSetQueue
    for (taskSet <- sortedTaskSets) {
      //< 如果有新的executor added，更新TaskSetManager可用的executor
      if (newExecAvail) {
        taskSet.executorAdded()
      }
    }

    var launchedTask = false
    //< 依次取出排序过的taskSet列表中的taskSet；
    //< 对于每个taskSet，取出其tasks覆盖的所有locality，从高到低依次遍历每个等级的locality；
    //< 取出了taskSet及本次要处理的locality后，根据该taskSet及locality遍历所有可用的worker，找出可以在各个worker上启动的task，加到tasks:Seq[Seq[TaskDescription]]中
    for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
      do {
        //< 获取tasks，tasks代表要在哪些worker上启动哪些tasks
        launchedTask = resourceOfferSingleTaskSet(
            taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
      } while (launchedTask)
    }

    if (tasks.size > 0) {
      hasLaunchedTask = true
    }
    return tasks
  }

  override def addSchedulable(schedulable: Schedulable) {
    require(schedulable != null)
    //< 当我们添加一个元素的时候，它会添加到队列的尾部，当我们获取一个元素时，它会返回队列头部的元素
    schedulableQueue.add(schedulable)
    schedulableNameToSchedulable.put(schedulable.name, schedulable)
    schedulable.parent = this
  }

  var taskSetSchedulingAlgorithm: SchedulingAlgorithm = {
    schedulingMode match {
      case SchedulingMode.FAIR =>
        new FairSchedulingAlgorithm()
      case SchedulingMode.FIFO =>
        new FIFOSchedulingAlgorithm()
    }
  }

  //< 利用排序算法taskSetSchedulingAlgorithm先对以本pool作为父pool的子pools做排序，再对排序后的pool中的每个TaskSetManager排序；
  //< 得到最终排好序的 ArrayBuffer[TaskSetManager]
  override def getSortedTaskSetQueue: ArrayBuffer[TaskSetManager] = {
    var sortedTaskSetQueue = new ArrayBuffer[TaskSetManager]
    val sortedSchedulableQueue =
      schedulableQueue.toSeq.sortWith(taskSetSchedulingAlgorithm.comparator)
    //< FIFO不会调到这里，直接走到下面的return  
    for (schedulable <- sortedSchedulableQueue) {
      sortedTaskSetQueue ++= schedulable.getSortedTaskSetQueue
    }
    sortedTaskSetQueue
  }

private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {
  override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
    val priority1 = s1.priority
    val priority2 = s2.priority
    var res = math.signum(priority1 - priority2)
    if (res == 0) {
      val stageId1 = s1.stageId
      val stageId2 = s2.stageId
      res = math.signum(stageId1 - stageId2)
    }
    if (res < 0) {
      true
    } else {
      false
    }
  }
}

private[spark] trait Schedulable {
  var parent: Pool
  // child queues
  def schedulableQueue: ConcurrentLinkedQueue[Schedulable]
  def schedulingMode: SchedulingMode
  def weight: Int
  def minShare: Int
  def runningTasks: Int
  def priority: Int
  def stageId: Int
  def name: String

    //< 省略若干代码
}

private[spark] class FairSchedulingAlgorithm extends SchedulingAlgorithm {
  override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
    val minShare1 = s1.minShare
    val minShare2 = s2.minShare
    val runningTasks1 = s1.runningTasks
    val runningTasks2 = s2.runningTasks
    val s1Needy = runningTasks1 < minShare1
    val s2Needy = runningTasks2 < minShare2
    val minShareRatio1 = runningTasks1.toDouble / math.max(minShare1, 1.0).toDouble
    val minShareRatio2 = runningTasks2.toDouble / math.max(minShare2, 1.0).toDouble
    val taskToWeightRatio1 = runningTasks1.toDouble / s1.weight.toDouble
    val taskToWeightRatio2 = runningTasks2.toDouble / s2.weight.toDouble
    var compare:Int = 0

    if (s1Needy && !s2Needy) {
      return true
    } else if (!s1Needy && s2Needy) {
      return false
    } else if (s1Needy && s2Needy) {
      compare = minShareRatio1.compareTo(minShareRatio2)
    } else {
      compare = taskToWeightRatio1.compareTo(taskToWeightRatio2)
    }

    if (compare < 0) {
      true
    } else if (compare > 0) {
      false
    } else {
      s1.name < s2.name
    }
  }
}

//< 事先创建好用于存放要在各个worker上launch的 List[workerId, ArrayBuffer[TaskDescription]]。
//< 由于task要使用的cores并不一定为1，所以每个worker上要launch得task并不一定等于可用的cores数
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))

var launchedTask = false
for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
  do {
    //< 获取tasks，tasks代表要在哪些worker上启动哪些tasks
    launchedTask = resourceOfferSingleTaskSet(
        taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
  } while (launchedTask)
}

private def resourceOfferSingleTaskSet(
    taskSet: TaskSetManager,
    maxLocality: TaskLocality,
    shuffledOffers: Seq[WorkerOffer],
    availableCpus: Array[Int],
    tasks: Seq[ArrayBuffer[TaskDescription]]) : Boolean = {
  var launchedTask = false

  //< 获取每个worker上要执行的tasks序列
  for (i <- 0 until shuffledOffers.size) {
    val execId = shuffledOffers(i).executorId
    val host = shuffledOffers(i).host
    if (availableCpus(i) >= CPUS_PER_TASK) {
      try {
        for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
          //< 将获得要在index为i的worker上执行的task，添加到tasks(i)中；这样就知道了要在哪个worker上执行哪些tasks了
          tasks(i) += task

          availableCpus(i) -= CPUS_PER_TASK
          assert(availableCpus(i) >= 0)
          launchedTask = true
        }
      } catch {
        case e: TaskNotSerializableException =>
          return launchedTask
      }
    }
  }
  return launchedTask
}

//< 如果资源是有locality特征的
if (maxLocality != TaskLocality.NO_PREF) {
  //< 获取当前taskSet允许执行的locality。getAllowedLocalityLevel随时间变化而变化
  allowedLocality = getAllowedLocalityLevel(curTime)
  //< 如果允许的locality级别低于maxLocality，则使用maxLocality覆盖允许的locality
  if (allowedLocality > maxLocality) {
    // We're not allowed to search for farther-away tasks
    //< 临时将允许的locality级别降低到资源允许的最高locality级别
    allowedLocality = maxLocality
  }
}

dequeueTask(execId, host, allowedLocality) match {
  case Some((index, taskLocality, speculative)) => {
    //< 进行各种信息更新操作
    
    addRunningTask(taskId)

    // We used to log the time it takes to serialize the task, but task size is already
    // a good proxy to task serialization time.
    // val timeTaken = clock.getTime() - startTime
    val taskName = s"task ${info.id} in stage ${taskSet.id}"
    sched.dagScheduler.taskStarted(task, info)
    return Some(new TaskDescription(taskId = taskId, attemptNumber = attemptNum, execId,
      taskName, index, serializedTask))
  }
  case _ =>
}

private def dequeueTask(execId: String, host: String, maxLocality: TaskLocality.Value)
    : Option[(Int, TaskLocality.Value, Boolean)] =
  {
    //< dequeueTaskFromList: 该方法获取list中一个可以launch的task，同时清除扫描过的已经执行的task。其实它从第二次开始首先扫描的一定是已经运行完成的task，因此是延迟清除
    // 同一个Executor，通过execId来查找相应的等待的task
    for (index <- dequeueTaskFromList(execId, getPendingTasksForExecutor(execId))) {
      return Some((index, TaskLocality.PROCESS_LOCAL, false))
    }

    // 通过主机名找到相应的Task,不过比之前的多了一步判断
    if (TaskLocality.isAllowed(maxLocality, TaskLocality.NODE_LOCAL)) {
      for (index <- dequeueTaskFromList(execId, getPendingTasksForHost(host))) {
        return Some((index, TaskLocality.NODE_LOCAL, false))
      }
    }


    if (TaskLocality.isAllowed(maxLocality, TaskLocality.NO_PREF)) {
      // Look for noPref tasks after NODE_LOCAL for minimize cross-rack traffic
      for (index <- dequeueTaskFromList(execId, pendingTasksWithNoPrefs)) {
        return Some((index, TaskLocality.PROCESS_LOCAL, false))
      }
    }

    // 通过Rack的名称查找Task
    if (TaskLocality.isAllowed(maxLocality, TaskLocality.RACK_LOCAL)) {
      for {
        rack <- sched.getRackForHost(host)
        index <- dequeueTaskFromList(execId, getPendingTasksForRack(rack))
      } {
        return Some((index, TaskLocality.RACK_LOCAL, false))
      }
    }

    // 查找那些preferredLocations为空的，不指定在哪里执行的Task来执行
    if (TaskLocality.isAllowed(maxLocality, TaskLocality.ANY)) {
      for (index <- dequeueTaskFromList(execId, allPendingTasks)) {
        return Some((index, TaskLocality.ANY, false))
      }
    }

    // find a speculative task if all others tasks have been scheduled
    // 最后没办法了，拖的时间太长了，只能启动推测执行了
    dequeueSpeculativeTask(execId, host, maxLocality).map {
      case (taskIndex, allowedLocality) => (taskIndex, allowedLocality, true)}
  }

def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
  for (task <- tasks.flatten) {
    val ser = SparkEnv.get.closureSerializer.newInstance()
    //< 序列化task
    val serializedTask = ser.serialize(task)
    //< 若序列化后的task的size大于等于Akka可用空间
    if (serializedTask.limit >= akkaFrameSize - AkkaUtils.reservedSizeBytes) {
      val taskSetId = scheduler.taskIdToTaskSetId(task.taskId)
      scheduler.activeTaskSets.get(taskSetId).foreach { taskSet =>
        try {
          var msg = "Serialized task %s:%d was %d bytes, which exceeds max allowed: " +
            "spark.akka.frameSize (%d bytes) - reserved (%d bytes). Consider increasing " +
            "spark.akka.frameSize or using broadcast variables for large values."
          msg = msg.format(task.taskId, task.index, serializedTask.limit, akkaFrameSize, AkkaUtils.reservedSizeBytes)
          //< 中止taskSet，标记为已完成；同时将该taskSet的状态置为isZombie（Zombie：僵尸）
          taskSet.abort(msg)
        } catch {
          case e: Exception => logError("Exception in error callback", e)
        }
      }
    }
    else {
      //< 若序列化后的task的size小于Akka可用空间，减去对应executor上的可用cores数并向对应的executor发送启动task消息
      val executorData = executorDataMap(task.executorId)
      executorData.freeCores -= scheduler.CPUS_PER_TASK
      executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
    }
  }
}