Gang 调度调试指南

// pkg/scheduler/plugins/gang/gang.go 第 57 行
func (gp *gangPlugin) OnSessionOpen(ssn *framework.Session) {
    ssn.AddJobValidFn(gp.Name(), validJobFn)           // 第 95 行
    ssn.AddReclaimableFn(gp.Name(), preemptableFn)      // 第 122 行
    ssn.AddPreemptableFn(gp.Name(), preemptableFn)      // 第 123 行
    ssn.AddJobOrderFn(gp.Name(), jobOrderFn)             // 第 149 行
    ssn.AddSubJobOrderFn(gp.Name(), subJobOrderFn)       // 第 175 行
    ssn.AddJobReadyFn(gp.Name(), jobReadyFn)             // 第 177 行
    ssn.AddSubJobReadyFn(gp.Name(), subJobReadyFn)       // 第 185 行
    ssn.AddJobPipelinedFn(gp.Name(), pipelinedFn)        // 第 197 行
    ssn.AddSubJobPipelinedFn(gp.Name(), subJobPipelinedFn)  // 第 199 行
    ssn.AddJobStarvingFns(gp.Name(), jobStarvingFn)      // 第 212 行
}

// gang.go 第 58-93 行
validJobFn := func(obj interface{}) *api.ValidateResult {
    job := obj.(*api.JobInfo)

    // 检查 1: 每个 Task 规格的有效 Pod 数量是否满足 Task 级别的 minAvailable
    if valid := job.CheckTaskValid(); !valid {
        return &api.ValidateResult{
            Pass:    false,
            Reason:  v1beta1.NotEnoughPodsOfTaskReason,
            Message: "Not enough valid pods of each task for gang-scheduling",
        }
    }

    // 检查 2: SubGroup 策略下，每个 SubGroup 的有效 Pod 数量
    if valid := job.CheckSubJobValid(); !valid {
        return &api.ValidateResult{
            Pass:    false,
            Reason:  v1beta1.NotEnoughPodsOfTaskReason,
            Message: "Not enough valid subGroups of each task for gang-scheduling",
        }
    }

    // 检查 3: 整体有效 Task 数量是否满足 Job 级别的 MinAvailable
    vtn := job.ValidTaskNum()
    if vtn < job.MinAvailable {
        return &api.ValidateResult{
            Pass:   false,
            Reason: v1beta1.NotEnoughPodsReason,
            Message: fmt.Sprintf("Not enough valid tasks for gang-scheduling, valid: %d, min: %d",
                vtn, job.MinAvailable),
        }
    }
    return nil
}

func (ji *JobInfo) CheckTaskValid() bool {
    // 如果 MinAvailable < TaskMinAvailableTotal，跳过此检查
    if ji.MinAvailable < ji.TaskMinAvailableTotal {
        return true
    }
    // 检查每个 Task 规格
    for taskSpec, minNum := range ji.TaskMinAvailable {
        validNum := int32(0)
        for _, task := range ji.Tasks {
            if task.Role == taskSpec {
                // 有效状态: AllocatedStatus || Succeeded || Pipelined || Pending || Waiting
                if AllocatedStatus(task.Status) || task.Status == Succeeded ||
                    task.Status == Pipelined || task.Status == Pending || task.Status == Waiting {
                    validNum++
                }
            }
        }
        if validNum < minNum {
            return false   // 该 Task 规格的有效 Pod 不足
        }
    }
    return true
}

func (ji *JobInfo) ValidTaskNum() int32 {
    occupied := 0
    for status, tasks := range ji.TaskStatusIndex {
        if AllocatedStatus(status) ||
            status == Succeeded ||
            status == Pipelined ||
            status == Pending ||
            status == Waiting {
            occupied += len(tasks)
        }
    }
    return int32(occupied)
}

ssn.AddJobReadyFn(gp.Name(), func(obj interface{}) bool {
    ji := obj.(*api.JobInfo)
    if ji.CheckTaskReady() && ji.CheckSubJobReady() && ji.IsReady() {
        return true
    }
    return false
})

func (ji *JobInfo) CheckTaskReady() bool {
    if ji.MinAvailable < ji.TaskMinAvailableTotal {
        return true  // 当 Job MinAvailable 小于所有 Task MinAvailable 之和时跳过
    }
    occupiedMap := ji.getJobAllocatedRoles()  // 获取每个 Role 的已分配数量
    for taskSpec, minNum := range ji.TaskMinAvailable {
        if occupiedMap[taskSpec] < minNum {
            return false  // 该 Role 的已分配数量不足
        }
    }
    return true
}

