Normal vs Low-Latency: 双通信哲学在 DeepEP V1 / V2 / DeepGEMM 中的三系统比较分析

deep_ep/buffers/legacy.py
  ├── Buffer (Normal 模式): 使用 internode.cu + intranode.cu
  └── Buffer (Low-Latency 模式): 使用 internode_ll.cu

# deep_ep/buffers/legacy.py:37-93
class Buffer:
    def __init__(self,
                 group,
                 num_nvl_bytes: int = 0,
                 num_rdma_bytes: int = 0,
                 low_latency_mode: bool = False,  # ← 模式选择参数
                 num_qps_per_rank: int = 24,
                 allow_nvlink_for_low_latency_mode: bool = True,
                 ...):
        self.low_latency_mode = low_latency_mode
        self.runtime = _C.Buffer(self.rank, self.group_size,
                                 num_nvl_bytes, num_rdma_bytes,
                                 low_latency_mode, ...)  # ← 传入 C++ 运行时

// csrc/kernels/legacy/internode.cu:487-516
enum class WarpRole {
    kRDMASender,              // RDMA 发送
    kRDMASenderCoordinator,   // RDMA 发送协调
    kRDMAAndNVLForwarder,     // RDMA + NVLink 转发 ← 关键角色
    kForwarderCoordinator,    // 转发协调
    kNVLReceivers             // NVLink 接收
};

const auto role_meta = [=]() -> std::pair<WarpRole, int> {
    if (is_forwarder) {
        if (warp_id < LEGACY_NUM_MAX_NVL_PEERS) {
            return {WarpRole::kRDMAAndNVLForwarder, ...};  // 转发 warp
        } else {
            return {WarpRole::kForwarderCoordinator, ...};
        }
    } else if (warp_id < kNumDispatchRDMASenderWarps) {
        return {WarpRole::kRDMASender, -1};  // RDMA 发送 warp
    }
    // ...
};

// csrc/kernels/legacy/internode.cu:527-529
// RDMA symmetric layout: Chunk 是通信粒度
auto rdma_channel_data = SymBuffer<uint8_t>(
    rdma_buffer_ptr,
    num_max_rdma_chunked_recv_tokens * num_bytes_per_token,  // ← Chunk 大小
    kNumRDMARanks, channel_id, num_channels);

// csrc/kernels/legacy/internode.cu:809-817
// 按 Chunk 粒度发送，而非单 Token
auto num_tokens_to_issue = min(num_tokens_processed, num_max_rdma_chunked_send_tokens);
auto dst_slot_idx = synced_last_issued_tail % num_max_rdma_chunked_recv_tokens;
EP_DEVICE_ASSERT(dst_slot_idx + num_tokens_to_issue <= num_max_rdma_chunked_recv_tokens);

// csrc/kernels/legacy/internode.cu:583-585
__shared__ volatile int forward_channel_head[LEGACY_NUM_MAX_NVL_PEERS][kNumRDMARanks];
__shared__ volatile bool forward_channel_retired[LEGACY_NUM_MAX_NVL_PEERS];
auto sync_forwarder_smem = []() {
    asm volatile("barrier.sync 1, %0;" ::"r"((LEGACY_NUM_MAX_NVL_PEERS + 1) * 32));
};

// csrc/kernels/legacy/internode.cu:87-90
template <bool kLowLatencyMode>
__forceinline__ __device__ int translate_dst_rdma_rank(const int dst_rdma_rank, const int nvl_rank) {
    // Low-Latency 模式: 每个 GPU 独立 RDMA rank
    // Normal 模式: 同一 node 的 GPU 共享 RDMA rank
    return kLowLatencyMode ? (dst_rdma_rank * LEGACY_NUM_MAX_NVL_PEERS + nvl_rank) : dst_rdma_rank;
}

template <bool kLowLatencyMode>
__forceinline__ __device__ void nvshmem_sync_with_same_gpu_idx(const nvshmem_team_t& rdma_team) {
    // Low-Latency: 仅同步 RDMA team
    // Normal: 全局同步
    kLowLatencyMode ? void(nvshmem_sync(rdma_team)) : nvshmem_sync_all();
}

