抢占流程调试指南

// pkg/scheduler/actions/preempt/preempt.go 第 161-227 行
for _, queue := range queues {
    for {
        preemptors := preemptorsMap[queue.UID]
        if preemptors == nil || preemptors.Empty() {
            break
        }
        preemptorJob := preemptors.Pop().(*api.JobInfo)
        stmt := framework.NewStatement(ssn)

        for {
            if !ssn.JobStarving(preemptorJob) {
                break  // Job 不再饥饿，停止抢占
            }
            preemptor := preemptorTasks[preemptorJob.UID].Pop().(*api.TaskInfo)
            assigned, err = pmpt.preempt(ssn, stmt, preemptor, filter, ph)
        }

        // 关键决策点：Commit 还是 Discard
        if ssn.JobPipelined(preemptorJob) {
            stmt.Commit()    // Job 已获得足够资源，提交操作
        } else {
            stmt.Discard()   // Job 未获得足够资源，回滚所有操作
            continue
        }
    }
}

// pkg/scheduler/actions/preempt/preempt.go 第 191-209 行
filter := func(task *api.TaskInfo) bool {
    if !api.PreemptableStatus(task.Status) { return false }      // 只抢占运行中的 Task
    if preemptor.BestEffort && !task.BestEffort { return false }  // BestEffort 不能抢占非 BestEffort
    if !task.Preemptable { return false }                         // 跳过不可抢占的 Pod
    job, found := ssn.Jobs[task.Job]
    if !found { return false }
    return job.Queue == preemptorJob.Queue && preemptor.Job != task.Job  // 同队列、不同 Job
}

// pkg/scheduler/actions/preempt/preempt.go 第 251-267 行
filter := func(task *api.TaskInfo) bool {
    if !api.PreemptableStatus(task.Status) { return false }
    if preemptor.BestEffort && !task.BestEffort { return false }
    if !task.Preemptable { return false }
    return preemptor.Job == task.Job  // 同 Job
}

// pkg/scheduler/plugins/gang/gang.go 第 207-212 行
jobStarvingFn := func(obj interface{}) bool {
    ji := obj.(*api.JobInfo)
    return ji.IsStarving()
}

// pkg/scheduler/api/job_info.go 第 1180-1182 行
func (ji *JobInfo) IsStarving() bool {
    return ji.WaitingTaskNum()+ji.ReadyTaskNum() < ji.MinAvailable
}

func (ssn *Session) JobStarving(obj interface{}) bool {
    var hasFound bool
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            if !isEnabled(plugin.EnabledJobStarving) { continue }
            jrf, found := ssn.jobStarvingFns[plugin.Name]
            if !found { continue }
            hasFound = true
            if !jrf(obj) {
                return false  // 任何一个 plugin 说不饥饿，就不饥饿
            }
        }
        if hasFound {
            return true  // 当前 tier 有 plugin 且全部说饥饿，返回 true
        }
    }
    return false  // 没有任何 plugin 注册，默认不饥饿
}

// proportion.go 第 357-361 行
ssn.AddPreemptiveFn(pp.Name(), func(obj interface{}, candidate interface{}) bool {
    queue := obj.(*api.QueueInfo)
    task := candidate.(*api.TaskInfo)
    return queueAllocatable(queue, task)  // 检查 allocated + task.Resreq <= deserved
})

func (ssn *Session) Preemptable(preemptor *api.TaskInfo, preemptees []*api.TaskInfo) []*api.TaskInfo {
    var victims []*api.TaskInfo
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            candidates, abstain := pf(preemptor, preemptees)
            if abstain == 0 { continue }          // Plugin 弃权
            if len(candidates) == 0 {
                victims = nil; break              // Plugin 明确拒绝所有候选
            }
            if victims == nil {
                victims = candidates              // 第一个投票的 plugin，初始化 victims
            } else {
                // 取交集
                intersection := intersect(victims, candidates)
                victims = intersection
            }
        }
        if victims != nil { return victims }      // 当前 tier 有结果，立即返回
    }
    return victims
}

