deepseek-ai--deepep/README.md

Note

本文档由 WeHub 基于上游 README 翻译整理，属于社区翻译，非官方中文文档。
English · 原始项目 · 上游 README
原作者、版权与许可证归属以原始项目及本仓库 LICENSE 文件为准。

DeepEP

DeepEP (DeepEveryParallel) 是一款面向现代机器学习训练与推理的高性能通信库。该库目前聚焦于专家并行（Expert Parallelism，EP）——提供高吞吐、低延迟的全对全（all-to-all）GPU 内核（MoE dispatch 与 combine），并支持包括 FP8 在内的低精度；同时也提供面向流水线并行（Pipeline Parallelism，PP）、上下文并行（Context Parallelism，CP）以及远程内存访问（Engram）的实验性原语，均设计为占用零或极少量 SM。所有内核均通过轻量级即时编译（Just-In-Time，JIT）模块在运行时编译，安装时无需进行 CUDA 编译。

尽管设计轻量，DeepEP 的性能在各种配置下均可达到或超过硬件带宽上限。

News

• V2 发布：对 Expert Parallelism 进行了完整重构——相较 V1 以数倍更少的 SM 资源实现极致性能，同时支持显著更大的 scale-up 与 scale-out 域。V2 也已从 NVSHMEM 后端切换为更轻量的 NCCL Gin 后端。

New features

• Fully JIT（Just-In-Time 编译）
• NCCL Gin backend
  • Header-only & lightweight
  • Able to reuse existing NCCL communicators
• EPv2
  • 高吞吐与低延迟 API 统一为单一 ElasticBuffer 接口，并采用新的 GEMM layout
  • 支持更大的 scale-up 与 scale-out 域（最高 EP2048）
  • 解析式 SM 与 QP 数量计算——不再需要自动调优
  • hybrid 与 direct 模式均继续支持
  • 对于类 V3 的传统训练，SM 占用从 24 降至 4 - 6，同时保持相当或更优的性能
• 0 SM Engram（with RDMA）
• 0 SM PP（with RDMA）
• 0 SM CP（with Copy Engine）

Notes

• Buffer 大小占用大于 V1
• 不再支持 0 SM RDMA 低延迟 EP
• Engram、PP 与 CP 为实验性功能

Still on-going features

• Elastic GPU & CPU buffers：连续的虚拟地址空间，底层映射到 GPU 与 CPU 物理内存的混合体，从而实现全自动、透明的 Engram 或不均衡 EP
• 通过利用 EP replay 处理负载不均衡，以减小中间 buffer 大小
• 面向 DP 与 TP 的 all-gather 更新与 reduce-scatter 实现

有关传统 V1 文档（基于 NVSHMEM），请参阅 docs/legacy.md。

Performance

按照 V3 的配置，我们以每 batch 8K tokens、7168 隐藏维度、top 8 experts、FP8 dispatching 与 BF16 combining 进行测试，得到如下结果：