// csrc/legacy/config.hpp:24-51
struct Config {
    int num_sms;
    int num_max_nvl_chunked_send_tokens;    // NVLink Chunk 发送
    int num_max_nvl_chunked_recv_tokens;    // NVLink Chunk 接收
    int num_max_rdma_chunked_send_tokens;   // RDMA Chunk 发送
    int num_max_rdma_chunked_recv_tokens;   // RDMA Chunk 接收
    // ...
    // 约束: send < recv / 2 (确保发送方总有空间)
    EP_HOST_ASSERT(num_max_rdma_chunked_send_tokens <= num_max_rdma_chunked_recv_tokens / 2);
};

// csrc/kernels/legacy/internode_ll.cu:128-155
template <bool kUseFP8, bool kUseUE8M0, int kHidden>
__global__ __launch_bounds__(1024, 1) void dispatch(...) {
    const auto responsible_expert_idx = sm_id * num_warp_groups + warp_group_id;

    // 发送阶段: 所有 warp 参与 FP8 cast + IBGDA send
    if (warp_id < num_warps - 1) {
        for (int token_idx = sm_id; token_idx < num_tokens; token_idx += num_sms) {
            // FP8 cast + 直接发送
            // ...
            if (dst_expert_idx >= 0) {
                // 直接通过 RDMA 发送，无 Forwarding
                nvshmemi_ibgda_put_nbi_warp(dst_ptr, src_ptr, num_bytes_per_msg, dst_rank, ...);
            }
        }
    }
}

// csrc/kernels/legacy/internode_ll.cu:179-182
// Message package: index at source (int), 3 reserved int fields, hidden data, FP8 scales
const size_t num_bytes_per_msg = sizeof(int4) + (kUseFP8 ? (kHidden + num_scales * sizeof(float)) : (kHidden * sizeof(nv_bfloat16)));
const size_t num_int4_per_msg = num_bytes_per_msg / sizeof(int4);

// csrc/kernels/legacy/internode_ll.cu:362-370
// 接收端按 Expert 组织，而非按 Rank
const auto rdma_recv_x_uint8 = static_cast<uint8_t*>(rdma_recv_x) +
    local_expert_idx * num_ranks * num_max_dispatch_tokens_per_rank * num_bytes_per_msg +
    src_rank * num_max_dispatch_tokens_per_rank * num_bytes_per_msg;
const auto recv_x_int4 = static_cast<int4*>(packed_recv_x) +
    local_expert_idx * num_ranks * num_max_dispatch_tokens_per_rank * hidden_int4;

// csrc/kernels/legacy/internode_ll.cu:383-408
// 等待 Token 到达，带超时检测
if (sub_warp_id == 1 and lane_id == 0) {
    auto start_time = clock64();
    uint64_t wait_recv_cost = 0;
    if (not is_rank_masked(mask_buffer_ptr, src_rank)) {
        while ((num_recv_tokens = ld_acquire_sys_global(rdma_recv_count + ...)) == 0
               && (wait_recv_cost = clock64() - start_time) <= LEGACY_NUM_TIMEOUT_CYCLES)
            ;
    }
    // 超时则 mask 该 rank
    if (wait_recv_cost > LEGACY_NUM_TIMEOUT_CYCLES) {
        atomicExch(mask_buffer_ptr + src_rank, 1);
    }
}

deep_ep/buffers/elastic.py
  └── ElasticBuffer
        ├── dispatch_impl<kIsScaleupNVLink=false>  → hybrid_dispatch.cuh (Normal: RDMA + NVLink)
        ├── dispatch_impl<kIsScaleupNVLink=true>   → dispatch.cuh (Low-Latency: 纯 NVLink)
        └── (num_scaleout_ranks == 1 时自动选择 direct)

// csrc/kernels/elastic/dispatch.hpp:53-78
static std::string generate_impl(const Args& args) {
    std::string header_name, func_name;
    if (args.num_scaleout_ranks == 1) {
        // 单节点模式 (类似 Low-Latency 场景)
        header_name = "dispatch";
        func_name = fmt::format("dispatch_impl<{}, {}, ...>",
            args.is_scaleup_nvlink,  // ← true = NVLink only
            ...);
    } else {
        // 跨节点模式 (类似 Normal 场景)
        header_name = "hybrid_dispatch";
        func_name = fmt::format("hybrid_dispatch_impl<{}, {}, ...>",
            args.num_scaleout_warps, args.num_forward_warps,  // ← 有 forwarding warps
            ...);
    }
    return fmt::format(R"(
#include <deep_ep/impls/{}.cuh>
...
)", header_name, func_name);
}

