DeepEP 测试体系总览

flowchart TB
    subgraph L4["🔺 第4层：集成测试 — 端到端正确性 + 性能"]
        T4A["test_ep.py (609L)<br/>Dispatch/Combine 全流程"]
        T4B["test_agrs.py (202L)<br/>AllGather/ReduceScatter"]
        T4C["test_pp.py (139L)<br/>Pipeline Parallel"]
    end

    subgraph L3["第3层：Legacy V1 测试 — 三种通信模式"]
        T3A["test_internode.py (395L)<br/>NVLink + RDMA"]
        T3B["test_intranode.py (311L)<br/>NVLink only"]
        T3C["test_low_latency.py (332L)<br/>IBGDA RDMA"]
    end

    subgraph L2["第2层：基础原语测试 — 单个机制验证"]
        T2A["test_barrier.py (62L)<br/>GPU Barrier 同步"]
        T2B["test_engram.py (124L)<br/>RDMA 远端内存"]
        T2C["test_gate.py (57L)<br/>MoE 负载生成"]
    end

    subgraph L1["第1层：测试基础设施 — 通用工具"]
        U1["envs.py<br/>分布式初始化"]
        U2["testing.py<br/>Benchmark 工具"]
        U3["refs.py<br/>PyTorch 参考实现"]
        U4["math.py<br/>FP8 编解码"]
        U5["event.py<br/>CUDA Event 封装"]
        U6["comm.py<br/>NCCL Communicator"]
    end

    L4 --> L3 --> L2 --> L1

# test_ep.py 中的典型验证模式
assert bitwise compare(dispatch_result, ref_dispatch_result)  # 不是 torch.allclose!

flowchart LR
    A["生成随机操作序列<br/>create/ag/fetch/destroy"] --> B["执行操作<br/>随机批量大小"]
    B --> C["确定性参考<br/>种子 = 42 + round×43 + rank"]
    C --> D["Bitwise 比对"]
    D --> E{"通过?"}
    E -->|Yes| F["Profiling<br/>bench_kineto"]
    E -->|No| G["立即失败<br/>报告不匹配位置"]

flowchart TD
    A["__main__: argparse 解析"] --> B["torch.multiprocessing.spawn<br/>启动 N 个测试进程"]
    B --> C["init_dist: NCCL 初始化<br/>默认 BF16 + CUDA"]
    C --> D["构造 Buffer<br/>ElasticBuffer / Buffer"]
    D --> E["enumerate / generate<br/>参数空间 / 随机序列"]
    E --> F["Reference 计算<br/>PyTorch CPU/GPU 参考实现"]
    F --> G["Kernel 执行<br/>dispatch / combine / agrs"]
    G --> H["正确性验证<br/>bitwise compare"]
    H --> I["性能基准<br/>bench_kineto + flush_l2"]
    I --> J{"do_pressure_test?"}
    J -->|Yes| K["循环 seed in range(1e9)<br/>重建 buffer 重复"]
    J -->|No| L["buffer.destroy()"]
    K --> L
    L --> M["dist.destroy_process_group()"]