DeepEP

DeepEP 是一款专为混合专家模型（MoE）和专家并行（EP）设计的通信库。它提供了高吞吐量、低延迟的全对全 GPU 内核，也称为 MoE 调度与组合。该库还支持低精度运算，包括 FP8。

为了配合 DeepSeek-V3 论文中提出的分组受限门控算法，DeepEP 提供了一组针对非对称域带宽转发优化的内核，例如从 NVLink 域向 RDMA 域的数据转发。这些内核具有高吞吐量，适用于训练和推理预填充任务。此外，它们还支持流式多处理器（SM）数量控制。

对于对延迟敏感的推理解码，DeepEP 包含一组纯 RDMA 的低延迟内核，以最大限度地减少延迟。该库还引入了一种基于钩子的通信-计算重叠方法，不会占用任何 SM 资源。

注意：本库中的实现可能与 DeepSeek-V3 论文中的描述存在一些细微差异。

性能

普通内核与 NVLink 和 RDMA 转发

我们在 H800 上测试了普通内核（NVLink 最大带宽约 160 GB/s），每台机器都连接到 CX7 InfiniBand 400 Gb/s RDMA 网卡（最大带宽约 50 GB/s）。我们遵循 DeepSeek-V3/R1 预训练设置（每批 4096 个 token，隐藏层大小 7168，top-4 分组，top-8 专家，FP8 调度和 BF16 组合）。

类型	调度 #EP	瓶颈带宽	组合 #EP	瓶颈带宽
节点内	8	153 GB/s (NVLink)	8	158 GB/s (NVLink)
节点间	16	43 GB/s (RDMA)	16	43 GB/s (RDMA)
节点间	32	58 GB/s (RDMA)	32	57 GB/s (RDMA)
节点间	64	51 GB/s (RDMA)	64	50 GB/s (RDMA)

新闻（2025.04.22）：在腾讯网络平台部门的优化下，性能提升了高达 30%，详情请参阅 #130。感谢他们的贡献！

低延迟内核与纯 RDMA

我们使用 H800 测试了低延迟内核，每台机器都连接到 CX7 InfiniBand 400 Gb/s RDMA 网卡（最大带宽约 50 GB/s）。我们遵循典型的 DeepSeek-V3/R1 生产环境设置（每批 128 个 token，隐藏层大小 7168，top-8 专家，FP8 调度和 BF16 组合）。

调度 #EP	延迟	RDMA 带宽	组合 #EP	延迟	RDMA 带宽
8	77 us	98 GB/s	8	114 us	127 GB/s
16	118 us	63 GB/s	16	195 us	74 GB/s
32	155 us	48 GB/s	32	273 us	53 GB/s
64	173 us	43 GB/s	64	314 us	46 GB/s
128	192 us	39 GB/s	128	369 us	39 GB/s
256	194 us	39 GB/s	256	360 us	40 GB/s

新闻（2025.06.05）：低延迟内核现在尽可能利用 NVLink，详情请参阅 #173。感谢他们的贡献！

快速入门

要求

• Ampere（SM80）、Hopper（SM90）GPU，或支持 SM90 PTX ISA 的其他架构
• Python 3.8 及以上版本
• CUDA 版本：
  • SM80 GPU 需要 CUDA 11.0 及以上版本
  • SM90 GPU 需要 CUDA 12.3 及以上版本
• PyTorch 2.1 及以上版本
• 节点内通信需使用 NVLink
• 节点间通信需使用 RDMA 网络

下载并安装 NVSHMEM 依赖

DeepEP 还依赖于 NVSHMEM。请参考我们的 NVSHMEM 安装指南 获取说明。

开发

# 构建并创建 SO 文件的符号链接
NVSHMEM_DIR=/path/to/installed/nvshmem python setup.py build
# 您可以根据自己的平台修改具体的 SO 文件名
ln -s build/lib.linux-x86_64-cpython-38/deep_ep_cpp.cpython-38-x86_64-linux-gnu.so

# 运行测试用例
# 注意：您可以根据自己的集群设置修改 `tests/utils.py` 中的 `init_dist` 函数，并在多个节点上启动
python tests/test_intranode.py
python tests/test_internode.py
python tests/test_low_latency.py

安装

NVSHMEM_DIR=/path/to/installed/nvshmem python setup.py install

安装环境变量

