DAGRL: 面向异构 DAG 调度的统一训练评测框架与 GNN-PPO 算法设计与实现

提交内容：

1. 参赛承诺书：DAGRL/参赛承诺书.pdf
2. 演示视频：通过网盘分享的文件：dagrl演示视频.mp4 链接: https://pan.baidu.com/s/1Qye3VkCK4wukO09WD3vNCw?pwd=r1vx 提取码: r1vx 复制这段内容后打开百度网盘手机App，操作更方便哦
3. 技术文档：DAGRL/技术文档.docx
4. 演示PPT：DAGRL/演示ppt.pptx
5. README.md: DAGRL/README.md

1. 项目介绍

1.1 DAGRL 框架

DAGRL 是一个调度算法评测框架，将场景管理、方法注册、统一评测和报告生成标准化。所有调度方法——启发式（HEFT、CPOP、PEFT）、学习型（GNN-PPO）、随机基线——通过统一的 Method 接口接入，type 字段即可切换。框架不介入方法内部实现。

主要创新	说明
Registry 方法注册	借鉴 MMDetection 设计，@METHOD.register_module() 一行装饰器注册，type 字段实例化。新方法接入无需修改框架代码，评测即插即用
ScheduleEngine 共享内核	集中式调度引擎作为唯一真实来源，所有方法平等引用。提供 EST/EFT 计算、可行性判断、任务放置与调度校验的统一接口，杜绝重复实现与不一致风险
Python Config 多继承	实验配置通过 _base_ 列表声明继承链，支持深度合并与逐层覆写。切换数据集或方法只需修改少量字段
标准化评测管道	参考基准计算 → 多方法批量评测 → 指标聚合 → JSON/CSV/Markdown/图表四格式报告输出，全自动
训练中断恢复	Checkpoint 完整保存权重、优化器动量、LR 调度器状态与训练步数。BC 和 PPO 两个阶段均支持断点精确续跑

1.2 GNN-PPO 算法

GNN-PPO 是框架内置的核心学习型调度算法，通过图注意力编码与强化学习在 HEFT 基础上进一步优化 makespan。策略网络由 GATv2 编码器（提取任务依赖）、ResourceEncoder（编码节点属性）、TaskResourceCrossAttention（任务-资源双向注意力交互）和 PairScorer（评分输出）四模块串联构成。训练采用三阶段递进：HEFT 专家轨迹生成 → Behavior Cloning 预训练 → Curriculum PPO 微调（50→100→300 任务渐进），推理支持 Greedy 与 Beam Search 两种策略。

主要创新	说明
TaskResourceCrossAttention	双向 Scaled Dot-Product Cross-Attention（4 头），使任务与资源在编码阶段即发生直接交互。任务动态关注最适合自身特征的资源节点，资源动态感知待调度任务的计算需求。Pair-Bias 机制将手工启发式特征（EST/EFT/通信时间）以注意力偏置注入交互权重，融合领域知识与可学习模式
Masked PPO	在 Rollout 采样、Update 重计算和 Advantage 归一化三环节统一引入动作掩码，将标准 PPO 适配到每步动态变化的合法动作空间，非法动作概率严格为零
Telescoping 稠密奖励	每步奖励 r_t = (\hat{M}(s_t) - \hat{M}(s_{t+1})) / M_{HEFT}，可正可负。累积满足 \sum r_t = (\hat{M}_0 - M_{RL}) / M_{HEFT}，训练目标与评测指标精确对齐
Shape-Bucketed BC 批处理	按 DAG 尺寸 (T, N) 分桶，同尺寸 transition 共享 GNN 前向，GPU 利用率从 <5% 提升至 ≥80%
Curriculum Learning	三阶段渐进训练（Phase 0: rnc50 → Phase 1: rnc50+100 → Phase 2: 全规模），由简到难稳定收敛

1.3 实验结果

实验采用 GrapheonRL STG-JSON 数据集（CC BY 4.0）。该数据集包含 180 个随机生成的唯一拓扑图，每个拓扑对应两种资源配置——homogeneous（同构，各节点处理速度相同）和 heterogeneous（异构，各节点速度差异显著）——以及三种 DAG 规模——rnc50（50 任务 / 3 节点）、rnc100（100 任务 / 5 节点）、rnc300（300 任务 / 8 节点），共 180 \times 2 \times 3 = 1080 个调度实例。我们将实例按基础拓扑图 ID 进行分层随机划分，70% 用于训练（756 个）、15% 用于验证（162 个）、15% 用于测试（162 个），确保同一拓扑的同构与异构变体始终位于同一子集，避免拓扑泄漏。