Arch	NIC type	Topo	Dispatch Bottleneck Bandwidth	Combine Bottleneck Bandwidth	#SMs
SM90	CX7	EP 8 x 2	90 GB/s (RDMA)	81 GB/s (RDMA)	12
SM90	CX7	EP 8 x 4	61 GB/s (RDMA)	61 GB/s (RDMA)	6
SM100	CX7	EP 8 x 2	90 GB/s (RDMA)	91 GB/s (RDMA)	12
SM100	N/A	EP 8	726 GB/s (NVLink)	740 GB/s (NVLink)	64 (Max perf)
SM100	N/A	EP 8	643 GB/s (NVLink)	675 GB/s (NVLink)	24 (Min #SM)

说明：结果为逻辑带宽。例如，在 EP 8 x 2 情形下，90 GB/s 实际包含本地 rank 流量。

与 V1 相比，V2 峰值性能最高可达 1.3 倍，同时最多可节省 4 倍 SM 数量。

我们暂时省略更大 EP 配置的结果，但鼓励感兴趣的用户自行 benchmark。根据我们的内部经验，我们预期内核在规模扩展时仍将持续饱和硬件带宽。

V1 性能数据请参阅 docs/legacy.md。

Quick start

Requirements

• Hopper (SM90) GPU，或其他支持 SM90 PTX ISA 的架构
• Python 3.8 及以上
• CUDA version
  • SM90 GPU 需 CUDA 12.3 及以上
• PyTorch 2.10 及以上
• NCCL 2.30.4 及以上
• 节点内通信需 NVLink
• 节点间通信需 RDMA 网络

Install NCCL dependency

我们建议使用 pip 安装 NCCL，以便 DeepEP 能在 Python 环境中自动定位它。可使用以下命令安装：

pip install "nvidia-nccl-cu13>=2.30.4" --no-deps

Install NVSHMEM dependency

DeepEP 也依赖 NVSHMEM 以支持传统方法。请参阅我们的 NVSHMEM Installation Guide 获取说明。

Development

# Build and make symbolic links for SO files
python setup.py build
# You may modify the specific SO names according to your own platform
ln -s build/lib.linux-x86_64-cpython-38/deep_ep_cpp.cpython-38-x86_64-linux-gnu.so

# Run test cases
# NOTES: you may modify the `init_dist` function in `tests/utils/envs.py`
# according to your own cluster settings, and launch into multiple nodes
python tests/elastic/test_ep.py
python tests/elastic/test_agrs.py
python tests/elastic/test_engram.py
python tests/elastic/test_pp.py

Installation

python setup.py install

然后，在你的 Python 项目中 import deep_ep，即可开始使用！

Interfaces and examples

Buffer initialization

在 V2 中，所有 EP 操作——高吞吐与低延迟——均统一在单一 ElasticBuffer 接口下。可通过直接指定 MoE 设置来初始化 buffer，最优 SM 与 QP 数量将以解析方式计算。

import torch
import torch.distributed as dist
from typing import Optional

from deep_ep import ElasticBuffer

# Communication buffer (will allocate at runtime)
_buffer: Optional[ElasticBuffer] = None

# Number of SMs to use for communication kernels (will be set at buffer creation)
_num_comm_sms: int = 0


def get_buffer(group: dist.ProcessGroup,
               num_max_tokens_per_rank: int,
               hidden: int,
               num_topk: int,
               num_experts: int,
               use_fp8_dispatch: bool = False) -> ElasticBuffer:
    """Initialize or retrieve the ElasticBuffer for EP communication."""
    global _buffer, _num_comm_sms

    # Check if we can reuse the existing buffer
    required_bytes = ElasticBuffer.get_buffer_size_hint(
        group, num_max_tokens_per_rank, hidden,
        num_topk=num_topk, use_fp8_dispatch=use_fp8_dispatch,
    )
    if _buffer is not None and _buffer.group == group and _buffer.num_bytes >= required_bytes:
        return _buffer

    # Allocate a new buffer with MoE settings
    # NOTES: V2 buffer size consumption is larger than V1
    _buffer = ElasticBuffer(
        group,
        num_max_tokens_per_rank=num_max_tokens_per_rank,
        hidden=hidden,
        num_topk=num_topk,
        use_fp8_dispatch=use_fp8_dispatch,
    )

    # V2 analytically calculates the optimal SM count — no more auto-tuning needed
    # You may also specify `num_sms` manually in dispatch/combine calls to override
    _num_comm_sms = _buffer.get_theoretical_num_sms(num_experts, num_topk)

    return _buffer

Example use in model training or inference prefilling

V2 将 dispatch 与 combine API 统一为单一 ElasticBuffer 接口。以下示例展示如何在训练（含反向传播）或推理 prefilling 中使用它们。

import torch
import torch.distributed as dist
from typing import Tuple, Union

from deep_ep import ElasticBuffer, EPHandle, EventOverlap


def dispatch_forward(x: Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
                     topk_idx: torch.Tensor, topk_weights: torch.Tensor,
                     num_experts: int,
                     num_max_tokens_per_rank: int,
                     expert_alignment: int = 1) -> \
        Tuple[Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
              torch.Tensor, torch.Tensor, EPHandle, EventOverlap]:
    """
    MoE dispatch: route tokens to the corresponding experts across all ranks.
    Supports both BF16 and FP8 (x as a tuple of [data, scale_factors]) inputs.
    """
    global _buffer, _num_comm_sms

    recv_x, recv_topk_idx, recv_topk_weights, handle, event = _buffer.dispatch(
        x,
        topk_idx=topk_idx,
        topk_weights=topk_weights,
        num_experts=num_experts,
        num_max_tokens_per_rank=num_max_tokens_per_rank,
        expert_alignment=expert_alignment,
        num_sms=_num_comm_sms,
        async_with_compute_stream=True,
    )

    # `handle` contains routing metadata for the subsequent combine call
    # `handle.num_recv_tokens_per_expert_list` provides per-expert token counts for GEMM
    # Use `event.current_stream_wait()` to synchronize the compute stream before using results
    return recv_x, recv_topk_idx, recv_topk_weights, handle, event


def dispatch_backward(grad_recv_x: torch.Tensor,
                      grad_recv_topk_weights: torch.Tensor,
                      handle: EPHandle) -> Tuple[torch.Tensor, torch.Tensor, EventOverlap]:
    """The backward pass of MoE dispatch is actually a combine."""
    global _buffer, _num_comm_sms

    combined_grad_x, combined_grad_topk_weights, event = _buffer.combine(
        grad_recv_x,
        handle=handle,
        topk_weights=grad_recv_topk_weights,
        num_sms=_num_comm_sms,
        async_with_compute_stream=True,
    )

    return combined_grad_x, combined_grad_topk_weights, event


def combine_forward(x: torch.Tensor,
                    handle: EPHandle) -> Tuple[torch.Tensor, EventOverlap]:
    """MoE combine: reduce expert outputs back to their original ranks."""
    global _buffer, _num_comm_sms

    combined_x, _, event = _buffer.combine(
        x,
        handle=handle,
        num_sms=_num_comm_sms,
        async_with_compute_stream=True,
    )

    return combined_x, event


def combine_backward(grad_combined_x: Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
                     handle: EPHandle) -> \
        Tuple[Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]], EventOverlap]:
    """The backward pass of MoE combine is actually a dispatch."""
    global _buffer, _num_comm_sms

    grad_x, _, _, _, event = _buffer.dispatch(
        grad_combined_x,
        handle=handle,
        num_sms=_num_comm_sms,
        async_with_compute_stream=True,
    )

    return grad_x, event