// deep_ep/include/deep_ep/impls/dispatch.cuh:17-30
template <bool kIsScaleupNVLink,      // ← 核心模式开关
          bool kDoCPUSync,
          bool kReuseSlotIndices,
          int kNumSMs,
          int kNumNotifyWarps, int kNumDispatchWarps,
          int kNumRanks,
          int kNumHiddenBytes, int kNumSFPacks,
          int kNumMaxTokensPerRank,
          int kNumExperts, int kNumTopk, int kExpertAlignment,
          int kNumQPs, int64_t kNumTimeoutCycles,
          ...>
__global__ void dispatch_impl(...)

// deep_ep/include/deep_ep/impls/hybrid_dispatch.cuh:14-32
template <..., int kNumScaleoutWarps, int kNumForwardWarps,  // ← Forward warps
          int kNumScaleoutRanks, int kNumScaleupRanks, ...>
__global__ void hybrid_dispatch_impl(...) {
    // Warp 角色:
    // 1. Notify warps: 元数据交换
    // 2. Scaleout warps: RDMA 发送
    // 3. Forward warps: NVLink 转发 ← 对应 Normal 的 Forwarding
}

// deep_ep/include/deep_ep/impls/hybrid_dispatch.cuh:88-97
auto scaleup_buffer = layout::BufferLayout<false>(
    token_layout, kNumScaleupRanks, kNumScaleoutRanks * kNumMaxTokensPerRank, buffer);
auto scaleout_send_buffer = layout::BufferLayout<false>(
    token_layout, 1, kNumMaxTokensPerRank, scaleup_buffer.get_buffer_end_ptr());
auto scaleout_recv_buffer = layout::BufferLayout<false>(
    token_layout, kNumScaleoutRanks, kNumChannels * kNumMaxTokensPerChannel, ...);

// deep_ep/include/deep_ep/impls/hybrid_dispatch.cuh:24-27
int kNumChannelsPerSM = kNumScaleoutWarps,
int kNumChannels = kNumScaleoutWarps * kNumSMs,
int kNumMaxTokensPerChannel = math::constexpr_ceil_div(kNumMaxTokensPerRank, kNumChannels),
int kScaleoutUpdateInterval = 6,           // ← Chunk 更新间隔
int kNumSlotsPerForwardChunk = kScaleoutUpdateInterval,  // ← Chunk 大小

// deep_ep/include/deep_ep/impls/dispatch.cuh:362-393
// Issue TMA NVLink stores (直接写入，无转发)
const auto dst_ptr = stored_dst_slot_idx >= 0 ?
    gin.get_sym_ptr<team_t>(recv_buffer.get_token_buffer(stored_dst_slot_idx).get_base_ptr(), stored_dst_rank_idx) :
    nullptr;
if (dst_ptr != nullptr)
    ptx::tma_store_1d(dst_ptr, tma_buffer.get_base_ptr(), tma_buffer.get_num_bytes<false>());

// Issue RDMA put (仅对非 NVLink 可达的 rank)
if constexpr (not kIsScaleupNVLink) {
    if (stored_dst_slot_idx >= 0 and dst_ptr == nullptr) {
        gin.put<team_t>(recv_buffer.get_token_buffer(stored_dst_slot_idx).get_base_ptr(),
                        send_buffer_ptr, tma_buffer.get_num_bytes<false>(), stored_dst_rank_idx);
    }
}

// deep_ep/include/deep_ep/impls/dispatch.cuh:278-292
// 每个 Token 独立处理，无 Chunk 聚合
const auto token_start = dispatch_warp_idx * kNumSMs + sm_idx;
const auto token_stride = kNumDispatchWarps * kNumSMs;
for (int token_idx = token_start; token_idx < num_tokens; token_idx += token_stride) {
    // Issue data TMA (单 Token)
    if (ptx::elect_one_sync()) {
        ptx::tma_load_1d(tma_buffer.get_hidden_ptr(),
                         math::advance_ptr(x, token_i64_idx * kNumHiddenBytes),
                         mbarrier_ptr, kNumHiddenBytes);
    }
    // ...
}

