GoDAG调度边界领域建模与架构设计 - 第1节

1.1 DAG 调度概念

有向无环图（Directed Acyclic Graph, DAG）是描述任务依赖关系的核心数学模型。在分布式调度领域，DAG 将计算任务建模为节点，将任务间的数据或控制依赖建模为有向边，且图中不存在任何环路。

1.1.1 调度问题的形式化定义

一个 DAG 调度实例可表示为四元组 ⟨G, P, C, O⟩：

1.1.2 关键调度指标

• makespan：从第一个任务开始到最后一个任务完成的总时间跨度
• 关键路径长度：图中最长依赖路径的累计计算时间，决定调度下限
• 调度长度比：实际 makespan 与关键路径长度的比值，衡量调度质量
• 资源利用率：各处理器繁忙时间与总时间的比值

1.2 GoDAG 库的起源与设计哲学

1.2.1 起源背景

GoDAG 诞生于 2020 年初，由分布式计算团队在构建大规模 ETL 管道时提炼而成。当时主流 DAG 框架（如 Apache Airflow、Argo Workflows）存在以下痛点：

• 调度粒度粗：以容器/进程为最小单元，无法表达函数级依赖
• 状态管理重：依赖外部数据库维护 DAG 状态，增加运维复杂度
• Go 生态缺失：Go 语言在云原生领域占据主导，但缺少原生 DAG 调度库

1.2.2 设计哲学

1.3 核心数据结构：Node / Edge / Graph

1.3.1 Node（节点）

Node 是 DAG 中的最小执行单元，封装了任务标识、计算逻辑和元数据。GoDAG 中的 Node 定义如下：

// Node 定义 DAG 中的一个任务节点
type Node struct {
    ID       string                 // 全局唯一标识
    Name     string                 // 可读名称
    Payload  NodePayload           // 执行负载（函数/命令/子图）
    Metadata map[string]interface{} // 扩展元数据
    weight   int64                 // 计算权重（用于调度决策）
}

// NodePayload 是节点执行逻辑的抽象接口
type NodePayload interface {
    Execute(ctx context.Context, input map[string]interface{}) (map[string]interface{}, error)
}

1.3.2 Edge（边）

Edge 表示节点间的依赖关系，包含源节点、目标节点以及可选的传输数据量：

// Edge 定义节点间的有向依赖
type Edge struct {
    SourceID string // 上游节点 ID
    TargetID string // 下游节点 ID
    DataSize int64  // 传输数据量（字节），用于通信代价计算
    Label    string // 边标签（可选）
}

1.3.3 Graph（图）

Graph 是 DAG 的容器，管理所有节点和边的增删查改，并提供拓扑验证能力：

// Graph 管理 DAG 的拓扑结构
type Graph struct {
    mu     sync.RWMutex
    nodes  map[string]*Node   // 节点集合
    edges  map[string][]*Edge // 邻接表（出边）
    inEdges map[string][]*Edge // 入边索引
}

// 核心方法
func (g *Graph) AddNode(n *Node) error
func (g *Graph) AddEdge(sourceID, targetID string) error
func (g *Graph) TopologicalSort() ([]*Node, error)
func (g *Graph) Validate() error

1.3.4 拓扑验证与环路检测

GoDAG 在 Validate() 方法中执行以下检查：

1. 节点存在性：所有边引用的 SourceID 和 TargetID 必须存在于 nodes 集合中
2. 环路检测：使用 Kahn 算法或 DFS 染色法检测是否存在环
3. 连通性：至少存在一个入口节点（入度为 0）和一个出口节点（出度为 0）
4. 权重合法性：所有节点 weight > 0

// Validate 执行完整的 DAG 合法性校验
func (g *Graph) Validate() error {
    g.mu.RLock()
    defer g.mu.RUnlock()

    // 1. 检查边引用的节点是否存在
    for _, edges := range g.edges {
        for _, e := range edges {
            if _, ok := g.nodes[e.SourceID]; !ok {
                return fmt.Errorf("node %s not found", e.SourceID)
            }
            if _, ok := g.nodes[e.TargetID]; !ok {
                return fmt.Errorf("node %s not found", e.TargetID)
            }
        }
    }

    // 2. 环路检测（Kahn 算法）
    inDegree := make(map[string]int)
    for id := range g.nodes {
        inDegree[id] = 0
    }
    for _, edges := range g.edges {
        for _, e := range edges {
            inDegree[e.TargetID]++
        }
    }

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

    visited := 0
    for len(queue) > 0 {
        id := queue[0]
        queue = queue[1:]
        visited++
        for _, e := range g.edges[id] {
            inDegree[e.TargetID]--
            if inDegree[e.TargetID] == 0 {
                queue = append(queue, e.TargetID)
            }
        }
    }

    if visited != len(g.nodes) {
        return fmt.Errorf("cycle detected: %d/%d nodes reachable", visited, len(g.nodes))
    }
    return nil
}