func (ji *JobInfo) IsReady() bool {
    return ji.ReadyTaskNum()+ji.PendingBestEffortTaskNum() >= ji.MinAvailable
}

func (ssn *Session) JobReady(obj interface{}) bool {
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            if !isEnabled(plugin.EnabledJobReady) { continue }
            jrf, found := ssn.jobReadyFns[plugin.Name]
            if !found { continue }
            if !jrf(obj) {
                return false  // 任何一个 plugin 说不就绪，就不就绪
            }
        }
    }
    return true  // 所有注册的 plugin 都说就绪（或没有 plugin 注册）
}

pipelinedFn := func(obj interface{}) int {
    ji := obj.(*api.JobInfo)
    if ji.CheckTaskPipelined() && ji.CheckSubJobPipelined() && ji.IsPipelined() {
        return util.Permit   // 返回正值表示 Pipelined
    }
    return util.Reject       // 返回负值表示未 Pipelined
}

func (ji *JobInfo) CheckTaskPipelined() bool {
    if ji.MinAvailable < ji.TaskMinAvailableTotal {
        return true
    }
    occupiedMap := map[string]int32{}
    for status, tasks := range ji.TaskStatusIndex {
        if AllocatedStatus(status) || status == Pipelined {  // 关键：包含 Pipelined
            for _, task := range tasks {
                occupiedMap[task.Role]++
            }
        }
    }
    for taskSpec, minNum := range ji.TaskMinAvailable {
        if occupiedMap[taskSpec] < minNum {
            return false
        }
    }
    return true
}

func (s *Statement) Commit() {
    klog.V(3).Info("Committing operations ...")
    for _, op := range s.operations {
        op.task.ClearLastTxContext()
        switch op.name {
        case Evict:
            err := s.evict(op.task, op.reason)    // 调用 cache.Evict -> 删除 Pod
        case Pipeline:
            s.pipeline(op.task)                    // 空操作 - Pipeline 状态已在记录时更新
        case Allocate:
            err := s.allocate(op.task)             // 调用 cache.AddBindTask -> 绑定 Pod
            if err != nil {
                s.unallocate(op.task)              // 失败时回滚
            }
        }
    }
}

func (s *Statement) Discard() {
    klog.V(3).Info("Discarding operations ...")
    for i := len(s.operations) - 1; i >= 0; i-- {   // 关键：逆序回滚
        op := s.operations[i]
        op.task.GenerateLastTxContext()
        switch op.name {
        case Evict:
            s.unevict(op.task)       // 恢复 Task 状态为 Running
        case Pipeline:
            s.UnPipeline(op.task)    // 恢复 Task 状态为 Pending，从节点移除
        case Allocate:
            s.unallocate(op.task)    // 恢复 Task 状态为 Pending，从节点移除
        }
    }
}

jobOrderFn := func(l, r interface{}) int {
    lv := l.(*api.JobInfo)
    rv := r.(*api.JobInfo)

    lReady := lv.IsReady()
    rReady := rv.IsReady()

    if lReady && rReady { return 0 }  // 两个都就绪，不区分
    if lReady { return 1 }             // l 就绪，r 未就绪 -> r 优先（返回正值 = l 排后）
    if rReady { return -1 }            // r 就绪，l 未就绪 -> l 优先（返回负值 = l 排前）
    return 0                           // 两个都未就绪，不区分
}

subJobOrderFn := func(l, r interface{}) int {
    lv := l.(*api.SubJobInfo)
    rv := r.(*api.SubJobInfo)
    lReady := lv.IsReady()
    rReady := rv.IsReady()
    // 同样的逻辑：未就绪的优先
    if lReady && rReady { return 0 }
    if lReady { return 1 }
    if rReady { return -1 }
    return 0
}