• NVSHMEM_DIR：NVSHMEM 目录的路径，若未指定则禁用所有节点间和低延迟功能
• DISABLE_SM90_FEATURES：0 或 1，是否禁用 SM90 功能，对于 SM90 设备或 CUDA 11 需要设置此变量
• TORCH_CUDA_ARCH_LIST：目标架构列表，例如 TORCH_CUDA_ARCH_LIST="9.0"
• DISABLE_AGGRESSIVE_PTX_INSTRS：0 或 1，是否禁用激进的加载/存储指令，详情请参阅 未定义行为 PTX 使用

然后，在您的 Python 项目中导入 deep_ep，即可开始使用！

网络配置

DeepEP 已在 InfiniBand 网络上进行了全面测试。然而，理论上它也兼容融合以太网上的 RDMA（RoCE）。

流量隔离

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

为防止不同类型流量之间的干扰，我们建议将工作负载划分到不同的虚拟通道中，如下所示：

• 使用普通内核的工作负载
• 使用低延迟内核的工作负载
• 其他工作负载

对于 DeepEP，您可以通过设置 NVSHMEM_IB_SL 环境变量来控制虚拟通道的分配。

自适应路由

自适应路由是 InfiniBand 交换机提供的一项高级路由功能，可以将流量均匀地分配到多条路径上。启用自适应路由可以完全消除由路由冲突引起的网络拥塞，但也会引入额外的延迟。我们建议采用以下配置以获得最佳性能：

• 在网络负载较重的环境中启用自适应路由
• 在网络负载较轻的环境中使用静态路由

拥塞控制

由于我们在生产环境中未观察到显著的拥塞现象，因此拥塞控制功能已被禁用。

接口与示例

模型训练或推理预填充中的使用示例

普通内核可用于模型训练或推理预填充阶段（不包含反向传播部分），如下示例代码所示。

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

from deep_ep import Buffer, EventOverlap

# 通信缓冲区（将在运行时分配）
_buffer: Optional[Buffer] = None

# 设置使用的 SM 数量

# 注释：这是一个静态变量
Buffer.set_num_sms(24)


# 您可以在框架初始化时调用此函数
def get_buffer(group: dist.ProcessGroup, hidden_bytes: int) -> Buffer:
    global _buffer

    # 注释：您也可以用所有测试得到的自动调优结果替换 `get_*_config`
    num_nvl_bytes, num_rdma_bytes = 0, 0
    for config in (Buffer.get_dispatch_config(group.size()), Buffer.get_combine_config(group.size())):
        num_nvl_bytes = max(config.get_nvl_buffer_size_hint(hidden_bytes, group.size()), num_nvl_bytes)
        num_rdma_bytes = max(config.get_rdma_buffer_size_hint(hidden_bytes, group.size()), num_rdma_bytes)

    # 如果缓冲区不存在或大小不足，则分配一个新的缓冲区
    if _buffer is None or _buffer.group != group or _buffer.num_nvl_bytes < num_nvl_bytes or _buffer.num_rdma_bytes < num_rdma_bytes:
        _buffer = Buffer(group, num_nvl_bytes, num_rdma_bytes)
    return _buffer


def get_hidden_bytes(x: torch.Tensor) -> int:
    t = x[0] if isinstance(x, tuple) else x
    return t.size(1) * max(t.element_size(), 2)


def dispatch_forward(x: Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
                     topk_idx: torch.Tensor, topk_weights: torch.Tensor,
                     num_experts: int, previous_event: Optional[EventOverlap] = None) -> \
        Tuple[Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]], torch.Tensor, torch.Tensor, List, Tuple, EventOverlap]:
    # 注释：可选的 `previous_event` 表示您希望将其作为调度内核依赖项的 CUDA 事件，
    # 这在通信与计算重叠时可能很有用。更多信息请参阅 `Buffer.dispatch` 的文档。
    global _buffer

    # 在实际调度之前计算布局
    num_tokens_per_rank, num_tokens_per_rdma_rank, num_tokens_per_expert, is_token_in_rank, previous_event = \
        _buffer.get_dispatch_layout(topk_idx, num_experts,
                                    previous_event=previous_event, async_finish=True,
                                    allocate_on_comm_stream=previous_event is not None)
    # 执行 MoE 调度
    # 注释：CPU 将等待 GPU 的信号到达，因此这与 CUDA 图不兼容。
    # 除非您指定 `num_worst_tokens`，但该标志仅适用于节点内部。
    # 更高级的用法请参阅 `dispatch` 函数的文档。
    recv_x, recv_topk_idx, recv_topk_weights, num_recv_tokens_per_expert_list, handle, event = \
        _buffer.dispatch(x, topk_idx=topk_idx, topk_weights=topk_weights,
                         num_tokens_per_rank=num_tokens_per_rank, num_tokens_per_rdma_rank=num_tokens_per_rdma_rank,
                         is_token_in_rank=is_token_in_rank, num_tokens_per_expert=num_tokens_per_expert,
                         previous_event=previous_event, async_finish=True,
                         allocate_on_comm_stream=True)
    # 关于事件管理，请参阅 `EventOverlap` 类的文档。
    return recv_x, recv_topk_idx, recv_topk_weights, num_recv_tokens_per_expert_list, handle, event