为实现通信-计算重叠（communication-computation overlap），请使用 EventOverlap 接口来管理通信流与计算流之间的依赖关系：

# After dispatch, overlap computation while communication is in-flight
recv_x, recv_topk_idx, recv_topk_weights, handle, event = dispatch_forward(...)

# ... do some independent computation here ...

# Wait for communication to finish before using results
event.current_stream_wait()

# Now safe to use recv_x, recv_topk_idx, recv_topk_weights

推理解码中的示例用法

在推理解码中，同样使用 ElasticBuffer。handle-caching 模式可在门控决策保持不变时跨迭代复用路由元数据，从而避免冗余的 CPU 同步。

import torch
from typing import Tuple, Optional, Union

from deep_ep import ElasticBuffer, EPHandle, EventOverlap


def decode_dispatch(x: Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
                    topk_idx: torch.Tensor, topk_weights: torch.Tensor,
                    num_experts: int,
                    num_max_tokens_per_rank: int,
                    cached_handle: Optional[EPHandle] = None) -> \
        Tuple[Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
              torch.Tensor, torch.Tensor, EPHandle, EventOverlap]:
    """
    MoE dispatch for inference decoding.
    If `cached_handle` is provided, the layout is reused without CPU synchronization.
    """
    global _buffer, _num_comm_sms

    if cached_handle is not None:
        # Reuse cached handle: skip layout recomputation and CPU sync
        recv_x, _, _, handle, event = _buffer.dispatch(
            x,
            handle=cached_handle,
            num_sms=_num_comm_sms,
            async_with_compute_stream=True,
        )
        return recv_x, cached_handle.topk_idx, None, handle, event

    recv_x, recv_topk_idx, recv_topk_weights, handle, event = _buffer.dispatch(
        x,
        topk_idx=topk_idx,
        topk_weights=topk_weights,
        num_experts=num_experts,
        num_max_tokens_per_rank=num_max_tokens_per_rank,
        num_sms=_num_comm_sms,
        async_with_compute_stream=True,
    )

    return recv_x, recv_topk_idx, recv_topk_weights, handle, event