func (ssn *Session) JobOrderFn(l, r interface{}) bool {
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            if !isEnabled(plugin.EnabledJobOrder) { continue }
            jof, found := ssn.jobOrderFns[plugin.Name]
            if !found { continue }
            if j := jof(l, r); j != 0 {
                return j < 0  // 负值 = l 排前 = 优先级更高
            }
        }
    }
    // 没有 plugin 决定的情况下，按创建时间排序
    lv := l.(*api.JobInfo)
    rv := r.(*api.JobInfo)
    if lv.CreationTimestamp.Equal(&rv.CreationTimestamp) {
        return lv.UID < rv.UID
    }
    return lv.CreationTimestamp.Before(&rv.CreationTimestamp)
}

func (sji *SubJobInfo) IsReady() bool {
    return sji.ReadyTaskNum()+sji.PendingBestEffortTaskNum() >= sji.MinAvailable
}

func (sji *SubJobInfo) IsPipelined() bool {
    return sji.WaitingTaskNum()+sji.ReadyTaskNum()+sji.PendingBestEffortTaskNum() >= sji.MinAvailable
}

func (ji *JobInfo) CheckSubJobValid() bool {
    subJobs := map[SubJobGID]int32{}
    for _, subJob := range ji.SubJobs {
        if _, ok := subJobs[subJob.GID]; !ok {
            subJobs[subJob.GID] = 0
        }
        // 统计每个 SubGroup 中有效的 SubJob 数量
        if subJob.ValidTaskNum() >= subJob.MinAvailable {
            subJobs[subJob.GID]++
        }
    }
    // 检查每个 SubGroup 是否满足 MinSubJobs
    // ...
}

func (ji *JobInfo) CheckSubJobReady() bool {
    if err := ji.checkSubJobCondition(func(subJob *SubJobInfo) bool {
        return subJob.IsReady()
    }); err != nil {
        return false
    }
    return true
}

func (ssn *Session) SubJobReady(job *api.JobInfo, subJob *api.SubJobInfo) bool {
    if !job.ContainsSubJobPolicy() {
        return ssn.JobReady(job)  // 无 SubJob 策略时退化为 JobReady
    }
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            if !isEnabled(plugin.EnabledSubJobReady) { continue }
            fn, found := ssn.subJobReadyFns[plugin.Name]
            if !found { continue }
            if !fn(subJob) { return false }
        }
    }
    return true
}

func (gp *gangPlugin) OnSessionClose(ssn *framework.Session) {
    var unreadyTaskCount int32
    var unScheduleJobCount int

    for _, job := range ssn.Jobs {
        if len(job.Tasks) == 0 { continue }

        if !job.IsReady() {
            // 计算不可调度的 Task 数量
            schedulableTaskNum := func() (num int32) {
                for _, task := range job.TaskStatusIndex[api.Pending] {
                    ctx := task.GetTransactionContext()
                    if task.LastTransaction != nil {
                        ctx = *task.LastTransaction
                    }
                    if api.AllocatedStatus(ctx.Status) {
                        num++
                    }
                }
                return num + job.ReadyTaskNum()
            }
            unreadyTaskCount = job.MinAvailable - schedulableTaskNum()

            msg := fmt.Sprintf("%v/%v tasks in gang unschedulable: %v",
                unreadyTaskCount, len(job.Tasks), job.FitError())

            // 更新 PodGroupCondition 为 Unschedulable
            jc := &scheduling.PodGroupCondition{
                Type:               scheduling.PodGroupUnschedulableType,
                Status:             v1.ConditionTrue,
                LastTransitionTime: metav1.Now(),
                TransitionID:       string(ssn.UID),
                Reason:             v1beta1.NotEnoughResourcesReason,
                Message:            msg,
            }
            ssn.UpdatePodGroupCondition(job, jc)
        } else {
            // Job 就绪 - 更新为 Scheduled
            jc := &scheduling.PodGroupCondition{
                Type:               scheduling.PodGroupScheduled,
                Status:             v1.ConditionTrue,
                Reason:             "tasks in gang are ready to be scheduled",
            }
            ssn.UpdatePodGroupCondition(job, jc)
        }
    }
}

schedulableTaskNum := func() (num int32) {
    for _, task := range job.TaskStatusIndex[api.Pending] {
        ctx := task.GetTransactionContext()
        if task.LastTransaction != nil {
            ctx = *task.LastTransaction      // 使用上一次事务的状态
        }
        if api.AllocatedStatus(ctx.Status) {
            num++                            // Pending 但上次事务中被 Allocated 的 Task
        }
    }
    return num + job.ReadyTaskNum()
}