def dispatch_backward(grad_recv_x: torch.Tensor, grad_recv_topk_weights: torch.Tensor, handle: Tuple) -> \
        Tuple[torch.Tensor, torch.Tensor, EventOverlap]:
    global _buffer

    # MoE 调度的反向过程实际上是组合操作。
    # 更高级的用法请参阅 `combine` 函数的文档。
    combined_grad_x, combined_grad_recv_topk_weights, event = \
        _buffer.combine(grad_recv_x, handle, topk_weights=grad_recv_topk_weights, async_finish=True)

    # 关于事件管理，请参阅 `EventOverlap` 类的文档。
    return combined_grad_x, combined_grad_recv_topk_weights, event


def combine_forward(x: torch.Tensor, handle: Tuple, previous_event: Optional[EventOverlap] = None) -> \
        Tuple[torch.Tensor, EventOverlap]:
    global _buffer

    # 执行 MoE 组合操作
    # 更高级的用法请参阅 `combine` 函数的文档。
    combined_x, _, event = _buffer.combine(x, handle, async_finish=True, previous_event=previous_event,
                                           allocate_on_comm_stream=previous_event is not None)

    # 关于事件管理，请参阅 `EventOverlap` 类的文档。
    return combined_x, event


def combine_backward(grad_combined_x: Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
                     handle: Tuple, previous_event: Optional[EventOverlap] = None) -> \
        Tuple[Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]], EventOverlap]:
    global _buffer

    # MoE 组合的反向过程实际上是调度操作。
    # 更高级的用法请参阅 `dispatch` 函数的文档。
    grad_x, _, _, _, _, event = _buffer.dispatch(grad_combined_x, handle=handle, async_finish=True,
                                                 previous_event=previous_event,
                                                 allocate_on_comm_stream=previous_event is not None)

    # 关于事件管理，请参阅 `EventOverlap` 类的文档。
    return grad_x, event

此外，在调度函数中，我们可能不知道当前进程需要接收多少个令牌。因此，会涉及 CPU 对 GPU 接收数量信号的隐式等待，如下图所示。

推理解码中的使用示例

低延迟内核可以用于推理解码阶段，如下方示例代码所示。

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

from deep_ep import Buffer

# 通信缓冲区（将在运行时分配）
# 注释：低延迟内核没有 SM 控制 API
_buffer: Optional[Buffer] = None