评测时，所有方法在测试集的全部 162 个场景上运行，覆盖了三种难度等级和两种资源模式的所有组合。以 HEFT makespan 为归一化分母，核心指标为 mean_ratio = M_method / M_HEFT（< 1 表示优于 HEFT）。

下表中第 2–5 列为同构资源环境下的结果，第 6–9 列为异构资源环境下的结果，第 10 列为全部 162 个测试场景的汇总。Ours 为 GNN-PPO（Greedy 推理）。

Method	同构 rnc50	同构 rnc100	同构 rnc300	同构 all	异构 rnc50	异构 rnc100	异构 rnc300	异构 all	全部
HEFT	1.0000	1.0000	1.0000	1.0000	1.0000	1.0000	1.0000	1.0000	1.0000
Random	1.5413	1.5963	1.5300	1.5558	3.0217	4.5014	4.3923	3.9718	2.7638
CPOP	1.0863	1.1056	1.1277	1.1065	1.2364	1.2454	1.3031	1.2616	1.1841
PEFT	1.0284	1.0240	1.0076	1.0200	0.8434	0.8658	0.9168	0.8753	0.9477
Ours	1.0029	1.0090	1.0212	1.0110	0.8236	0.8590	0.9369	0.8732	0.9421

综合全部 162 个测试场景，GNN-PPO 以 mean_ratio = 0.9421 在所有方法中取得最优，优于 PEFT（0.9477）、HEFT（1.0000）、CPOP（1.1841）和 Random（2.7638）。在同构环境下，GNN-PPO 与 HEFT 表现接近（1.0110 vs 1.0000），显著优于 Random 和 CPOP；在异构环境下，GNN-PPO 以 0.8732 超越 HEFT 约 12.7%，同样略优于 PEFT（0.8753）。从难度维度看，GNN-PPO 在 rnc50 和 rnc100 的中等规模场景上优势最为突出，在 rnc300 大规模场景上仍保持对 HEFT 和 CPOP 的明显优势。Random 在异构大 DAG 场景下表现极差（mean_ratio > 4.3），说明随着动作空间的扩大，随机策略迅速失效，而学习型方法能够有效应对这种复杂性增长。

2. 快速开始

2.1 安装

使用 uv（推荐）：

curl -LsSf https://astral.sh/uv/install.sh | sh
git clone https://gitlink.org.cn/IBY0zeT2tG/jysdqhxxdybdygjszyglddff.git dagrl
cd dagrl
conda create -n dagrl python=3.11
conda activate dagrl
uv sync
source .venv/bin/activate
pip install -e ".[dev]"

uv 读取 pyproject.toml 自动解析依赖。需要开发依赖（pytest 等）时使用 uv sync --dev。要求 Python ≥ 3.11, PyTorch ≥ 2.5, PyTorch Geometric ≥ 2.5。

2.2 数据准备

python tools/prepare_data.py \
    --data-root data/raw/grapheonrl \
    --variants rnc50 rnc100 rnc300 \
    --modes homogeneous heterogeneous

此命令扫描 data/raw/grapheonrl/ 下的所有 workflow JSON 文件，按基础图 ID 生成训练/验证/测试划分（默认 70/15/15），写入 data/splits/。

2.3 评测（无需训练）

python tools/eval.py \
    --split test \
    --methods heft cpop peft random gnn_ppo \
    --gnn-checkpoint outputs/checkpoints/ppo_best_actor.pt \
    --gnn-strategy greedy \
    --output outputs/evaluation/test_set

输出到 outputs/evaluation/test_set/：