def decode_combine(x: torch.Tensor,
                   handle: EPHandle) -> Tuple[torch.Tensor, EventOverlap]:
    """MoE combine for inference decoding."""
    global _buffer, _num_comm_sms

    combined_x, _, event = _buffer.combine(
        x,
        handle=handle,
        num_sms=_num_comm_sms,
        async_with_compute_stream=True,
    )

    return combined_x, event

环境变量

本库提供了一些可能有用的环境变量：

• 通用
  • EP_BUFFER_DEBUG: 0 或 1，打印缓冲区初始化、SM 近似与后端调试信息，默认 0
  • EP_SUPPRESS_NCCL_CHECK: 0 或 1，抑制 NCCL 版本不匹配检查，默认 0
  • EP_AVOID_RECORD_STREAM: 0 或 1，避免对输出张量执行 record_stream，默认 0
  • EP_NUM_TOPK_IDX_BITS: 整数，覆盖 top-k 索引编码的比特数，默认 0（auto）
• 网络
  • EP_NIC_NAME: 字符串，用于查询网卡属性的默认网卡名称，默认 mlx5_0
  • EP_OVERRIDE_RDMA_SL: 整数，覆盖用于流量隔离的 RDMA 服务等级索引
  • EP_DISABLE_GIN: 0 或 1，禁用 NCCL Gin 后端（回退到非 Gin 路径），默认 0
• JIT
  • EP_JIT_DEBUG: 0 或 1，打印 JIT 调试信息，默认 0
  • EP_JIT_CACHE_DIR: 字符串，已编译内核的缓存目录，默认 $HOME/.deep_ep
  • EP_JIT_NVCC_COMPILER: 字符串，NVCC 编译器路径；默认为 torch.utils.cpp_extension.CUDA_HOME
  • EP_JIT_CPP_STANDARD: 整数，C++ 标准版本，默认 20
  • EP_JIT_PRINT_COMPILER_COMMAND: 0 或 1，打印编译命令，默认 0
  • EP_JIT_PTXAS_VERBOSE: 0 或 1，显示详细的 PTXAS 输出，默认 0
  • EP_JIT_PTXAS_CHECK: 0 或 1，断言已编译内核不使用本地内存，默认 0
  • EP_JIT_WITH_LINEINFO: 0 或 1，为性能分析工具嵌入源码行信息，默认 0
  • EP_JIT_DUMP_ASM: 0 或 1，同时转储 PTX 与 SASS，默认 0
  • EP_JIT_DUMP_PTX: 0 或 1，转储 PTX 输出，默认 0
  • EP_JIT_DUMP_SASS: 0 或 1，转储 SASS 输出，默认 0
• 调试与性能分析
  • EP_GIN_GDAKI_DEBUG: 0 或 1，启用 NCCL Gin GDAKI 调试输出，默认 0
  • EP_USE_NVIDIA_TOOLS: 0 或 1，在外部 NVIDIA 工具下运行时跳过内部性能分析，默认 0
  • EP_DISABLE_BARRIER_PROFILING: 0 或 1，在基准测试中禁用基于 barrier 的通信性能分析，默认 0
• 构建
  • EP_NCCL_ROOT_DIR: 字符串，NCCL 安装目录路径；若未设置，则从 Python 环境自动检测
  • EP_NVSHMEM_ROOT_DIR: 字符串，NVSHMEM 安装目录路径；若未设置，则从 Python 环境自动检测
  • TORCH_CUDA_ARCH_LIST: 字符串，目标 CUDA 架构列表，例如 "9.0"
  • DISABLE_SM90_FEATURES: 0 或 1，为旧版方法禁用 SM90 特性，默认 0
  • DISABLE_AGGRESSIVE_PTX_INSTRS: 0 或 1，在旧版方法中禁用激进的 load/store 指令，默认 0

部分环境变量是持久化的：它们在构建时被捕获，并作为默认值写入已安装的包中。在导入时，除非被当前环境变量覆盖，否则会自动应用这些默认值。持久化变量包括：EP_JIT_CACHE_DIR、EP_JIT_PRINT_COMPILER_COMMAND、EP_NUM_TOPK_IDX_BITS、EP_NCCL_ROOT_DIR。