# 您可以在框架初始化时调用此函数
def get_buffer(group: dist.ProcessGroup, num_max_dispatch_tokens_per_rank: int, hidden: int, num_experts: int) -> Buffer:
    # 注意：低延迟模式会比普通模式占用更多空间
    # 因此我们建议 `num_max_dispatch_tokens_per_rank`（解码引擎中的实际批大小）应小于 256
    global _buffer
    num_rdma_bytes = Buffer.get_low_latency_rdma_size_hint(num_max_dispatch_tokens_per_rank, hidden, group.size(), num_experts)

    # 如果缓冲区不存在或大小不足，则分配一个新的缓冲区
    if _buffer is None or _buffer.group != group or not _buffer.low_latency_mode or _buffer.num_rdma_bytes < num_rdma_bytes:
        # 注意：为了获得最佳性能，QP 数量**必须**等于本地专家的数量
        assert num_experts % group.size() == 0
        _buffer = Buffer(group, 0, num_rdma_bytes, low_latency_mode=True, num_qps_per_rank=num_experts // group.size())
    return _buffer


def low_latency_dispatch(hidden_states: torch.Tensor, topk_idx: torch.Tensor, num_max_dispatch_tokens_per_rank: int, num_experts: int):
    global _buffer

    # 执行 MoE 调度，兼容 CUDA 图（但您在重放时可能需要恢复部分缓冲区状态）
    recv_hidden_states, recv_expert_count, handle, event, hook = \
        _buffer.low_latency_dispatch(hidden_states, topk_idx, num_max_dispatch_tokens_per_rank, num_experts,
                                     async_finish=False, return_recv_hook=True)

    # 注意：只有调用 `hook()` 时，实际张量才会被接收，
    # 这对于双批次重叠很有用，而且**不会占用任何 SM**。
    # 如果您不想进行重叠，请将 `return_recv_hook` 设置为 False。
    # 稍后，您可以使用我们的 GEMM 库以这种特定格式进行计算。
    return recv_hidden_states, recv_expert_count, handle, event, hook


def low_latency_combine(hidden_states: torch.Tensor,
                        topk_idx: torch.Tensor, topk_weights: torch.Tensor, handle: Tuple):
    global _buffer

    # 执行 MoE 组合，兼容 CUDA 图（但您在重放时可能需要恢复部分缓冲区状态）
    combined_hidden_states, event_overlap, hook = \
        _buffer.low_latency_combine(hidden_states, topk_idx, topk_weights, handle,
                                    async_finish=False, return_recv_hook=True)

    # 注意：行为与调度内核中描述的一致
    return combined_hidden_states, event_overlap, hook

对于两个微批次的重叠，可以参考下图。借助我们的接收钩子接口，RDMA 网络流量会在后台进行，而不会占用计算部分的任何 GPU SM。不过请注意，重叠的部分是可以调整的，即注意力/调度/MoE/组合这四个部分的执行时间不一定完全相同。您可以根据自己的工作负载调整各个阶段的设置。

路线图

• 支持 AR
• 重构低延迟模式下的 AR 代码
• 支持 A100（仅节点内）
• 为低延迟调度内核支持 BF16
• 支持 NVLink 协议用于节点内低延迟内核
• 使用 TMA 复制代替 LD/ST
  • 节点内内核
  • 节点间内核
  • 低延迟内核
• 实现无 SM 内核及相应重构
• 完全移除未定义行为的 PTX 指令

注意事项

更简单的潜在整体设计

当前的 DeepEP 实现使用队列作为通信缓冲区，虽然节省了内存，但也引入了复杂性和潜在的死锁问题。如果您基于 DeepEP 实现自己的版本，建议使用固定大小、按最大容量分配的缓冲区，以简化设计并提升性能。有关这一替代方案的详细讨论，请参阅：https://github.com/deepseek-ai/DeepEP/issues/39。

未定义行为的 PTX 使用

• 为了追求极致性能，我们发现并使用了一种未定义行为的 PTX 用法：利用只读 PTX ld.global.nc.L1::no_allocate.L2::256B 来读取易失性数据。PTX 修饰符 .nc 表示使用非相干缓存。但在 Hopper 架构上，通过结合 .L1::no_allocate，我们测试确认其正确性有保障，并且性能显著提升。我们推测原因可能是：非相干缓存与 L1 缓存统一，而 L1 修饰符不仅仅是一个提示，更是一种强约束，因此只要 L1 中没有脏数据，就能保证正确性。
• 最初，由于 NVCC 无法自动展开易失性读取的 PTX 指令，我们尝试使用 __ldg（即 ld.nc）。即使与手动展开的易失性读取相比，它的速度也快得多（很可能得益于编译器的额外优化）。然而，结果可能会不正确或包含脏数据。在查阅 PTX 文档后，我们发现 Hopper 架构上的 L1 和非相干缓存是统一的。我们推测，.L1::no_allocate 可能会解决这个问题，从而促成了这一发现。
• 如果您发现内核在某些其他平台上无法正常工作，可以在 setup.py 中添加 DISABLE_AGGRESSIVE_PTX_INSTRS=1 来禁用该功能，或者提交一个问题。

在您的集群上进行自动调优

为了在您的集群上获得更好的性能，我们建议运行所有测试，并采用最佳的自动调优配置。默认配置是在 DeepSeek 的内部集群上优化得到的。

许可证

本代码库采用 MIT 许可证 发布，但引用 NVSHMEM 的代码（包括 csrc/kernels/ibgda_device.cuh 和 third-party/nvshmem.patch）则适用 NVSHMEM SLA。

实验性分支

• 零拷贝
  • 移除了 PyTorch 张量与通信缓冲区之间的拷贝，从而显著减少了常规内核对 SM 的占用。
  • 该 PR 由 腾讯网络平台部 贡献。
• Eager
  • 通过使用低延迟协议，消除了 RDMA 原子操作引入的额外 RTT 延迟。
• Hybrid-EP
  • 一种新的后端实现，利用 TMA 指令以最小化 SM 使用并支持更大的 NVLink 域。
  • 针对单批次场景实现了细粒度的通信-计算重叠。
  • 支持非 NVLink 环境下的 PCIe 内核。
  • 支持 NVFP4 数据类型。
• AntGroup-Opt
  • 该优化系列由 蚂蚁集团网络平台部 贡献。
  • Normal-SMFree 通过解耦通信内核执行与 NIC 令牌传输，将 SM 从 RDMA 路径中解放出来，从而释放 SM 用于计算。
  • LL-SBO 利用信号机制将 Down GEMM 计算与 Combine Send 通信重叠，以降低端到端延迟。
  • LL-Layered 采用轨道优化的转发和数据合并策略，优化跨节点 LL 算子的通信，进一步降低延迟。
• Mori-EP
  • 基于 MORI 后端（低延迟模式）提供的 ROCm/AMD GPU 支持。

社区分叉

• uccl/uccl-ep - 允许在异构 GPU（如 Nvidia、AMD）和 NIC（如 EFA、Broadcom、CX7）上运行 DeepEP。
• Infrawaves/DeepEP_ibrc_dual-ports_multiQP - 在 IBRC 传输中增加了多 QP 解决方案和双端口 NIC 支持。
• antgroup/DeepXTrace - 一款诊断分析工具，用于高效、精准地定位运行缓慢的节点。
• ROCm/mori - AMD 面向高性能 AI 工作负载的新一代通信库（如 Wide EP、KVCache 传输、集体通信）。

引用

如果您使用本代码库或认为我们的工作具有价值，请引用以下内容：

@misc{deepep2025,
      title={DeepEP：高效的专家并行通信库},
      author={Chenggang Zhao 和 Shangyan Zhou 和 Liyue Zhang 和 Chengqi Deng 和 Zhean Xu 和 Yuxuan Liu 和 Kuai Yu 和 Jiashi Li 和 Liang Zhao},
      year={2025},
      publisher = {GitHub},
      howpublished = {\url{https://github.com/deepseek-ai/DeepEP}},
}

DeepEP 快速上手指南

DeepEP 是专为混合专家模型（MoE）和专家并行（EP）设计的高性能通信库。它提供高吞吐、低延迟的 All-to-All GPU 内核（即 MoE 的 Dispatch 和 Combine），支持 FP8 等低精度运算，并针对 NVLink 与 RDMA 跨域传输进行了优化，适用于训练及推理场景。

环境准备

系统要求

• GPU 架构：Ampere (SM80)、Hopper (SM90) 或其他支持 SM90 PTX ISA 的架构。
• Python：3.8 及以上版本。
• PyTorch：2.1 及以上版本。
• CUDA 版本：
  • SM80 GPU：CUDA 11.0+
  • SM90 GPU：CUDA 12.3+
• 网络硬件：
  • 节点内通信：需配备 NVLink。
  • 节点间通信：需配备 RDMA 网络（推荐 InfiniBand，理论上兼容 RoCE）。

前置依赖

DeepEP 强依赖 NVSHMEM。在编译安装前，请务必先安装 NVSHMEM。

• 请参考官方提供的 NVSHMEM 安装指南 完成依赖安装。
• 记录 NVSHMEM 的安装路径，后续安装步骤中需要使用。

安装步骤

1. 克隆代码

git clone https://github.com/deepseek-ai/DeepEP.git
cd DeepEP

2. 编译与安装

设置 NVSHMEM_DIR 环境变量指向你的 NVSHMEM 安装目录，然后执行安装命令。

# 将 /path/to/installed/nvshmem 替换为实际路径
export NVSHMEM_DIR=/path/to/installed/nvshmem

# 执行安装
python setup.py install

可选配置环境变量

根据具体硬件环境，你可能需要设置以下变量：

• DISABLE_SM90_FEATURES: 设为 1 可禁用 SM90 特性（适用于 SM90 设备但使用 CUDA 11 的情况）。
• TORCH_CUDA_ARCH_LIST: 指定目标架构列表，例如 export TORCH_CUDA_ARCH_LIST="9.0"。
• DISABLE_AGGRESSIVE_PTX_INSTRS: 设为 1 可禁用激进的加载/存储指令（用于解决潜在的未定义行为问题）。

验证安装（开发模式）

如需运行测试用例验证安装是否成功（需修改 tests/utils.py 中的 init_dist 以适配集群配置）：

python tests/test_intranode.py
python tests/test_internode.py
python tests/test_low_latency.py

基本使用

以下示例展示了如何在模型训练或推理预填充（Prefilling）阶段使用 DeepEP 进行 MoE 的 Dispatch 和 Combine 操作。

核心代码示例

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

from deep_ep import Buffer, EventOverlap

# 全局通信缓冲區 (运行时分配)
_buffer: Optional[Buffer] = None

# 设置使用的 SM 数量 (静态变量)
Buffer.set_num_sms(24)

def get_buffer(group: dist.ProcessGroup, hidden_bytes: int) -> Buffer:
    """获取或初始化通信缓冲区"""
    global _buffer

    num_nvl_bytes, num_rdma_bytes = 0, 0
    # 计算所需的缓冲区大小
    for config in (Buffer.get_dispatch_config(group.size()), Buffer.get_combine_config(group.size())):
        num_nvl_bytes = max(config.get_nvl_buffer_size_hint(hidden_bytes, group.size()), num_nvl_bytes)
        num_rdma_bytes = max(config.get_rdma_buffer_size_hint(hidden_bytes, group.size()), num_rdma_bytes)

    # 如果缓冲区不存在或大小不足，则重新分配
    if _buffer is None or _buffer.group != group or _buffer.num_nvl_bytes < num_nvl_bytes or _buffer.num_rdma_bytes < num_rdma_bytes:
        _buffer = Buffer(group, num_nvl_bytes, num_rdma_bytes)
    return _buffer

def dispatch_forward(x: Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]],
                     topk_idx: torch.Tensor, topk_weights: torch.Tensor,
                     num_experts: int, previous_event: Optional[EventOverlap] = None):
    """
    执行 MoE Dispatch (前向)
    返回: recv_x, recv_topk_idx, recv_topk_weights, num_recv_tokens_per_expert_list, handle, event
    """
    global _buffer
    
    # 1. 计算布局 (Layout)
    num_tokens_per_rank, num_tokens_per_rdma_rank, num_tokens_per_expert, is_token_in_rank, previous_event = \
        _buffer.get_dispatch_layout(topk_idx, num_experts,
                                    previous_event=previous_event, async_finish=True,
                                    allocate_on_comm_stream=previous_event is not None)
    
    # 2. 执行 Dispatch
    recv_x, recv_topk_idx, recv_topk_weights, num_recv_tokens_per_expert_list, handle, event = \
        _buffer.dispatch(x, topk_idx=topk_idx, topk_weights=topk_weights,
                         num_tokens_per_rank=num_tokens_per_rank, num_tokens_per_rdma_rank=num_tokens_per_rdma_rank,
                         is_token_in_rank=is_token_in_rank, num_tokens_per_expert=num_tokens_per_expert,
                         previous_event=previous_event, async_finish=True,
                         allocate_on_comm_stream=True)
    
    return recv_x, recv_topk_idx, recv_topk_weights, num_recv_tokens_per_expert_list, handle, event

def combine_forward(x: torch.Tensor, handle: Tuple, previous_event: Optional[EventOverlap] = None):
    """
    执行 MoE Combine (前向)
    返回: combined_x, event
    """
    global _buffer
    
    combined_x, _, event = _buffer.combine(x, handle, async_finish=True, previous_event=previous_event,
                                           allocate_on_comm_stream=previous_event is not None)
    return combined_x, event

# 使用示例流程
# 假设已初始化 dist 进程组 group 和输入数据 x, topk_idx, topk_weights
# buffer = get_buffer(group, hidden_bytes)
# dispatched_data, ..., handle, event = dispatch_forward(x, topk_idx, topk_weights, num_experts)
# ... (专家计算) ...
# final_output, _ = combine_forward(dispatched_data, handle)

网络配置建议

• 流量隔离：建议通过设置环境变量 NVSHMEM_IB_SL 指定虚拟车道（Virtual Lanes），将普通内核、低延迟内核及其他业务的流量隔离开。
• 自适应路由：在高负载网络环境中建议开启自适应路由以避免拥塞；低负载环境下建议使用静态路由以降低延迟。