文件	内容
summary.json	聚合指标（程序化消费）
per_instance.csv	每个场景的详细结果（Excel 分析）
report.md	可读性报告，含综合对比表和按 variant 分组统计
charts/*.png	柱状图和分组对比图

2.4 训练 GNN-PPO

# BC 预训练
python tools/train.py config/methods/gnn_ppo_bc.py

# PPO 微调
python tools/train.py config/methods/gnn_ppo_ppo.py

训练中断后恢复：在 config/methods/gnn_ppo_ppo.py 中添加一行 resume_from：

method = dict(
    type="GNNPPOMethod",
    resume_from="outputs/checkpoints/gnn_ppo",
    ppo=dict(total_updates=2000, ...),
)

自动检测目录中步数更靠后的 checkpoint，恢复权重、优化器状态与训练步数后继续。

2.5 单场景推理

python tools/infer.py \
    --workflow data/raw/grapheonrl/rnc50/heterogeneous/rand0000_hetero.json \
    --system data/raw/grapheonrl/rnc50/heterogeneous/system.json \
    --checkpoint outputs/checkpoints/ppo_best_actor.pt \
    --strategy beam_search --beam-width 5

3. 使用指南

3.1 项目结构

DAGRL/
├── dagrl/                         # Python 包（pip install -e .）
│   ├── registry.py                #   Registry 类 + METHOD / DATASET 注册表
│   ├── config.py                  #   Config.fromfile() + _base_ 多继承
│   ├── core/                      #   共享基础设施
│   │   ├── schema.py              #     Task / ResourceNode / WorkflowInstance
│   │   ├── loader.py              #     JSON → WorkflowInstance（含校验）
│   │   ├── engine.py              #     ScheduleEngine（EST/EFT/可行性/放置/验证）
│   │   ├── split.py               #     数据划分
│   │   ├── validator.py           #     调度校验
│   │   ├── export.py              #     调度导出
│   │   └── features.py            #     速度选择 / GPU 检测等工具函数
│   ├── methods/                   #   所有调度方法（平级，不相互依赖）
│   │   ├── base.py                #     Scenario + ScheduleResult + BaseMethod
│   │   ├── heft/ / cpop/ / peft/ / random/   # 启发式基线
│   │   ├── gnn_ppo/               #     GNN-PPO v1（GATv2 + Pair MLP）
│   │   └── gnn_ppo_v2/            #     GNN-PPO v2（GATv2 + Cross-Attention）
│   ├── datasets/                  #   场景管理
│   └── evaluation/                #   评测逻辑（Evaluator / Metrics / Reporter）
├── config/                        # Python 配置文件
│   ├── _base_/                    #   公共配置（数据集 / 运行时）
│   └── methods/                   #   每个方法一个可复现的训练配置
├── tools/                         # CLI 入口（prepare_data / train / eval / infer）
├── data/                          # 原始数据 + 划分文件
├── outputs/                       # 生成产物（checkpoints / logs / evaluation）
└── tests/                         # 132 项单元测试

3.2 评测多个方法

python tools/eval.py \
    --dataset grapheonrl \
    --split test \
    --methods heft cpop peft random gnn_ppo \
    --gnn-checkpoint outputs/checkpoints/ppo_best_actor.pt \
    --gnn-strategy beam_search \
    --reference heft \
    --output outputs/evaluation/test_set

参数	说明
--methods	参与评测的方法列表，大小写不敏感
--reference	参考基准（用于计算 makespan 比率，默认 heft）
--gnn-checkpoint	GNN-PPO 的 actor checkpoint 路径
--gnn-strategy	greedy（确定性 argmax）或 beam_search（K-beam 搜索）
--gnn-beam-width	Beam 宽度 K（默认 5）
--gnn-branch-factor	每 beam 分支数 B（默认 5）
--output	输出目录

3.3 训练 GNN-PPO

python tools/train.py config/methods/gnn_ppo_bc.py    # BC 预训练
python tools/train.py config/methods/gnn_ppo_ppo.py   # PPO 微调

命令行覆写配置：

python tools/train.py config/methods/gnn_ppo_ppo.py --device cpu --work-dir my_experiment

配置文件示例（config/methods/gnn_ppo_ppo.py）：

_base_ = ["../_base_/dataset_base.py", "../_base_/runtime_base.py"]

method = dict(
    type="GNNPPOMethod", device="cuda", seed=42,
    resume_from="outputs/checkpoints/gnn_ppo",
    ppo=dict(
        total_updates=2000, episodes_per_update=16,
        gamma=0.99, gae_lambda=0.95, clip_range=0.2,
        actor_learning_rate=0.0002, critic_learning_rate=0.0003,
        reward_type="completion",
        curriculum=[
            dict(phase=0, fraction=0.30, weights={"rnc50": 1.0, "rnc100": 0.0, "rnc300": 0.0}),
            dict(phase=1, fraction=0.40, weights={"rnc50": 0.5, "rnc100": 0.5, "rnc300": 0.0}),
            dict(phase=2, fraction=0.30, weights={"rnc50": 0.3, "rnc100": 0.4, "rnc300": 0.3}),
        ],
    ),
)

3.4 接入新方法

启发式方法只需实现 schedule(scenario)：

# dagrl/methods/greedy_eft/method.py
import math, time
from dagrl.core.engine import ScheduleEngine
from dagrl.methods.base import BaseMethod, ScheduleResult
from dagrl.registry import METHOD

@METHOD.register_module()
class GreedyEFTMethod(BaseMethod):
    name = "GreedyEFT"

    def __init__(self):
        self._engine = ScheduleEngine()

    def schedule(self, scenario) -> ScheduleResult:
        t0 = time.perf_counter()
        instance = scenario.instance
        self._engine.reset(instance)
        N = instance.num_nodes
        for _ in range(instance.num_tasks):
            ready = self._engine.ready_tasks()
            best_eft, best_task, best_node = math.inf, -1, -1
            for t_idx in ready:
                for n_idx in range(N):
                    if not self._engine.is_feasible(t_idx, n_idx): continue
                    est = self._engine.estimate(t_idx, n_idx)
                    if est.feasible and est.eft < best_eft:
                        best_eft, best_task, best_node = est.eft, t_idx, n_idx
            self._engine.place(best_task, best_node)
        validation = self._engine.validate()
        return ScheduleResult(
            scenario_id=scenario.id, method_name=self.name,
            makespan=self._engine.makespan(),
            status="feasible" if validation.valid else "invalid",
            schedule=self._engine.export_schedule(),
            wall_time_s=time.perf_counter() - t0,
        )

可训练方法额外实现 train()、save()、load()，通过 tools/train.py 启动训练。

3.5 接入新数据集

实现 BaseDataset 子类，通过 @DATASET.register_module() 注册。每个 Scenario 需包含 id（唯一标识）、instance（WorkflowInstance）和 metadata（自由标注字典）。格式类似 GrapheonRL 的数据可直接复用 GrapheonRLDataset 并调整目录扫描逻辑。

4. 参考

4.1 内置方法

方法	类型	说明
HEFT	启发式	Heterogeneous Earliest Finish Time（Topcuoglu 2002）
CPOP	启发式	Critical-Path-on-a-Processor（Topcuoglu 2002）
PEFT	启发式	Predict Earliest Finish Time — OCT 前瞻（Arabnejad 2014）
Random	基线	从 action_mask 均匀采样合法动作
GNN-PPO	学习型	GATv2 + Cross-Attention + BC + PPO，详见 §1.2
GNN-PPO-V2	学习型	在 v1 基础上引入 TaskResourceCrossAttention 双向交互

4.2 核心抽象

class BaseMethod(ABC):
    name: str                                       # 方法标识
    schedule(scenario) → ScheduleResult             # 必须实现
    train(train_scenarios, val_scenarios, cfg)      # 可选
    save(path) / load(path)                         # 可选

@dataclass
class ScheduleResult:
    scenario_id: str;  method_name: str;  makespan: float
    status: str        # "feasible" | "invalid"
    schedule: dict     # task_id → {node, start_time, finish_time}
    wall_time_s: float

@dataclass
class Scenario:
    id: str                    # "rnc50_homogeneous_rand0000"
    instance: WorkflowInstance # DAG + 资源
    metadata: dict             # {variant, mode, workflow_path, ...}

4.3 评测指标

指标	说明
mean_ratio	mean(M_method / M_reference)，< 1 表示优于基准
win_rate_vs_reference	ratio < 1 的占比
tie_rate_vs_reference	ratio ≈ 1 的占比
loss_rate_vs_reference	ratio > 1 的占比
valid_rate	调度方案合法的场景占比
by_variant	按 rnc50 / rnc100 / rnc300 分组统计
by_mode	按 homogeneous / heterogeneous 分组统计

4.4 测试

pytest tests/ -q   # 132 项测试，覆盖 loader / DAG / HEFT / CPOP / PEFT / reward / action_mask / checkpoint 等

License

MIT