preemptableFn := func(preemptor *api.TaskInfo, preemptees []*api.TaskInfo) ([]*api.TaskInfo, int) {
    var victims []*api.TaskInfo
    jobOccupiedMap := map[api.JobID]int32{}

    for _, preemptee := range preemptees {
        job := ssn.Jobs[preemptee.Job]
        if _, found := jobOccupiedMap[job.UID]; !found {
            jobOccupiedMap[job.UID] = job.ReadyTaskNum()
        }
        if jobOccupiedMap[job.UID] > job.MinAvailable {
            jobOccupiedMap[job.UID]--
            victims = append(victims, preemptee)    // 可以被抢占
        }
        // 否则跳过 - 不能破坏 Gang 约束
    }
    return victims, util.Permit
}

preemptableFn := func(preemptor *api.TaskInfo, preemptees []*api.TaskInfo) ([]*api.TaskInfo, int) {
    latt := drf.jobAttrs[preemptor.Job]
    lalloc := latt.allocated.Clone().Add(preemptor.Resreq)
    _, ls := drf.calculateShare(lalloc, drf.totalResource)  // preemptor 的 DRF 份额

    for _, preemptee := range preemptees {
        ralloc := allocations[preemptee.Job].Sub(preemptee.Resreq)
        _, rs := drf.calculateShare(ralloc, drf.totalResource)  // preemptee 移除后的 DRF 份额

        if ls < rs || math.Abs(ls-rs) <= shareDelta {
            victims = append(victims, preemptee)  // preemptor 份额 <= preemptee，允许抢占
        }
    }
    return victims, util.Permit
}

victimsQueue := util.NewPriorityQueue(func(l, r interface{}) bool {
    lv := l.(*api.TaskInfo)
    rv := r.(*api.TaskInfo)
    if lv.Job == rv.Job {
        return !ssn.TaskOrderFn(l, r)        // 同 Job：反向 TaskOrder（低优先级在前）
    }
    // 不同 Job：反向 JobOrder
    if lvJob.Queue != rvJob.Queue {
        return ssn.VictimQueueOrderFn(...)    // 不同队列：使用 VictimQueueOrder
    }
    return !ssn.JobOrderFn(lvJob, rvJob)      // 同队列：反向 JobOrder
})

// 模拟移除 - proportion plugin 的实现
// pkg/scheduler/plugins/proportion/proportion.go 第 433-447 行
ssn.AddSimulateRemoveTaskFn(pp.Name(), func(ctx context.Context, cycleState fwk.CycleState,
    taskToSchedule *api.TaskInfo, taskToRemove *api.TaskInfo, nodeInfo *api.NodeInfo) error {
    state, _ := getProportionState(cycleState)
    job := ssn.Jobs[taskToRemove.Job]
    attr := state.queueAttrs[job.Queue]
    attr.allocated.Sub(taskToRemove.Resreq)   // 模拟减少队列已分配资源
    updateQueueAttrShare(attr)
    return nil
})

func (ra *Action) reclaimForTask(ssn *framework.Session, stmt *framework.Statement,
    task *api.TaskInfo, job *api.JobInfo) {
    // 1. 筛选候选节点
    totalNodes := ssn.FilterOutUnschedulableAndUnresolvableNodesForTask(task)
    predicateNodes, _ := predicateHelper.PredicateNodes(task, totalNodes,
        ssn.PredicateForPreemptAction, ...)   // 注意：使用 PreemptAction 的 Predicate（更宽松）

    for _, n := range predicateNodes {
        // 2. 收集该节点上其他队列的 Running Task 作为候选 reclaimee
        var reclaimees []*api.TaskInfo
        for _, taskOnNode := range n.Tasks {
            if taskOnNode.Status != api.Running { continue }
            j := ssn.Jobs[taskOnNode.Job]
            if j.Queue != job.Queue {              // 必须是不同队列
                q := ssn.Queues[j.Queue]
                if q.Reclaimable() {               // 队列允许被回收
                    reclaimees = append(reclaimees, taskOnNode.Clone())
                }
            }
        }

        // 3. 调用 Plugin 筛选 Victims
        victims := ssn.Reclaimable(task, reclaimees)

        // 4. 逐个驱逐 Victim 直到资源满足
        for !victimsQueue.Empty() {
            reclaimee := victimsQueue.Pop().(*api.TaskInfo)
            stmt.Evict(reclaimee, "reclaim")
            availableResources.Add(reclaimee.Resreq)
            if resreq.LessEqual(availableResources, api.Zero) {
                break   // 资源已足够
            }
        }

        // 5. Pipeline Task 到节点
        if task.InitResreq.LessEqual(availableResources, api.Zero) {
            stmt.Pipeline(task, n.Name, evictionOccurred)
            break
        }
    }
}