更多细节请参阅测试代码，或查阅相应的 Python 文档。

网络配置

DeepEP 已在 InfiniBand 网络上完成充分测试。不过，理论上它也兼容融合以太网上的 RDMA（RDMA over Converged Ethernet，RoCE）。

流量隔离

InfiniBand 通过虚拟通道（Virtual Lanes，VL）支持流量隔离。

为避免不同类型流量之间的相互干扰，我们建议按以下方式将工作负载划分到不同的虚拟通道：

• 专家并行（expert-parallel）工作负载
• 其他工作负载

对于 DeepEP V2，可通过设置 sl_idx 参数或 EP_OVERRIDE_RDMA_SL 环境变量来控制虚拟通道分配。

自适应路由

自适应路由（adaptive routing）是 InfiniBand 交换机提供的高级路由特性，可将流量均匀分布到多条路径上。尽管自适应路由会引入额外延迟，我们仍建议在所有网络负载条件下启用它。

拥塞控制

拥塞控制已禁用，因为它会损害最大带宽。若在某些场景下拥塞不可避免，我们建议将这些工作负载分配到低优先级虚拟通道。

PCI 原子模式

若硬件支持，建议使用以下命令设置网卡的 PCI_ATOMIC_MODE，以提升 RDMA 原子操作性能：

sudo mlxconfig -y -d mlx5_$i set PCI_ATOMIC_MODE=4

实验分支

• Zero-copy
  • 消除 PyTorch 张量与通信缓冲区之间的拷贝，可显著降低常规模核的 SM 占用
  • 该 PR 由**腾讯网络平台部（Tencent Network Platform Department）**编写
• Eager
  • 使用低延迟协议可消除 RDMA 原子操作（atomic OPs）引入的额外 RTT 延迟
• Hybrid-EP
  • 基于 TMA 指令的新后端实现，实现最小 SM 占用并支持更大的 NVLink 域
  • 针对单批次场景的细粒度通信-计算重叠
  • 为非 NVLink 环境提供 PCIe 内核支持
  • 支持 NVFP4 数据类型
• AntGroup-Opt
  • 该优化系列由**蚂蚁集团网络平台部（AntGroup Network Platform Department）**编写
  • Normal-SMFree 通过将通信内核执行与 NIC token 传输解耦，从 RDMA 路径中消除 SM 占用，从而释放 SM 用于计算
  • LL-SBO 通过信令机制将 Down GEMM 计算与 Combine Send 通信重叠，以降低端到端延迟
  • LL-Layered 使用 rail 优化转发与数据合并优化跨节点 LL 算子通信，以降低延迟
• Mori-EP
  • 基于 MORI 后端（低延迟模式）的 ROCm/AMD GPU 支持
• nvDev
  • 基于 V2 的分支，集成最新 CUDA 特性，例如 Compute Fabric Transport（CFT），可在较小 token 规模下带来更优延迟。

社区分支

• uccl/uccl-ep - 支持在异构 GPU（如 Nvidia、AMD）和 NIC（如 EFA、Broadcom、CX7）上运行 DeepEP
• Infrawaves/DeepEP_ibrc_dual-ports_multiQP - 在 IBRC 传输中添加多 QP（Queue Pair）方案及双端口 NIC 支持
• antgroup/DeepXTrace - 用于高效、精确定位慢 rank 的诊断分析工具
• ROCm/mori - AMD 面向性能关键型 AI 工作负载（如 Wide EP、KVCache transfer、Collectives）的下一代通信库

致谢

DeepEP V2 基于 NCCL Gin backend 构建。感谢 @sjeaugey、@pakmarkthub、@sb17v、@xiaofanl-nvidia 以及 NCCL 团队的支持！

许可证

本代码仓库在 the MIT License 下发布。

引用

@misc{deepep2025,
      title={DeepEP: an efficient expert-parallel communication library},
      author={Chenggang Zhao and Shangyan Zhou and Liyue Zhang and Chengqi Deng and Zhean Xu and Yuxuan Liu and Kuai Yu and Jiashi Li and Liang Zhao},
      year={2025},
      publisher = {GitHub},
      howpublished = {\url{https://github.com/deepseek-ai/DeepEP}},
}