1. DAG 调度基础：定义、问题背景与核心价值

1.1 有向无环图（DAG）的数学定义

有向无环图（Directed Acyclic Graph, DAG）是图论中一种特殊的有向图，它由一组顶点（Vertex）和一组有向边（Directed Edge）组成，且图中不存在任何有向环路。在任务编排场景中，每个顶点代表一个计算任务或执行单元，每条有向边表示任务之间的依赖关系——即前驱任务完成后才能启动后继任务。

1.2 调度问题的背景与挑战

现代分布式系统面临的任务编排场景日益复杂：一个业务流水线可能包含数十个异构任务，涉及数据预处理、模型推理、结果聚合、异常处理等多个阶段。这些任务之间存在显式的依赖关系，且执行时间、资源消耗、失败重试策略各不相同。调度问题的核心目标是在满足依赖约束的前提下，最小化整体完成时间（Makespan），同时最大化资源利用率。

调度问题的主要挑战包括：

• 依赖复杂性：任务间存在多级扇入/扇出关系，单一任务可能等待多个前驱完成。
• 异构性：任务的计算类型（CPU 密集型、I/O 密集型、GPU 加速）和资源需求差异巨大。
• 动态性：任务执行时间受数据量、系统负载影响，难以精确预估。
• 容错性：部分任务失败后需要局部重试或重新调度，不能影响已完成的依赖。
• 规模扩展：当 DAG 顶点数达到数千甚至数万时，调度器的决策延迟必须保持在毫秒级。

1.3 DAG 在任务编排中的核心价值

有向无环图之所以成为分布式任务编排的事实标准，源于以下四个核心价值：

1. 依赖显式化：DAG 将任务间的数据流和控制流以图结构明确表达，消除了隐式依赖带来的竞态条件和死锁风险。
2. 并行度最大化：通过分析 DAG 的拓扑结构，调度器可以识别出所有无依赖关系的任务，将它们并发执行，从而充分利用集群资源。
3. 可终止性保证：无环性确保调度过程必然在有限步内结束，不存在循环等待或无限递归。
4. 增量可观测：每个任务的状态（等待中、运行中、已完成、失败）可以独立追踪，便于构建监控、日志和告警系统。

1.4 DAG 调度与普通任务队列的对比

1.5 关键路径与调度边界

在 DAG 调度中，关键路径（Critical Path） 是从起始节点到终止节点的最长路径，其长度决定了整个 DAG 的理论最短完成时间（Makespan 下界）。调度器的核心任务之一就是识别关键路径，并优先保障关键路径上的任务获得资源，避免因非关键任务的资源竞争导致整体延迟。

1.6 Go 语言中的 DAG 表示示例

在 Go 中，DAG 通常使用邻接表或邻接矩阵表示。以下是一个基于邻接表的 DAG 定义与拓扑排序实现：

// DAGNode 表示 DAG 中的一个任务节点
type DAGNode struct {
    ID       string
    DependOn []string // 依赖的前驱节点 ID 列表
    Run      func() error
}

// DAG 结构体
type DAG struct {
    nodes map[string]*DAGNode
    edges map[string][]string // 邻接表：前驱 → 后继
}

// TopologicalSort 返回拓扑排序后的节点 ID 列表
func (d *DAG) TopologicalSort() ([]string, error) {
    inDegree := make(map[string]int)
    for id := range d.nodes {
        inDegree[id] = 0
    }
    for _, successors := range d.edges {
        for _, succ := range successors {
            inDegree[succ]++
        }
    }

    queue := []string{}
    for id, deg := range inDegree {
        if deg == 0 {
            queue = append(queue, id)
        }
    }

    result := []string{}
    for len(queue) > 0 {
        u := queue[0]
        queue = queue[1:]
        result = append(result, u)
        for _, v := range d.edges[u] {
            inDegree[v]--
            if inDegree[v] == 0 {
                queue = append(queue, v)
            }
        }
    }

    if len(result) != len(d.nodes) {
        return nil, fmt.Errorf("DAG contains cycle")
    }
    return result, nil
}