ssn.AddOverusedFn(pp.Name(), func(obj interface{}) bool {
    queue := obj.(*api.QueueInfo)
    attr := pp.queueOpts[queue.UID]
    overused := attr.deserved.LessEqual(attr.allocated, api.Zero)
    // overused = true 当 deserved <= allocated（队列已超用）
    return overused
})

ssn.AddReclaimableFn(pp.Name(), func(reclaimer *api.TaskInfo, reclaimees []*api.TaskInfo) ([]*api.TaskInfo, int) {
    var victims []*api.TaskInfo
    allocations := map[api.QueueID]*api.Resource{}

    for _, reclaimee := range reclaimees {
        job := ssn.Jobs[reclaimee.Job]
        attr := pp.queueOpts[job.Queue]
        allocated := allocations[job.Queue]

        // 只有当队列 allocated > deserved 时才回收
        if !allocated.LessEqual(attr.deserved, api.Zero) {
            allocated.Sub(reclaimee.Resreq)
            victims = append(victims, reclaimee)
        }
    }
    return victims, util.Permit
})

func (pmpt *Action) calculateNumCandidates(numNodes int) int {
    n := (numNodes * pmpt.minCandidateNodesPercentage) / 100  // 默认 10%
    if n < pmpt.minCandidateNodesAbsolute { n = pmpt.minCandidateNodesAbsolute }  // 最小 1
    if n > pmpt.maxCandidateNodesAbsolute { n = pmpt.maxCandidateNodesAbsolute }  // 最大 100
    if n > numNodes { n = numNodes }
    return n
}

func (pmpt *Action) DryRunPreemption(...) ([]*candidate, map[string]api.Status, error) {
    candidates := newCandidateList(numCandidates)
    ctx, cancel := context.WithCancel(context.Background())

    checkNode := func(i int) {
        nodeInfoCopy := potentialNodes[(offset+i)%len(potentialNodes)].Clone()
        stateCopy := state.Clone()

        victims, status := SelectVictimsOnNode(ctx, stateCopy, preemptor, ...)
        if status.IsSuccess() && len(victims) != 0 {
            candidates.add(&candidate{victims: victims, name: nodeInfoCopy.Name})
            if candidates.size() >= numCandidates {
                cancel()  // 找到足够候选，取消剩余检查
            }
        }
    }

    // 并行检查所有潜在节点
    workqueue.ParallelizeUntil(ctx, pmpt.topologyAwarePreemptWorkerNum,
        len(potentialNodes), checkNode)
    return candidates.get(), nodeStatuses, errs
}

// preempt.go 第 302 行
predicateNodes, _ := predicateHelper.PredicateNodes(
    preemptor, allNodes,
    ssn.PredicateForPreemptAction,  // 更宽松：允许 Unschedulable 节点
    pmpt.enablePredicateErrorCache, ssn.NodesInShard)

// Commit 时执行的操作（statement.go 第 418-440 行）
func (s *Statement) Commit() {
    for _, op := range s.operations {
        switch op.name {
        case Evict:    s.evict(op.task, op.reason)   // 真正执行 Pod 驱逐
        case Pipeline: s.pipeline(op.task)            // 标记 Task 为 Pipelined
        case Allocate: s.allocate(op.task)            // 绑定 Task 到节点
        }
    }
}

// Discard 时执行的回滚操作（statement.go 第 392-415 行）
func (s *Statement) Discard() {
    for i := len(s.operations) - 1; i >= 0; i-- {   // 逆序回滚
        switch op.name {
        case Evict:    s.unevict(op.task)             // 恢复被驱逐的 Task
        case Pipeline: s.UnPipeline(op.task)           // 取消 Pipeline
        case Allocate: s.unallocate(op.task)           // 取消分配
        }
    }
}