# deep_ep/buffers/elastic.py:728-834
@weak_lru(maxsize=None)
def get_theoretical_num_sms(self, num_experts, num_topk,
                            num_scaleout_topk: int = 0,
                            rdma_gbs: float = 0, nvlink_gbs: float = 0,
                            sm_read_gbs: float = 200, sm_write_gbs: float = 50) -> int:
    # 根据带宽模型自动决策 SM 数量
    # 考虑 rdma_traffic, nvlink_traffic, sm_read, sm_write
    # 选择瓶颈带宽，计算所需 SM 数
    num_sms = max(
        bounded_gbs / bounded_traffic * sm_read / sm_read_gbs,
        bounded_gbs / bounded_traffic * sm_write / sm_write_gbs,
    )
    num_sms = align(max(4, math.ceil(num_sms * 1.25)), 2)
    num_sms = min(num_sms, num_device_sms)
    return num_sms

deep_gemm/mega/__init__.py
  └── fp8_fp4_mega_moe / bf16_mega_moe
        └── csrc/apis/mega.hpp
              └── sm100_fp8_fp4_mega_moe (单一实现)

// csrc/apis/mega.hpp:157-286
static void fp8_fp4_mega_moe(
    const torch::Tensor& y,
    const std::tuple<torch::Tensor, torch::Tensor>& l1_weights_,
    const std::tuple<torch::Tensor, torch::Tensor>& l2_weights_,
    const torch::Tensor& sym_buffer,
    const std::vector<int64_t>& sym_buffer_ptrs, const int& rank_idx,
    const int& num_max_tokens_per_rank,
    const int& num_experts, const int& num_topk,
    const std::tuple<int, int, int>& recipe,
    const std::string& activation,
    const std::optional<float>& activation_clamp_opt,
    const bool& fast_math) {
    // 无 mode / kernel_type / is_low_latency 参数
    if (arch_major == 10) {
        sm100_fp8_fp4_mega_moe(y, ...);  // 单一实现
    }
}

// csrc/jit_kernels/heuristics/mega_moe.hpp:183-248
static MegaMoEConfig get_mega_moe_config(
    const int& num_ranks, const int& num_experts, const int& num_experts_per_rank,
    const int& num_max_tokens_per_rank, const int& num_tokens, const int& num_topk,
    const int& hidden, const int& intermediate_hidden, ...) {
    // 直接计算配置，无 if/else 模式分支
    const int block_n = 128;
    const int num_stages = (smem_capacity - smem_fixed) / smem_size_per_stage;
    DG_HOST_ASSERT(num_stages >= 2);
    // ...
}

// sm100_fp8_fp4_mega_moe.cuh:545-550
// TMA load token from remote rank into shared memory
if (cute::elect_one_sync()) {
    ptx::tma_load_1d(
        pull_buffer.get_base_ptr(),
        sym_buffer.map(input_token_buffer.get_data_buffer(src_token_idx).get_base_ptr(),
                       current_rank_in_expert_idx),
        pull_mbarrier, kHidden);
}

DeepEP Normal:
  Source GPU → IB Sending → RDMA Network → IB-to-NVLink Forwarding → NVLink Receiving → Target GPU
                                          ↑
                                   GPU 作为通信中继 (因 NIC-GPU 拓扑不对称)

DeepGEMM:
  Target GPU → NVLink Pull → Source GPU
  (直接读取远端 Symmetric Buffer，无需中继)

// sm100_fp8_fp4_mega_moe.cuh:671-675
// Wait the entire token arrival for linear 1
if (block_phase == sched::BlockPhase::Linear1) {
    const auto ptr = workspace.get_l1_arrival_count_ptr(pool_block_idx);
    const auto expected = scheduler.template get_valid_m<false>();
    while (ptx::ld_acq(ptr) != expected);  // ← Arrival Count 驱动
}

// csrc/jit_kernels/heuristics/mega_moe.hpp:76-113
static std::tuple<int, int, int, int, int> get_block_config_for_mega_moe(...) {
    float num_expected_tokens_per_expert = static_cast<float>(num_tokens) * num_ranks * num_topk / num_experts;
    if (num_expected_tokens_per_expert <= 8.5) {
        // RL long-tail rollout
        return {2, 16, 8, 256, 2};   // 最小 block_m
    } else if (num_expected_tokens_per_expert <= 16.5) {
        // Decoding, small batch
        return {2, 32, 16, 128, 2};
    } else if (num_expected_tokens_per_expert <= 32.5) {
        return {2, 64, 32, 128, 1};
    } else if (num_expected_tokens_per_expert <= 64.5) {
        return {2, 96, 16, 128, 2};
    } else if (num_expected_tokens_per_expert <= 96.5) {
        return {2, 128, 32, 128, 2};
    } else {
        // Prefill, or large EP decoding
        return {2, 192, 32, 128, 2};  // 最大 block_m
    }
}

DeepEP V1:
  Normal:     [Dispatch] → [Chunk] → [NVLink/RDMA + Forwarding] → [Combine]
  Low-Latency:[Dispatch] → [Direct RDMA]                          → [Combine]

DeepEP V2:
  scaleout>1: [Notify] → [Scaleout Send] → [Forward] → [Scaleup Receive] → [Copy Epilogue]
  scaleout=1: [Notify] → [Dispatch (NVLink or RDMA)]                           → [Copy Epilogue]

DeepGEMM:
  Single:     [Dispatch NVLink Pull] → [L1 GEMM + SwiGLU] → [L2 GEMM] → [Combine NVLink Write-back]
              └────────────────────── Persistent ────────────────────────────┘

DeepEP V1 (物理分离)
    │
    │ 问题: 两套代码维护成本高，模式切换不灵活
    ▼
DeepEP V2 (逻辑统一)
    │
    │ 方法: JIT 编译 + 模板参数 (kIsScaleupNVLink, num_scaleout_ranks)
    │ 效果: 同一份代码，编译期分化
    ▼
DeepGEMM (硬件消除)
    │
    │ 方法: NVLink Symmetric Memory + Pull-based 通信
    │ 效果: 硬件能力抹平模式分化，单内核覆盖全场景
    ▼
未来方向: ?
    - CFT (Compute Fabric Transport) 进一步降低延迟
    - 跨节点扩展 (DeepGEMM 当前仅支持单节点)

Agent 6 分析的拓扑:
  Intra-node:  NVLink 全连接 (对称)
  Inter-node:  RDMA (IB) (NIC-GPU 不对称)

Normal 模式必须处理:
  GPU → NVLink domain → NIC → RDMA → NVLink domain → GPU
         ↑
    Forwarding 解决 NIC-GPU 不对称

Low-Latency 模式:
  GPU → Direct RDMA → GPU
  (旁路 NVLink 域)

// Token 流 → Chunk → 网络传输
// Token 是调度粒度，Chunk 是通信粒度
Token Stream → Chunk → Network Transfer

// hybrid_dispatch.cuh 中的 Chunk
int kScaleoutUpdateInterval = 6,           // Chunk 更新间隔
int kNumSlotsPerForwardChunk = kScaleoutUpdateInterval,  // Chunk 大小

Normal:
  Token Buffer → Dispatch Buffer → Chunk Buffer → NVLink / RDMA Pipeline → Receive Buffer → Expert Buffer

Low-Latency:
  Token Buffer → Direct RDMA → Receive Buffer → Expert Buffer

# 统一 Buffer，支持多种后端
class ElasticBuffer:
    """
    支持:
    - high-throughput expert-parallel all-to-all (NVLink + RDMA)
    - Engram (remote KV cache fetch, RDMA)
    - pipeline-parallel send/recv (PP, NVLink)
    - all-gather reduce-scatter (AGRS, NVLink)
    """

# deep_gemm/mega/__init__.py:18-66
class SymmBuffer:
    """NVLink Symmetric Memory Buffer"""
    # 单一 Buffer，包含所有输入/输出视图
    # x, x_sf, topk_idx, topk_weights
    # l1_acts, l1_acts_sf, l2_acts, l2_acts_sf

Normal 模式存在的前提:
  1. NIC-GPU 拓扑不对称 (部分 GPU 直连 NIC)
  2. 需要 Forwarding 中继
  3. 需要 Chunk 聚合提高带宽利用率

Low-Latency 模式存在的前提:
  1. 每个 GPU 可直接访问 RDMA (或旁路 Forwarding)
  2. 单 Token 延迟敏感，不能等待 Chunk 聚合
  3. 浅流水线减少 Stage 延迟