fix benchmark

Signed-off-by: Woosuk Kwon <woosuk@inferact.ai>

csrc/standalone/custom_ar/bench_moe_allgather.py

@@ -0,0 +1,200 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Benchmark: Lamport all-gather vs NCCL."""
+
+import ctypes
+import os
+import sys
+
+import torch
+import torch.distributed as dist
+
+_cudart = ctypes.CDLL("libcudart.so")
+IPC = 64
+
+
+def _cc(r):
+    if r:
+        raise RuntimeError(f"err {r}")
+
+
+def ipc_buf(sz, rank, ws):
+    p = ctypes.c_void_p()
+    _cc(_cudart.cudaMalloc(ctypes.byref(p), sz))
+    _cc(_cudart.cudaMemset(p, 0, sz))
+    _cc(_cudart.cudaDeviceSynchronize())
+    h = (ctypes.c_byte * IPC)()
+    _cc(_cudart.cudaIpcGetMemHandle(ctypes.byref(h), p))
+    ah = [None] * ws
+    dist.all_gather_object(ah, bytes(h))
+    ptrs = []
+    for i in range(ws):
+        if i == rank:
+            ptrs.append(p.value)
+        else:
+            hh = (ctypes.c_byte * IPC)(*ah[i])
+            pp = ctypes.c_void_p()
+            _cc(_cudart.cudaIpcOpenMemHandle(ctypes.byref(pp), hh, ctypes.c_uint(1)))
+            ptrs.append(pp.value)
+    return ptrs
+
+
+def gpu_timer(fn, warmup=20, repeats=200):
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    s = torch.cuda.Event(enable_timing=True)
+    e = torch.cuda.Event(enable_timing=True)
+    s.record()
+    for _ in range(repeats):
+        fn()
+    e.record()
+    torch.cuda.synchronize()
+    return s.elapsed_time(e) / repeats * 1000
+
+
+def gpu_timer_graph(fn, warmup=20, repeats=200):
+    """Time with CUDA graph to exclude CPU overhead."""
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    g = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(g):
+        fn()
+    for _ in range(5):
+        g.replay()
+    torch.cuda.synchronize()
+    s = torch.cuda.Event(enable_timing=True)
+    e = torch.cuda.Event(enable_timing=True)
+    s.record()
+    for _ in range(repeats):
+        g.replay()
+    e.record()
+    torch.cuda.synchronize()
+    return s.elapsed_time(e) / repeats * 1000
+
+
+def main():
+    dist.init_process_group("nccl")
+    rank = dist.get_rank()
+    ws = dist.get_world_size()
+    torch.cuda.set_device(rank)
+    dev = f"cuda:{rank}"
+
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    if rank == 0:
+        from moe_allgather import _load_lib
+
+        lib = _load_lib()
+    dist.barrier()
+    if rank != 0:
+        from moe_allgather import _load_lib
+
+        lib = _load_lib()
+    dist.barrier()
+    from moe_allgather import MoeAllGather
+
+    max_size = 8 * 1024 * 1024
+    bp = ipc_buf(max_size, rank, ws)
+    dist.barrier()
+
+    class FakeCA:
+        pass
+
+    ca = FakeCA()
+    ca.rank = rank
+    ca.world_size = ws
+    ca.device = torch.device(dev)
+    ca.buffer_ptrs = bp
+    ca.max_size = max_size
+    ag = MoeAllGather(ca)
+    dist.barrier()
+
+    configs = [
+        ("1tok", 1),
+        ("2tok", 2),
+        ("4tok", 4),
+        ("8tok", 8),
+        ("16tok", 16),
+        ("32tok", 32),
+        ("64tok", 64),
+        ("128tok", 128),
+        ("256tok", 256),
+    ]
+    topk = 8
+    hd = 3584
+    sd = 448
+
+    if rank == 0:
+        print(f"world_size={ws}, max_per_rank={ag.max_per_rank} bytes")
+        print(f"{'config':<12} {'lamport_graph':>10} {'nccl_graph':>10} {'speedup':>8}")
+        print("-" * 65)
+
+    for name, N in configs:
+        # Check if data fits in buffer.
+        cursor = 0
+        per_tok = topk * 4 + topk * 4 + hd + sd
+        cursor = N * per_tok
+        cursor = (cursor + 15) & ~15
+        if cursor > ag.max_per_rank:
+            if rank == 0:
+                print(f"{name:<12} {'skip (too large)':>40}")
+            continue
+
+        ids = torch.randint(0, 256, (N, topk), dtype=torch.int32, device=dev)
+        wt = torch.randn(N, topk, dtype=torch.float32, device=dev).abs()
+        hs = torch.randint(0, 255, (N, hd), dtype=torch.uint8, device=dev)
+        sc = torch.randint(0, 255, (N, sd), dtype=torch.uint8, device=dev)
+        inputs = [ids, wt, hs, sc]
+
+        # Custom Lamport kernel.
+        c_outs = [
+            torch.empty(N * ws, *t.shape[1:], dtype=t.dtype, device=dev) for t in inputs
+        ]
+
+        def run_lamport():
+            lib.moe_all_gather(
+                ag._buf_ptrs_ptr,
+                ag._counters_ptr,
+                rank,
+                ws,
+                ag.seg_capacity,
+                ag.rank_stride,
+                inputs,
+                c_outs,
+            )
+
+        # Lamport with CUDA graph.
+        try:
+            lam_g_us = gpu_timer_graph(run_lamport)
+        except Exception as ex:
+            lam_g_us = float("nan")
+            if rank == 0:
+                print(f"  [graph capture failed: {ex}]")
+
+        # NCCL 1×AG (concat into one tensor).
+        cat_inp = torch.cat(
+            [t.reshape(N, -1).contiguous().view(torch.uint8) for t in inputs],
+            dim=1,
+        ).contiguous()
+        cat_out = torch.empty(N * ws, cat_inp.shape[1], dtype=torch.uint8, device=dev)
+
+        def run_nccl():
+            dist.all_gather_into_tensor(cat_out, cat_inp)
+
+        # NCCL with CUDA graph.
+        try:
+            nccl_g_us = gpu_timer_graph(run_nccl)
+        except Exception:
+            nccl_g_us = float("nan")
+
+        if rank == 0:
+            speedup = nccl_g_us / lam_g_us if lam_g_us > 0 else float("nan")
+            print(f"{name:<12} {lam_g_us:>9.1f}µ {nccl_g_us:>9.1f}µ {speedup:>7.2f}x")
+
+    dist.barrier()
+    dist.destroy_process_group()
+
+
+if __name__ == "__main__":
+    main()

csrc/standalone/custom_ar/moe_allgather.cu

@@ -0,0 +1,301 @@
+// Lamport-based MoE all-gather kernel for EP dispatch.
+//
+// Replaces the flag-barrier approach with a Lamport sentinel protocol
+// (inspired by FlashInfer's trtllm_allreduce_fusion).
+//
+// Key advantages over the flag-barrier approach:
+//   - No explicit barriers (sentinels provide per-element synchronization).
+//   - Push model: NVLink writes (fire-and-forget) instead of NVLink reads.
+//   - Triple buffering: no end barrier needed.
+//
+// Gathers the MoE dispatch tensors from all EP ranks:
+//   - topk_ids      [N, topk]     int32
+//   - topk_weights  [N, topk]     float32 / bfloat16
+//   - hidden_states [N, D_h]      uint8 (NVFP4) / bfloat16
+//   - quant_scales  [N, D_s]      (optional)
+//
+// Double-buffer layout in each rank's IPC buffer:
+//   [Segment 0][Segment 1]
+//   Each segment: [Rank 0 slot][Rank 1 slot]...[Rank N-1 slot]
+//   Each rank slot: packed tensors at 16-byte aligned offsets.
+//
+// Sentinel: 0x80000000 (negative-zero in float32).  The writer replaces
+// any data word matching the sentinel with 0 before pushing.  The reader
+// spin-loads (volatile) until no sentinel words remain in the vector.
+
+#include <cuda.h>
+#include <cuda_runtime.h>
+#include <torch/extension.h>
+
+#include <c10/cuda/CUDAGuard.h>
+#include <c10/cuda/CUDAStream.h>
+
+#define DINLINE __device__ __forceinline__
+
+constexpr uint32_t SENTINEL = 0x80000000u;
+constexpr int kMaxBlocks = 36;
+
+// ---------------------------------------------------------------------------
+// Volatile 128-bit load/store and sentinel helpers
+// ---------------------------------------------------------------------------
+
+static DINLINE int4 ld128v(const void* addr) {
+  int4 v;
+  asm volatile("ld.volatile.global.v4.b32 {%0,%1,%2,%3}, [%4];"
+               : "=r"(v.x), "=r"(v.y), "=r"(v.z), "=r"(v.w)
+               : "l"(addr));
+  return v;
+}
+
+static DINLINE bool has_sentinel(int4 v) {
+  return reinterpret_cast<uint32_t&>(v.x) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.y) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.z) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.w) == SENTINEL;
+}
+
+static DINLINE int4 remove_sentinel(int4 v) {
+  if (reinterpret_cast<uint32_t&>(v.x) == SENTINEL) v.x = 0;
+  if (reinterpret_cast<uint32_t&>(v.y) == SENTINEL) v.y = 0;
+  if (reinterpret_cast<uint32_t&>(v.z) == SENTINEL) v.z = 0;
+  if (reinterpret_cast<uint32_t&>(v.w) == SENTINEL) v.w = 0;
+  return v;
+}
+
+// ---------------------------------------------------------------------------
+// Lamport all-gather kernel
+// ---------------------------------------------------------------------------
+//
+// Phase 1 — PUSH: each rank writes its packed data to ALL peers' current
+//           segment via regular stores (NVLink push, fire-and-forget).
+// Phase 2 — CLEAR: each rank writes sentinels to the OLDEST segment of
+//           its own buffer, preparing it for reuse.
+// Phase 3 — POLL + SCATTER: each rank volatile-loads from its own current
+//           segment, spinning until sentinels disappear, then scatters
+//           directly to per-tensor output arrays.
+// Phase 4 — ADVANCE: one thread advances the triple-buffer ring counter.
+
+template <int ngpus, int nbufs>
+__global__ void __launch_bounds__(512, 1) moe_allgather_lamport_kernel(
+    int64_t* buf_ptrs,  // [ngpus] IPC buffer base addresses (device)
+    int* counters,      // [0] = unused, [1] = ring (0/1/2), [2] = prev total_sz
+    int rank,
+    int seg_capacity,  // bytes per segment
+    int rank_stride,   // bytes per rank-slot within a segment
+    int total_sz,      // int4 units of actual packed data per rank
+    // inputs (up to 4)
+    const void* inp0, const void* inp1, const void* inp2, const void* inp3,
+    int off0, int sz0, int off1, int sz1, int off2, int sz2, int off3, int sz3,
+    // outputs (up to 4)
+    void* out0, void* out1, void* out2, void* out3) {
+  using V = int4;
+  const int tid = blockIdx.x * blockDim.x + threadIdx.x;
+  const int stride = gridDim.x * blockDim.x;
+
+  // Read segment index and previous clear size.
+  const int seg = counters[1];            // 0 or 1
+  const int prev_total_sz = counters[2];  // set by previous invocation
+  const int cur_seg = seg;
+  const int old_seg = 1 - seg;
+
+  char* bufs[ngpus];
+#pragma unroll
+  for (int r = 0; r < ngpus; r++)
+    bufs[r] = reinterpret_cast<char*>(buf_ptrs[r]) + cur_seg * seg_capacity;
+
+  // Sentinel vector for clearing.
+  V sent;
+  sent.x = sent.y = sent.z = sent.w = static_cast<int>(SENTINEL);
+
+  // ---- Phase 1: PUSH local data to ALL peers ----
+  // Write to peer_r's buffer at [rank * rank_stride + off_i].
+
+#define PUSH(idx, inp_ptr, off_val, sz_val)                                 \
+  if constexpr (nbufs > (idx)) {                                            \
+    const V* src = reinterpret_cast<const V*>(inp_ptr);                     \
+    for (int i = tid; i < (sz_val); i += stride) {                          \
+      V val = remove_sentinel(src[i]);                                      \
+      _Pragma("unroll") for (int r = 0; r < ngpus; r++) {                   \
+        reinterpret_cast<V*>(bufs[r] + rank * rank_stride + (off_val))[i] = \
+            val;                                                            \
+      }                                                                     \
+    }                                                                       \
+  }
+
+  PUSH(0, inp0, off0, sz0)
+  PUSH(1, inp1, off1, sz1)
+  PUSH(2, inp2, off2, sz2)
+  PUSH(3, inp3, off3, sz3)
+#undef PUSH
+
+  // ---- Phase 2: CLEAR only the previously-written data in oldest segment ----
+  // Only clear what the previous invocation actually wrote (per rank-slot).
+  if (prev_total_sz > 0) {
+    char* clr_base =
+        reinterpret_cast<char*>(buf_ptrs[rank]) + old_seg * seg_capacity;
+#pragma unroll
+    for (int r = 0; r < ngpus; r++) {
+      V* clr = reinterpret_cast<V*>(clr_base + r * rank_stride);
+      for (int i = tid; i < prev_total_sz; i += stride) clr[i] = sent;
+    }
+  }
+
+  // ---- Phase 3: POLL + SCATTER ----
+  // Volatile-load from own buffer; spin until sentinel gone; scatter to output.
+  char* my = bufs[rank];
+
+#define POLL(idx, out_ptr, off_val, sz_val)                                \
+  if constexpr (nbufs > (idx)) {                                           \
+    for (int i = tid; i < (sz_val); i += stride) {                         \
+      _Pragma("unroll") for (int s = 0; s < ngpus; s++) {                  \
+        V val;                                                             \
+        do {                                                               \
+          val = ld128v(                                                    \
+              reinterpret_cast<V*>(my + s * rank_stride + (off_val)) + i); \
+        } while (has_sentinel(val));                                       \
+        reinterpret_cast<V*>(out_ptr)[s * (sz_val) + i] = val;             \
+      }                                                                    \
+    }                                                                      \
+  }
+
+  POLL(0, out0, off0, sz0)
+  POLL(1, out1, off1, sz1)
+  POLL(2, out2, off2, sz2)
+  POLL(3, out3, off3, sz3)
+#undef POLL
+
+  // ---- Phase 4: ADVANCE ring counter + store clear size for next call ----
+  // Stream serialization ensures the next kernel sees these updates.
+  if (blockIdx.x == 0 && threadIdx.x == 0) {
+    counters[1] = 1 - seg;
+    counters[2] = total_sz;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Sentinel initialization kernel
+// ---------------------------------------------------------------------------
+
+__global__ void lamport_init_kernel(uint32_t* buf, int n) {
+  int tid = blockIdx.x * blockDim.x + threadIdx.x;
+  int stride = gridDim.x * blockDim.x;
+  for (int i = tid; i < n; i += stride) buf[i] = SENTINEL;
+}
+
+// ---------------------------------------------------------------------------
+// Host launcher
+// ---------------------------------------------------------------------------
+
+struct TensorDesc {
+  void* inp;
+  int off;
+  int sz;
+  int64_t nbytes;
+};
+
+static TensorDesc make_desc(torch::Tensor& inp, int64_t& cursor) {
+  TORCH_CHECK(inp.is_contiguous(), "input must be contiguous");
+  int64_t nbytes = inp.numel() * inp.element_size();
+  TORCH_CHECK(nbytes % 16 == 0, "tensor byte size must be multiple of 16, got ",
+              nbytes);
+  cursor = (cursor + 15) & ~15;
+  TensorDesc d;
+  d.inp = inp.data_ptr();
+  d.off = static_cast<int>(cursor);
+  d.sz = static_cast<int>(nbytes / 16);
+  d.nbytes = nbytes;
+  cursor += nbytes;
+  return d;
+}
+
+void lamport_init(int64_t buf_ptr, int64_t nbytes) {
+  auto stream = c10::cuda::getCurrentCUDAStream().stream();
+  int n = static_cast<int>(nbytes / 4);
+  lamport_init_kernel<<<256, 256, 0, stream>>>(
+      reinterpret_cast<uint32_t*>(buf_ptr), n);
+}
+
+void moe_all_gather(int64_t buf_ptrs_ptr, int64_t counters_ptr, int64_t rank,
+                    int64_t world_size, int64_t seg_capacity,
+                    int64_t rank_stride, std::vector<torch::Tensor>& inputs,
+                    std::vector<torch::Tensor>& outputs) {
+  auto stream = c10::cuda::getCurrentCUDAStream().stream();
+  int n = static_cast<int>(inputs.size());
+  TORCH_CHECK(n >= 2 && n <= 4, "2-4 input tensors required");
+  TORCH_CHECK(inputs.size() == outputs.size());
+
+  int64_t cursor = 0;
+  TensorDesc descs[4] = {};
+  for (int i = 0; i < n; i++) descs[i] = make_desc(inputs[i], cursor);
+  TORCH_CHECK(cursor % 16 == 0);
+  int total_sz = static_cast<int>(cursor / 16);
+  TORCH_CHECK(cursor <= rank_stride, "packed data (", cursor,
+              " bytes) exceeds rank_stride (", rank_stride, " bytes)");
+
+  int ws = static_cast<int>(world_size);
+  for (int i = 0; i < n; i++) {
+    TORCH_CHECK(outputs[i].is_contiguous());
+    TORCH_CHECK(outputs[i].numel() == inputs[i].numel() * ws);
+  }
+
+  int r = static_cast<int>(rank);
+  int threads = 512;
+  int blocks =
+      std::max(1, std::min(kMaxBlocks, (total_sz + threads - 1) / threads));
+
+  void *inps[4] = {}, *outs[4] = {};
+  int offs[4] = {}, szs[4] = {};
+  for (int i = 0; i < n; i++) {
+    inps[i] = descs[i].inp;
+    offs[i] = descs[i].off;
+    szs[i] = descs[i].sz;
+    outs[i] = outputs[i].data_ptr();
+  }
+
+  auto* bp = reinterpret_cast<int64_t*>(buf_ptrs_ptr);
+  auto* ct = reinterpret_cast<int*>(counters_ptr);
+  int sc = static_cast<int>(seg_capacity);
+  int rs = static_cast<int>(rank_stride);
+
+#define KL(ng, nb)                                                        \
+  moe_allgather_lamport_kernel<ng, nb><<<blocks, threads, 0, stream>>>(   \
+      bp, ct, r, sc, rs, total_sz, inps[0], inps[1], inps[2], inps[3],    \
+      offs[0], szs[0], offs[1], szs[1], offs[2], szs[2], offs[3], szs[3], \
+      outs[0], outs[1], outs[2], outs[3]);
+
+#define GPU_CASE(ng) \
+  case ng:           \
+    switch (n) {     \
+      case 2:        \
+        KL(ng, 2);   \
+        break;       \
+      case 3:        \
+        KL(ng, 3);   \
+        break;       \
+      case 4:        \
+        KL(ng, 4);   \
+        break;       \
+    }                \
+    break;
+
+  switch (ws) {
+    GPU_CASE(2)
+    GPU_CASE(4)
+    GPU_CASE(6)
+    GPU_CASE(8)
+    default:
+      TORCH_CHECK(false, "world_size must be 2, 4, 6, or 8");
+  }
+#undef GPU_CASE
+#undef KL
+}
+
+// ---------------------------------------------------------------------------
+// Python binding
+// ---------------------------------------------------------------------------
+
+PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
+  m.def("moe_all_gather", &moe_all_gather, "Lamport MoE all-gather");
+  m.def("lamport_init", &lamport_init,
+        "Initialize Lamport buffer with sentinels");
+}

csrc/standalone/custom_ar/moe_allgather.py

@@ -0,0 +1,116 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Lamport-based fused MoE all-gather for EP dispatch.
+
+JIT-compiles the CUDA kernel on first use (cached afterwards).
+Uses a Lamport sentinel protocol (push writes + per-element sync)
+with triple-buffered IPC regions — no explicit barriers.
+
+Usage:
+    ag = MoeAllGather(custom_allreduce)
+    ids_g, wt_g, hs_g, sc_g = ag.gather(topk_ids, topk_weights, hidden, scales)
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import torch
+
+_lib = None
+
+
+def _load_lib():
+    global _lib
+    if _lib is not None:
+        return _lib
+
+    from torch.utils.cpp_extension import load
+
+    src = str(Path(__file__).with_name("moe_allgather.cu"))
+    _lib = load(
+        name="moe_allgather_kernel",
+        sources=[src],
+        extra_cuda_cflags=["-O3", "--use_fast_math"],
+        verbose=os.environ.get("MOE_AG_VERBOSE", "") == "1",
+    )
+    return _lib
+
+
+class MoeAllGather:
+    """Lamport-based MoE dispatch all-gather with triple buffering."""
+
+    def __init__(self, ca_comm):
+        self.rank = ca_comm.rank
+        self.world_size = ca_comm.world_size
+        self.device = ca_comm.device
+        self.buffer_ptrs = ca_comm.buffer_ptrs
+        self.max_size = ca_comm.max_size
+
+        ws = self.world_size
+        # Double-buffer layout: 2 segments, each with ws rank-slots.
+        # Safe because kernels in the same stream are serialized, and the
+        # Use the FIRST half of the IPC buffer (second half reserved for
+        # MoeReduceScatter) to avoid overlapping writes.
+        half_size = (self.max_size // 2) & ~15
+
+        # Double-buffer layout within our half: 2 segments, each ws rank-slots.
+        # seg_capacity and rank_stride are 16-byte aligned.
+        self.seg_capacity = (half_size // 2) & ~15
+        self.rank_stride = (self.seg_capacity // ws) & ~15
+        self.max_per_rank = self.rank_stride  # max packed bytes per rank
+
+        # Buffer pointer array on device (no offset — first half).
+        self._buf_ptrs = torch.zeros(
+            8, dtype=torch.int64, device=f"cuda:{self.device.index}"
+        )
+        for i in range(ws):
+            self._buf_ptrs[i] = self.buffer_ptrs[i]
+        self._buf_ptrs_ptr = self._buf_ptrs.data_ptr()
+
+        # Counters on device: [0]=unused, [1]=seg (0/1), [2]=prev_total_sz.
+        self._counters = torch.zeros(
+            3, dtype=torch.int32, device=f"cuda:{self.device.index}"
+        )
+        self._counters_ptr = self._counters.data_ptr()
+
+        # Initialize our half with sentinel values.
+        lib = _load_lib()
+        lib.lamport_init(self.buffer_ptrs[self.rank], half_size)
+        torch.accelerator.synchronize(self.device)
+
+    def gather(
+        self,
+        topk_ids: torch.Tensor,
+        topk_weights: torch.Tensor,
+        hidden_states: torch.Tensor,
+        scales: torch.Tensor | None = None,
+    ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor | None]:
+        lib = _load_lib()
+        ws = self.world_size
+
+        inputs = [topk_ids, topk_weights, hidden_states]
+        if scales is not None:
+            inputs.append(scales)
+
+        outputs = [
+            torch.empty((t.shape[0] * ws, *t.shape[1:]), dtype=t.dtype, device=t.device)
+            for t in inputs
+        ]
+
+        lib.moe_all_gather(
+            self._buf_ptrs_ptr,
+            self._counters_ptr,
+            self.rank,
+            self.world_size,
+            self.seg_capacity,
+            self.rank_stride,
+            inputs,
+            outputs,
+        )
+
+        if scales is not None:
+            return outputs[0], outputs[1], outputs[2], outputs[3]
+        return outputs[0], outputs[1], outputs[2], None

csrc/standalone/custom_ar/moe_reduce_scatter.cu

@@ -0,0 +1,261 @@
+// Lamport-based MoE reduce-scatter kernel for EP combine.
+//
+// JIT-compilable via torch.utils.cpp_extension — no vLLM build required.
+//
+// Reduce-scatters a bf16 tensor [N_total, D] across EP ranks.  Each rank
+// contributes its partial MoE output; the kernel sums all contributions
+// and each rank receives its own slice of the result.
+//
+// Protocol (same as the all-gather variant):
+//   1. PUSH: write own data to all peers' Lamport buffers (NVLink push).
+//   2. CLEAR: write sentinels to old segment of own buffer.
+//   3. POLL + REDUCE: volatile-load all peers' data for own slice,
+//      accumulate in fp32, convert back to bf16, store to output.
+//   4. ADVANCE: toggle double-buffer index.
+//
+// Sentinel: 0x80000000 (two bf16 negative-zeros packed in uint32).
+// For bf16 reduce, replacing -0 with +0 is lossless.
+
+#include <cuda.h>
+#include <cuda_bf16.h>
+#include <cuda_runtime.h>
+#include <torch/extension.h>
+
+#include <c10/cuda/CUDAGuard.h>
+#include <c10/cuda/CUDAStream.h>
+
+#define DINLINE __device__ __forceinline__
+
+constexpr uint32_t SENTINEL = 0x80000000u;
+constexpr int kMaxBlocks = 36;
+
+// ---------------------------------------------------------------------------
+// Volatile 128-bit load and sentinel helpers
+// ---------------------------------------------------------------------------
+
+static DINLINE int4 ld128v(const void* addr) {
+  int4 v;
+  asm volatile("ld.volatile.global.v4.b32 {%0,%1,%2,%3}, [%4];"
+               : "=r"(v.x), "=r"(v.y), "=r"(v.z), "=r"(v.w)
+               : "l"(addr));
+  return v;
+}
+
+static DINLINE bool has_sentinel(int4 v) {
+  return reinterpret_cast<uint32_t&>(v.x) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.y) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.z) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.w) == SENTINEL;
+}
+
+static DINLINE int4 remove_sentinel(int4 v) {
+  if (reinterpret_cast<uint32_t&>(v.x) == SENTINEL) v.x = 0;
+  if (reinterpret_cast<uint32_t&>(v.y) == SENTINEL) v.y = 0;
+  if (reinterpret_cast<uint32_t&>(v.z) == SENTINEL) v.z = 0;
+  if (reinterpret_cast<uint32_t&>(v.w) == SENTINEL) v.w = 0;
+  return v;
+}
+
+// ---------------------------------------------------------------------------
+// bf16 ↔ fp32 helpers for int4 (8 bf16 values = 16 bytes)
+// ---------------------------------------------------------------------------
+
+// Accumulate 8 bf16 values from an int4 into 8 fp32 accumulators.
+static DINLINE void accumulate_bf16(float* acc, int4 v) {
+  const __nv_bfloat16* bp = reinterpret_cast<const __nv_bfloat16*>(&v);
+#pragma unroll
+  for (int k = 0; k < 8; k++) acc[k] += __bfloat162float(bp[k]);
+}
+
+// Convert 8 fp32 accumulators to bf16 and pack into int4.
+static DINLINE int4 fp32_to_bf16_int4(const float* acc) {
+  int4 out;
+  __nv_bfloat16* bp = reinterpret_cast<__nv_bfloat16*>(&out);
+#pragma unroll
+  for (int k = 0; k < 8; k++) bp[k] = __float2bfloat16(acc[k]);
+  return out;
+}
+
+// ---------------------------------------------------------------------------
+// Lamport reduce-scatter kernel
+// ---------------------------------------------------------------------------
+
+template <int ngpus>
+__global__ void __launch_bounds__(512, 1) moe_rs_lamport_kernel(
+    int64_t* buf_ptrs,  // [ngpus] IPC buffer base addresses (device)
+    int* counters,      // [0] = unused, [1] = seg (0/1), [2] = prev total_sz
+    int rank,
+    int seg_capacity,   // bytes per segment
+    int rank_stride,    // bytes per rank-slot within a segment
+    const void* input,  // [N_total, D] bf16 — full input
+    void* output,       // [N_per_rank, D] bf16 — this rank's reduced slice
+    int total_sz,       // int4 units of full input per rank
+    int slice_off,      // int4 offset to this rank's slice within packed data
+    int slice_sz) {     // int4 units of this rank's slice
+  using V = int4;
+  const int tid = blockIdx.x * blockDim.x + threadIdx.x;
+  const int stride = gridDim.x * blockDim.x;
+
+  // Read segment index and previous clear size.
+  const int seg = counters[1];
+  const int prev_total_sz = counters[2];
+  const int cur_seg = seg;
+  const int old_seg = 1 - seg;
+
+  char* bufs[ngpus];
+#pragma unroll
+  for (int r = 0; r < ngpus; r++)
+    bufs[r] = reinterpret_cast<char*>(buf_ptrs[r]) + cur_seg * seg_capacity;
+
+  V sent;
+  sent.x = sent.y = sent.z = sent.w = static_cast<int>(SENTINEL);
+
+  // ---- Phase 1: PUSH full input to ALL peers ----
+  {
+    const V* src = reinterpret_cast<const V*>(input);
+    for (int i = tid; i < total_sz; i += stride) {
+      V val = remove_sentinel(src[i]);
+#pragma unroll
+      for (int r = 0; r < ngpus; r++)
+        reinterpret_cast<V*>(bufs[r] + rank * rank_stride)[i] = val;
+    }
+  }
+
+  // ---- Phase 2: CLEAR old segment ----
+  if (prev_total_sz > 0) {
+    char* clr_base =
+        reinterpret_cast<char*>(buf_ptrs[rank]) + old_seg * seg_capacity;
+#pragma unroll
+    for (int r = 0; r < ngpus; r++) {
+      V* clr = reinterpret_cast<V*>(clr_base + r * rank_stride);
+      for (int i = tid; i < prev_total_sz; i += stride) clr[i] = sent;
+    }
+  }
+
+  // ---- Phase 3: POLL + REDUCE for own slice ----
+  // Read all ranks' data at [slice_off, slice_off + slice_sz) from own buffer,
+  // sum in fp32, store bf16 result.
+  {
+    char* my = bufs[rank];
+    V* dst = reinterpret_cast<V*>(output);
+
+    for (int i = tid; i < slice_sz; i += stride) {
+      float acc[8] = {0, 0, 0, 0, 0, 0, 0, 0};
+
+#pragma unroll
+      for (int s = 0; s < ngpus; s++) {
+        V val;
+        do {
+          val = ld128v(reinterpret_cast<V*>(my + s * rank_stride) + slice_off +
+                       i);
+        } while (has_sentinel(val));
+        accumulate_bf16(acc, val);
+      }
+
+      dst[i] = fp32_to_bf16_int4(acc);
+    }
+  }
+
+  // ---- Phase 4: ADVANCE ----
+  if (blockIdx.x == 0 && threadIdx.x == 0) {
+    counters[1] = 1 - seg;
+    counters[2] = total_sz;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Sentinel initialization
+// ---------------------------------------------------------------------------
+
+__global__ void lamport_init_kernel(uint32_t* buf, int n) {
+  int tid = blockIdx.x * blockDim.x + threadIdx.x;
+  int stride = gridDim.x * blockDim.x;
+  for (int i = tid; i < n; i += stride) buf[i] = SENTINEL;
+}
+
+// ---------------------------------------------------------------------------
+// Host launcher
+// ---------------------------------------------------------------------------
+
+void lamport_init(int64_t buf_ptr, int64_t nbytes) {
+  auto stream = c10::cuda::getCurrentCUDAStream().stream();
+  int n = static_cast<int>(nbytes / 4);
+  lamport_init_kernel<<<256, 256, 0, stream>>>(
+      reinterpret_cast<uint32_t*>(buf_ptr), n);
+}
+
+void moe_reduce_scatter(int64_t buf_ptrs_ptr, int64_t counters_ptr,
+                        int64_t rank, int64_t world_size, int64_t seg_capacity,
+                        int64_t rank_stride, torch::Tensor input,
+                        torch::Tensor output) {
+  auto stream = c10::cuda::getCurrentCUDAStream().stream();
+  TORCH_CHECK(input.is_contiguous(), "input must be contiguous");
+  TORCH_CHECK(output.is_contiguous(), "output must be contiguous");
+  TORCH_CHECK(input.scalar_type() == torch::kBFloat16,
+              "input must be bf16, got ", input.scalar_type());
+  TORCH_CHECK(output.scalar_type() == torch::kBFloat16, "output must be bf16");
+
+  int ws = static_cast<int>(world_size);
+  int r = static_cast<int>(rank);
+
+  // Input: [N_total, D], Output: [N_per_rank, D]
+  int64_t N_total = input.size(0);
+  int64_t D = input.size(1);
+  TORCH_CHECK(N_total % ws == 0, "N_total must be divisible by world_size");
+  int64_t N_per_rank = N_total / ws;
+  TORCH_CHECK(output.size(0) == N_per_rank);
+  TORCH_CHECK(output.size(1) == D);
+
+  int64_t input_bytes = input.numel() * input.element_size();
+  TORCH_CHECK(input_bytes % 16 == 0,
+              "input byte size must be multiple of 16, got ", input_bytes);
+  TORCH_CHECK(input_bytes <= rank_stride, "input (", input_bytes,
+              " bytes) exceeds rank_stride (", rank_stride, " bytes)");
+
+  int total_sz = static_cast<int>(input_bytes / 16);
+  int slice_sz = total_sz / ws;
+  int slice_off = r * slice_sz;
+
+  int threads = 512;
+  int blocks =
+      std::max(1, std::min(kMaxBlocks, (total_sz + threads - 1) / threads));
+
+  auto* bp = reinterpret_cast<int64_t*>(buf_ptrs_ptr);
+  auto* ct = reinterpret_cast<int*>(counters_ptr);
+  int sc = static_cast<int>(seg_capacity);
+  int rs = static_cast<int>(rank_stride);
+
+#define LAUNCH(ng)                                                      \
+  moe_rs_lamport_kernel<ng><<<blocks, threads, 0, stream>>>(            \
+      bp, ct, r, sc, rs, input.data_ptr(), output.data_ptr(), total_sz, \
+      slice_off, slice_sz);
+
+  switch (ws) {
+    case 2:
+      LAUNCH(2);
+      break;
+    case 4:
+      LAUNCH(4);
+      break;
+    case 6:
+      LAUNCH(6);
+      break;
+    case 8:
+      LAUNCH(8);
+      break;
+    default:
+      TORCH_CHECK(false, "world_size must be 2, 4, 6, or 8");
+  }
+#undef LAUNCH
+}
+
+// ---------------------------------------------------------------------------
+// Python binding
+// ---------------------------------------------------------------------------
+
+PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
+  m.def("moe_reduce_scatter", &moe_reduce_scatter,
+        "Lamport MoE reduce-scatter");
+  m.def("lamport_init", &lamport_init,
+        "Initialize Lamport buffer with sentinels");
+}

csrc/standalone/custom_ar/moe_reduce_scatter.py

@@ -0,0 +1,102 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Lamport-based MoE reduce-scatter for EP combine.
+
+JIT-compiles the CUDA kernel on first use (cached afterwards).
+Uses the same Lamport sentinel protocol as the all-gather kernel.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import torch
+
+_lib = None
+
+
+def _load_lib():
+    global _lib
+    if _lib is not None:
+        return _lib
+
+    from torch.utils.cpp_extension import load
+
+    src = str(Path(__file__).with_name("moe_reduce_scatter.cu"))
+    _lib = load(
+        name="moe_reduce_scatter_kernel",
+        sources=[src],
+        extra_cuda_cflags=["-O3", "--use_fast_math"],
+        verbose=os.environ.get("MOE_RS_VERBOSE", "") == "1",
+    )
+    return _lib
+
+
+class MoeReduceScatter:
+    """Lamport-based MoE combine reduce-scatter with double buffering."""
+
+    def __init__(self, ca_comm):
+        self.rank = ca_comm.rank
+        self.world_size = ca_comm.world_size
+        self.device = ca_comm.device
+        self.buffer_ptrs = ca_comm.buffer_ptrs
+        self.max_size = ca_comm.max_size
+
+        ws = self.world_size
+        # Use the SECOND half of the IPC buffer (first half reserved for
+        # MoeAllGather) to avoid overlapping writes.
+        half_size = (self.max_size // 2) & ~15
+        self.buffer_offset = half_size
+
+        # Double-buffer layout within our half: 2 segments, each ws rank-slots.
+        self.seg_capacity = (half_size // 2) & ~15
+        self.rank_stride = (self.seg_capacity // ws) & ~15
+        self.max_per_rank = self.rank_stride
+
+        # Buffer pointer array on device — offset to our half.
+        self._buf_ptrs = torch.zeros(
+            8, dtype=torch.int64, device=f"cuda:{self.device.index}"
+        )
+        for i in range(ws):
+            self._buf_ptrs[i] = self.buffer_ptrs[i] + self.buffer_offset
+        self._buf_ptrs_ptr = self._buf_ptrs.data_ptr()
+
+        # Counters: [0]=unused, [1]=seg (0/1), [2]=prev_total_sz.
+        self._counters = torch.zeros(
+            3, dtype=torch.int32, device=f"cuda:{self.device.index}"
+        )
+        self._counters_ptr = self._counters.data_ptr()
+
+        # Initialize our half with sentinels.
+        lib = _load_lib()
+        lib.lamport_init(self.buffer_ptrs[self.rank] + self.buffer_offset, half_size)
+        torch.accelerator.synchronize(self.device)
+
+    def reduce_scatter(
+        self,
+        input: torch.Tensor,
+    ) -> torch.Tensor:
+        """Reduce-scatter input [N_total, D] bf16 → output [N_per_rank, D] bf16."""
+        lib = _load_lib()
+        ws = self.world_size
+
+        assert input.dim() == 2
+        N_total, D = input.shape
+        assert N_total % ws == 0
+        N_per_rank = N_total // ws
+
+        output = torch.empty((N_per_rank, D), dtype=input.dtype, device=input.device)
+
+        lib.moe_reduce_scatter(
+            self._buf_ptrs_ptr,
+            self._counters_ptr,
+            self.rank,
+            self.world_size,
+            self.seg_capacity,
+            self.rank_stride,
+            input,
+            output,
+        )
+        return output

csrc/standalone/custom_ar/moe_rs_fused.cu

@@ -0,0 +1,308 @@
+// Lamport reduce-scatter fused with residual add + RMSNorm.
+//
+// Replaces three separate kernels (RS + residual_add + RMSNorm) with one:
+//   1. PUSH: write MoE output to all peers' Lamport buffers.
+//   2. CLEAR: write sentinels to old segment.
+//   3. POLL+REDUCE+FUSE (per-token):
+//      a. Volatile-load from all peers, sum in fp32.
+//      b. Add residual.
+//      c. Compute RMSNorm (block reduction for variance).
+//      d. Store normed output + updated residual.
+//
+// Saves: one kernel launch (~3-5µs) + one global memory round-trip per layer.
+
+#include <cuda.h>
+#include <cuda_bf16.h>
+#include <cuda_runtime.h>
+#include <torch/extension.h>
+
+#include <c10/cuda/CUDAGuard.h>
+#include <c10/cuda/CUDAStream.h>
+
+#define DINLINE __device__ __forceinline__
+
+constexpr uint32_t SENTINEL = 0x80000000u;
+constexpr int kMaxBlocks = 36;
+
+// Each token has D=7168 bf16 values = 896 int4 vectors.
+// With 512 threads: ceil(896/512) = 2 int4 per thread = 16 fp32 values.
+constexpr int kMaxValsPerThread = 16;
+
+static DINLINE int4 ld128v(const void* addr) {
+  int4 v;
+  asm volatile("ld.volatile.global.v4.b32 {%0,%1,%2,%3}, [%4];"
+               : "=r"(v.x), "=r"(v.y), "=r"(v.z), "=r"(v.w)
+               : "l"(addr));
+  return v;
+}
+
+static DINLINE bool has_sentinel(int4 v) {
+  return reinterpret_cast<uint32_t&>(v.x) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.y) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.z) == SENTINEL |
+         reinterpret_cast<uint32_t&>(v.w) == SENTINEL;
+}
+
+static DINLINE int4 remove_sentinel(int4 v) {
+  if (reinterpret_cast<uint32_t&>(v.x) == SENTINEL) v.x = 0;
+  if (reinterpret_cast<uint32_t&>(v.y) == SENTINEL) v.y = 0;
+  if (reinterpret_cast<uint32_t&>(v.z) == SENTINEL) v.z = 0;
+  if (reinterpret_cast<uint32_t&>(v.w) == SENTINEL) v.w = 0;
+  return v;
+}
+
+// bf16 helpers
+static DINLINE void accumulate_bf16(float* acc, int4 v) {
+  const __nv_bfloat16* bp = reinterpret_cast<const __nv_bfloat16*>(&v);
+#pragma unroll
+  for (int k = 0; k < 8; k++) acc[k] += __bfloat162float(bp[k]);
+}
+
+static DINLINE void add_bf16_to_fp32(float* dst, int4 v) {
+  const __nv_bfloat16* bp = reinterpret_cast<const __nv_bfloat16*>(&v);
+#pragma unroll
+  for (int k = 0; k < 8; k++) dst[k] += __bfloat162float(bp[k]);
+}
+
+static DINLINE int4 fp32_to_bf16_int4(const float* vals) {
+  int4 out;
+  __nv_bfloat16* bp = reinterpret_cast<__nv_bfloat16*>(&out);
+#pragma unroll
+  for (int k = 0; k < 8; k++) bp[k] = __float2bfloat16(vals[k]);
+  return out;
+}
+
+// Block-level tree reduction in shared memory.
+static DINLINE float block_reduce_sum(float val, float* smem) {
+  smem[threadIdx.x] = val;
+  __syncthreads();
+  for (int s = blockDim.x / 2; s > 0; s >>= 1) {
+    if (threadIdx.x < s) smem[threadIdx.x] += smem[threadIdx.x + s];
+    __syncthreads();
+  }
+  return smem[0];
+}
+
+// ---------------------------------------------------------------------------
+// Fused reduce-scatter + residual + RMSNorm kernel
+// ---------------------------------------------------------------------------
+
+template <int ngpus>
+__global__ void __launch_bounds__(512, 1) moe_rs_fused_kernel(
+    int64_t* buf_ptrs, int* counters, int rank, int seg_capacity,
+    int rank_stride,
+    const void* input,        // [N_total, D] bf16 — MoE output
+    const void* residual_in,  // [N_per_rank, D] bf16 — skip connection
+    const void* gamma,        // [D] bf16 — RMSNorm weight
+    void* normed_out,         // [N_per_rank, D] bf16 — normed result
+    void* residual_out,       // [N_per_rank, D] bf16 — updated residual
+    int total_sz,             // int4 units of full input
+    int slice_off,            // int4 offset to this rank's slice
+    int slice_sz,             // int4 units of this rank's slice
+    int D_int4,               // int4 units per token (hidden_dim * 2 / 16)
+    int N_per_rank,           // tokens in this rank's slice
+    float eps) {              // RMSNorm epsilon
+  using V = int4;
+  const int tid = blockIdx.x * blockDim.x + threadIdx.x;
+  const int stride = gridDim.x * blockDim.x;
+
+  const int seg = counters[1];
+  const int prev_total_sz = counters[2];
+  const int cur_seg = seg;
+  const int old_seg = 1 - seg;
+
+  char* bufs[ngpus];
+#pragma unroll
+  for (int r = 0; r < ngpus; r++)
+    bufs[r] = reinterpret_cast<char*>(buf_ptrs[r]) + cur_seg * seg_capacity;
+
+  V sent;
+  sent.x = sent.y = sent.z = sent.w = static_cast<int>(SENTINEL);
+
+  // ---- Phase 1: PUSH full MoE output to ALL peers ----
+  {
+    const V* src = reinterpret_cast<const V*>(input);
+    for (int i = tid; i < total_sz; i += stride) {
+      V val = remove_sentinel(src[i]);
+#pragma unroll
+      for (int r = 0; r < ngpus; r++)
+        reinterpret_cast<V*>(bufs[r] + rank * rank_stride)[i] = val;
+    }
+  }
+
+  // ---- Phase 2: CLEAR old segment ----
+  if (prev_total_sz > 0) {
+    char* clr_base =
+        reinterpret_cast<char*>(buf_ptrs[rank]) + old_seg * seg_capacity;
+#pragma unroll
+    for (int r = 0; r < ngpus; r++) {
+      V* clr = reinterpret_cast<V*>(clr_base + r * rank_stride);
+      for (int i = tid; i < prev_total_sz; i += stride) clr[i] = sent;
+    }
+  }
+
+  // ---- Phase 3: Fused POLL + REDUCE + RESIDUAL + RMSNORM ----
+  // Each block handles one token. Only first N_per_rank blocks participate.
+  if (blockIdx.x < N_per_rank) {
+    const int token = blockIdx.x;
+    const int token_off = slice_off + token * D_int4;  // in the full buffer
+    char* my = bufs[rank];
+
+    extern __shared__ float smem[];
+
+    // Register storage for intermediate fp32 values.
+    float local_vals[kMaxValsPerThread];
+    int n_vals = 0;
+    float partial_sum_sq = 0.0f;
+
+    // Pass 1: poll + reduce + add residual + compute sum_sq
+    for (int pos = threadIdx.x; pos < D_int4; pos += blockDim.x) {
+      float acc[8] = {0, 0, 0, 0, 0, 0, 0, 0};
+
+      // Poll all ranks' data for this position.
+#pragma unroll
+      for (int s = 0; s < ngpus; s++) {
+        V val;
+        do {
+          val = ld128v(reinterpret_cast<V*>(my + s * rank_stride) + token_off +
+                       pos);
+        } while (has_sentinel(val));
+        accumulate_bf16(acc, val);
+      }
+
+      // Add residual.
+      V res = reinterpret_cast<const V*>(residual_in)[token * D_int4 + pos];
+      add_bf16_to_fp32(acc, res);
+
+      // Store in registers and accumulate sum_sq.
+#pragma unroll
+      for (int k = 0; k < 8; k++) {
+        local_vals[n_vals++] = acc[k];
+        partial_sum_sq += acc[k] * acc[k];
+      }
+    }
+
+    // Block-level reduction: total sum of squares.
+    float total_sum_sq = block_reduce_sum(partial_sum_sq, smem);
+    float rms_scale = rsqrtf(total_sum_sq / (D_int4 * 8) + eps);
+
+    // Pass 2: apply RMSNorm, store outputs.
+    n_vals = 0;
+    const V* gamma_v = reinterpret_cast<const V*>(gamma);
+    for (int pos = threadIdx.x; pos < D_int4; pos += blockDim.x) {
+      V gv = gamma_v[pos];
+      const __nv_bfloat16* gp = reinterpret_cast<const __nv_bfloat16*>(&gv);
+
+      // Build normed output and residual output.
+      float normed_fp32[8], res_fp32[8];
+#pragma unroll
+      for (int k = 0; k < 8; k++) {
+        float val = local_vals[n_vals++];
+        res_fp32[k] = val;  // residual_out
+        normed_fp32[k] = val * rms_scale * __bfloat162float(gp[k]);  // normed
+      }
+
+      reinterpret_cast<V*>(normed_out)[token * D_int4 + pos] =
+          fp32_to_bf16_int4(normed_fp32);
+      reinterpret_cast<V*>(residual_out)[token * D_int4 + pos] =
+          fp32_to_bf16_int4(res_fp32);
+    }
+  }
+
+  // ---- Phase 4: ADVANCE ----
+  if (blockIdx.x == 0 && threadIdx.x == 0) {
+    counters[1] = 1 - seg;
+    counters[2] = total_sz;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Sentinel init + host launcher
+// ---------------------------------------------------------------------------
+
+__global__ void lamport_init_kernel(uint32_t* buf, int n) {
+  int tid = blockIdx.x * blockDim.x + threadIdx.x;
+  int stride = gridDim.x * blockDim.x;
+  for (int i = tid; i < n; i += stride) buf[i] = SENTINEL;
+}
+
+void lamport_init(int64_t buf_ptr, int64_t nbytes) {
+  auto stream = c10::cuda::getCurrentCUDAStream().stream();
+  int n = static_cast<int>(nbytes / 4);
+  lamport_init_kernel<<<256, 256, 0, stream>>>(
+      reinterpret_cast<uint32_t*>(buf_ptr), n);
+}
+
+void moe_rs_fused(int64_t buf_ptrs_ptr, int64_t counters_ptr, int64_t rank,
+                  int64_t world_size, int64_t seg_capacity, int64_t rank_stride,
+                  torch::Tensor input,         // [N_total, D] bf16
+                  torch::Tensor residual_in,   // [N_per_rank, D] bf16
+                  torch::Tensor gamma,         // [D] bf16
+                  torch::Tensor normed_out,    // [N_per_rank, D] bf16
+                  torch::Tensor residual_out,  // [N_per_rank, D] bf16
+                  double eps) {
+  auto stream = c10::cuda::getCurrentCUDAStream().stream();
+  TORCH_CHECK(input.is_contiguous() && residual_in.is_contiguous());
+  TORCH_CHECK(gamma.is_contiguous() && normed_out.is_contiguous());
+  TORCH_CHECK(residual_out.is_contiguous());
+  TORCH_CHECK(input.scalar_type() == torch::kBFloat16);
+
+  int ws = static_cast<int>(world_size);
+  int r = static_cast<int>(rank);
+  int64_t N_total = input.size(0);
+  int64_t D = input.size(1);
+  TORCH_CHECK(N_total % ws == 0);
+  int N_per_rank = static_cast<int>(N_total / ws);
+
+  int64_t input_bytes = input.numel() * input.element_size();
+  TORCH_CHECK(input_bytes % 16 == 0);
+  TORCH_CHECK(input_bytes <= rank_stride);
+
+  int total_sz = static_cast<int>(input_bytes / 16);
+  int D_int4 = static_cast<int>(D * 2 / 16);  // bf16 elements → int4 units
+  int slice_sz = total_sz / ws;
+  int slice_off = r * slice_sz;
+
+  int threads = 512;
+  // Need at least N_per_rank blocks for Phase 3 (one per token).
+  int blocks = std::max(
+      N_per_rank, std::min(kMaxBlocks, (total_sz + threads - 1) / threads));
+
+  auto* bp = reinterpret_cast<int64_t*>(buf_ptrs_ptr);
+  auto* ct = reinterpret_cast<int*>(counters_ptr);
+  int sc = static_cast<int>(seg_capacity);
+  int rs = static_cast<int>(rank_stride);
+  int smem = threads * sizeof(float);
+
+#define LAUNCH(ng)                                                      \
+  moe_rs_fused_kernel<ng><<<blocks, threads, smem, stream>>>(           \
+      bp, ct, r, sc, rs, input.data_ptr(), residual_in.data_ptr(),      \
+      gamma.data_ptr(), normed_out.data_ptr(), residual_out.data_ptr(), \
+      total_sz, slice_off, slice_sz, D_int4, N_per_rank,                \
+      static_cast<float>(eps));
+
+  switch (ws) {
+    case 2:
+      LAUNCH(2);
+      break;
+    case 4:
+      LAUNCH(4);
+      break;
+    case 6:
+      LAUNCH(6);
+      break;
+    case 8:
+      LAUNCH(8);
+      break;
+    default:
+      TORCH_CHECK(false, "world_size must be 2, 4, 6, or 8");
+  }
+#undef LAUNCH
+}
+
+PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
+  m.def("moe_rs_fused", &moe_rs_fused,
+        "Fused Lamport reduce-scatter + residual + RMSNorm");
+  m.def("lamport_init", &lamport_init,
+        "Initialize Lamport buffer with sentinels");
+}

csrc/standalone/custom_ar/moe_rs_fused.py

@@ -0,0 +1,29 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+JIT wrapper for the fused reduce-scatter + residual + RMSNorm kernel.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+_lib = None
+
+
+def _load_lib():
+    global _lib
+    if _lib is not None:
+        return _lib
+
+    from torch.utils.cpp_extension import load
+
+    src = str(Path(__file__).with_name("moe_rs_fused.cu"))
+    _lib = load(
+        name="moe_rs_fused_kernel",
+        sources=[src],
+        extra_cuda_cflags=["-O3", "--use_fast_math"],
+        verbose=os.environ.get("MOE_RS_FUSED_VERBOSE", "") == "1",
+    )
+    return _lib

csrc/standalone/custom_ar/test_moe_allgather.py

@@ -0,0 +1,137 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Test for the Lamport-based MoE all-gather kernel."""
+
+import ctypes
+import os
+import sys
+
+import torch
+import torch.distributed as dist
+
+_cudart = ctypes.CDLL("libcudart.so")
+IPC = 64
+
+
+def _cc(r):
+    if r:
+        raise RuntimeError(f"CUDA err {r}")
+
+
+def ipc_buf(sz, rank, ws):
+    p = ctypes.c_void_p()
+    _cc(_cudart.cudaMalloc(ctypes.byref(p), sz))
+    _cc(_cudart.cudaMemset(p, 0, sz))
+    _cc(_cudart.cudaDeviceSynchronize())
+    h = (ctypes.c_byte * IPC)()
+    _cc(_cudart.cudaIpcGetMemHandle(ctypes.byref(h), p))
+    ah = [None] * ws
+    dist.all_gather_object(ah, bytes(h))
+    ptrs = []
+    for i in range(ws):
+        if i == rank:
+            ptrs.append(p.value)
+        else:
+            hh = (ctypes.c_byte * IPC)(*ah[i])
+            pp = ctypes.c_void_p()
+            _cc(_cudart.cudaIpcOpenMemHandle(ctypes.byref(pp), hh, ctypes.c_uint(1)))
+            ptrs.append(pp.value)
+    return ptrs
+
+
+def main():
+    dist.init_process_group("nccl")
+    rank = dist.get_rank()
+    ws = dist.get_world_size()
+    torch.cuda.set_device(rank)
+    dev = f"cuda:{rank}"
+
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    if rank == 0:
+        from moe_allgather import MoeAllGather, _load_lib
+
+        _load_lib()
+    dist.barrier()
+    from moe_allgather import MoeAllGather, _load_lib
+
+    _load_lib()
+    dist.barrier()
+
+    max_size = 8 * 1024 * 1024
+    bp = ipc_buf(max_size, rank, ws)
+    dist.barrier()
+
+    # Build a fake ca_comm-like object.
+    class FakeCA:
+        pass
+
+    ca = FakeCA()
+    ca.rank = rank
+    ca.world_size = ws
+    ca.device = torch.device(dev)
+    ca.buffer_ptrs = bp
+    ca.max_size = max_size
+    # meta_ptrs not needed for Lamport approach
+    ag = MoeAllGather(ca)
+    dist.barrier()
+
+    errors = 0
+
+    # Test with various token counts.
+    for N in [1, 4, 16, 64]:
+        topk = 8
+        hd = 3584
+        sd = 448
+        ids = (
+            torch.arange(N * topk, dtype=torch.int32, device=dev) + rank * 1000
+        ).reshape(N, topk)
+        wt = torch.ones(N, topk, dtype=torch.float32, device=dev) * (rank + 1) * 0.1
+        hs = torch.full((N, hd), rank + 1, dtype=torch.uint8, device=dev)
+        sc = torch.full((N, sd), rank + 1, dtype=torch.uint8, device=dev)
+
+        ids_g, wt_g, hs_g, sc_g = ag.gather(ids, wt, hs, sc)
+
+        for src in range(ws):
+            s, e = src * N, (src + 1) * N
+            exp_ids = (
+                torch.arange(N * topk, dtype=torch.int32, device=dev) + src * 1000
+            ).reshape(N, topk)
+            if not torch.equal(ids_g[s:e], exp_ids):
+                print(f"[{rank}] FAIL ids src={src} N={N}")
+                errors += 1
+            exp_wt = torch.full(
+                (N, topk), (src + 1) * 0.1, dtype=torch.float32, device=dev
+            )
+            if not torch.allclose(wt_g[s:e], exp_wt):
+                print(f"[{rank}] FAIL wt src={src} N={N}")
+                errors += 1
+            exp_hs = torch.full((N, hd), src + 1, dtype=torch.uint8, device=dev)
+            if not torch.equal(hs_g[s:e], exp_hs):
+                print(f"[{rank}] FAIL hs src={src} N={N}")
+                errors += 1
+            exp_sc = torch.full((N, sd), src + 1, dtype=torch.uint8, device=dev)
+            if not torch.equal(sc_g[s:e], exp_sc):
+                print(f"[{rank}] FAIL sc src={src} N={N}")
+                errors += 1
+
+        # Without scales.
+        ids_g2, wt_g2, hs_g2, _ = ag.gather(ids, wt, hs)
+        for src in range(ws):
+            s, e = src * N, (src + 1) * N
+            exp_ids = (
+                torch.arange(N * topk, dtype=torch.int32, device=dev) + src * 1000
+            ).reshape(N, topk)
+            if not torch.equal(ids_g2[s:e], exp_ids):
+                print(f"[{rank}] FAIL no-sc ids src={src} N={N}")
+                errors += 1
+
+    dist.barrier()
+    print(
+        f"[rank {rank}] {'PASSED' if errors == 0 else f'FAILED ({errors})'} (ws={ws})"
+    )
+    dist.destroy_process_group()
+    return errors
+
+
+if __name__ == "__main__":
+    sys.exit(main())

csrc/standalone/custom_ar/test_moe_allgather_stress.py

@@ -0,0 +1,173 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Stress test: random data, check bitwise correctness against NCCL."""
+
+import ctypes
+import os
+import sys
+
+import torch
+import torch.distributed as dist
+
+_cudart = ctypes.CDLL("libcudart.so")
+IPC = 64
+
+
+def _cc(r):
+    if r:
+        raise RuntimeError(f"CUDA err {r}")
+
+
+def ipc_buf(sz, rank, ws):
+    p = ctypes.c_void_p()
+    _cc(_cudart.cudaMalloc(ctypes.byref(p), sz))
+    _cc(_cudart.cudaMemset(p, 0, sz))
+    _cc(_cudart.cudaDeviceSynchronize())
+    h = (ctypes.c_byte * IPC)()
+    _cc(_cudart.cudaIpcGetMemHandle(ctypes.byref(h), p))
+    ah = [None] * ws
+    dist.all_gather_object(ah, bytes(h))
+    ptrs = []
+    for i in range(ws):
+        if i == rank:
+            ptrs.append(p.value)
+        else:
+            hh = (ctypes.c_byte * IPC)(*ah[i])
+            pp = ctypes.c_void_p()
+            _cc(_cudart.cudaIpcOpenMemHandle(ctypes.byref(pp), hh, ctypes.c_uint(1)))
+            ptrs.append(pp.value)
+    return ptrs
+
+
+def main():
+    dist.init_process_group("nccl")
+    rank = dist.get_rank()
+    ws = dist.get_world_size()
+    torch.cuda.set_device(rank)
+    dev = f"cuda:{rank}"
+
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    if rank == 0:
+        from moe_allgather import _load_lib
+
+        _load_lib()
+    dist.barrier()
+    from moe_allgather import MoeAllGather, _load_lib
+
+    _load_lib()
+    dist.barrier()
+
+    max_size = 8 * 1024 * 1024
+    bp = ipc_buf(max_size, rank, ws)
+    dist.barrier()
+
+    class FakeCA:
+        pass
+
+    ca = FakeCA()
+    ca.rank = rank
+    ca.world_size = ws
+    ca.device = torch.device(dev)
+    ca.buffer_ptrs = bp
+    ca.max_size = max_size
+    ag = MoeAllGather(ca)
+    dist.barrier()
+
+    topk = 8
+    hd = 3584
+    sd = 448
+    errors = 0
+    total_checks = 0
+    sentinel_collisions = 0
+
+    for trial in range(200):
+        # All ranks must use the same N for NCCL reference.
+        N_tensor = torch.randint(1, 65, (1,), device=dev)
+        dist.broadcast(N_tensor, src=0)
+        N = N_tensor.item()
+        # Random data including possible sentinel values
+        ids = torch.randint(0, 256, (N, topk), dtype=torch.int32, device=dev)
+        wt = torch.randn(N, topk, dtype=torch.float32, device=dev)
+        hs = torch.randint(0, 256, (N, hd), dtype=torch.uint8, device=dev)
+        sc = torch.randint(0, 256, (N, sd), dtype=torch.uint8, device=dev)
+
+        # Count sentinel patterns in hidden_states (as uint32 view)
+        hs_u32 = hs.view(torch.int32)
+        sentinel_collisions += (hs_u32 == 0x80000000).sum().item()
+
+        # Custom kernel
+        ids_g, wt_g, hs_g, sc_g = ag.gather(ids, wt, hs, sc)
+
+        # NCCL reference
+        ids_ref = torch.empty(N * ws, topk, dtype=torch.int32, device=dev)
+        wt_ref = torch.empty(N * ws, topk, dtype=torch.float32, device=dev)
+        hs_ref = torch.empty(N * ws, hd, dtype=torch.uint8, device=dev)
+        sc_ref = torch.empty(N * ws, sd, dtype=torch.uint8, device=dev)
+        dist.all_gather_into_tensor(ids_ref, ids)
+        dist.all_gather_into_tensor(wt_ref, wt)
+        dist.all_gather_into_tensor(hs_ref, hs)
+        dist.all_gather_into_tensor(sc_ref, sc)
+
+        # Compare
+        if not torch.equal(ids_g, ids_ref):
+            mismatches = (ids_g != ids_ref).sum().item()
+            if trial < 5 or mismatches > 0:
+                print(f"[{rank}] trial={trial} ids MISMATCH: {mismatches} elements")
+            errors += 1
+        if not torch.equal(wt_g, wt_ref):
+            # Check for -0 vs +0 differences
+            bit_diff = wt_g.view(torch.int32) != wt_ref.view(torch.int32)
+            neg_zero_mask = wt_ref.view(torch.int32) == 0x80000000
+            real_errors = bit_diff & ~neg_zero_mask
+            if real_errors.any():
+                print(
+                    f"[{rank}] trial={trial} wt MISMATCH (non-negzero): {real_errors.sum().item()}"
+                )
+                errors += 1
+        if not torch.equal(hs_g, hs_ref):
+            mismatches = (hs_g != hs_ref).sum().item()
+            # Check if mismatches are due to sentinel collision
+            hs_g_u32 = hs_g.view(torch.int32)
+            hs_ref_u32 = hs_ref.view(torch.int32)
+            diff_mask = hs_g_u32 != hs_ref_u32
+            sentinel_mask = (hs_ref_u32 == 0x80000000) & diff_mask
+            non_sentinel = diff_mask & ~sentinel_mask
+            if non_sentinel.any():
+                print(
+                    f"[{rank}] trial={trial} hs NON-SENTINEL MISMATCH: {non_sentinel.sum().item()}"
+                )
+                errors += 1
+            elif sentinel_mask.any():
+                if trial < 3:
+                    print(
+                        f"[{rank}] trial={trial} hs sentinel collision: "
+                        f"{sentinel_mask.sum().item()} words (expected rare)"
+                    )
+        if not torch.equal(sc_g, sc_ref):
+            mismatches = (sc_g != sc_ref).sum().item()
+            sc_g_u32 = sc_g.view(torch.int32) if sc_g.numel() % 4 == 0 else None
+            if sc_g_u32 is not None:
+                sc_ref_u32 = sc_ref.view(torch.int32)
+                diff_mask = sc_g_u32 != sc_ref_u32
+                sentinel_mask = (sc_ref_u32 == 0x80000000) & diff_mask
+                non_sentinel = diff_mask & ~sentinel_mask
+                if non_sentinel.any():
+                    print(
+                        f"[{rank}] trial={trial} sc NON-SENTINEL MISMATCH: {non_sentinel.sum().item()}"
+                    )
+                    errors += 1
+
+        total_checks += 1
+
+    dist.barrier()
+    print(
+        f"[rank {rank}] {total_checks} trials, {errors} real errors, "
+        f"{sentinel_collisions} sentinel patterns in hs data. "
+        f"{'PASSED' if errors == 0 else 'FAILED'}"
+    )
+    dist.destroy_process_group()
+    return errors
+
+
+if __name__ == "__main__":
+    sys.exit(main())

csrc/standalone/custom_ar/test_moe_reduce_scatter.py

@@ -0,0 +1,204 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Test + benchmark for Lamport MoE reduce-scatter."""
+
+import ctypes
+import os
+import sys
+
+import torch
+import torch.distributed as dist
+
+_cudart = ctypes.CDLL("libcudart.so")
+IPC = 64
+
+
+def _cc(r):
+    if r:
+        raise RuntimeError(f"CUDA err {r}")
+
+
+def ipc_buf(sz, rank, ws):
+    p = ctypes.c_void_p()
+    _cc(_cudart.cudaMalloc(ctypes.byref(p), sz))
+    _cc(_cudart.cudaMemset(p, 0, sz))
+    _cc(_cudart.cudaDeviceSynchronize())
+    h = (ctypes.c_byte * IPC)()
+    _cc(_cudart.cudaIpcGetMemHandle(ctypes.byref(h), p))
+    ah = [None] * ws
+    dist.all_gather_object(ah, bytes(h))
+    ptrs = []
+    for i in range(ws):
+        if i == rank:
+            ptrs.append(p.value)
+        else:
+            hh = (ctypes.c_byte * IPC)(*ah[i])
+            pp = ctypes.c_void_p()
+            _cc(_cudart.cudaIpcOpenMemHandle(ctypes.byref(pp), hh, ctypes.c_uint(1)))
+            ptrs.append(pp.value)
+    return ptrs
+
+
+def gpu_timer_graph(fn, warmup=20, repeats=200):
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    g = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(g):
+        fn()
+    for _ in range(5):
+        g.replay()
+    torch.cuda.synchronize()
+    s = torch.cuda.Event(enable_timing=True)
+    e = torch.cuda.Event(enable_timing=True)
+    s.record()
+    for _ in range(repeats):
+        g.replay()
+    e.record()
+    torch.cuda.synchronize()
+    return s.elapsed_time(e) / repeats * 1000
+
+
+def main():
+    dist.init_process_group("nccl")
+    rank = dist.get_rank()
+    ws = dist.get_world_size()
+    torch.cuda.set_device(rank)
+    dev = f"cuda:{rank}"
+
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    if rank == 0:
+        from moe_reduce_scatter import _load_lib
+
+        _load_lib()
+    dist.barrier()
+    if rank != 0:
+        from moe_reduce_scatter import _load_lib
+
+        _load_lib()
+    dist.barrier()
+    from moe_reduce_scatter import MoeReduceScatter
+
+    max_size = 8 * 1024 * 1024
+    bp = ipc_buf(max_size, rank, ws)
+    dist.barrier()
+
+    class FakeCA:
+        pass
+
+    ca = FakeCA()
+    ca.rank = rank
+    ca.world_size = ws
+    ca.device = torch.device(dev)
+    ca.buffer_ptrs = bp
+    ca.max_size = max_size
+    rs = MoeReduceScatter(ca)
+    dist.barrier()
+
+    D = 7168  # DeepSeek V3 hidden_dim
+    errors = 0
+
+    # ---- Correctness tests ----
+    for N_per_rank in [1, 4, 16]:
+        N_total = N_per_rank * ws
+
+        # Each rank gets a deterministic input.
+        torch.manual_seed(42)
+        # All ranks create the SAME "ground truth" inputs for each rank.
+        all_inputs = [
+            torch.randn(N_total, D, dtype=torch.bfloat16, device=dev) for _ in range(ws)
+        ]
+        # This rank's input is all_inputs[rank].
+        my_input = all_inputs[rank]
+
+        # Custom reduce-scatter.
+        custom_out = rs.reduce_scatter(my_input)
+
+        # NCCL reference: reduce_scatter_tensor.
+        nccl_out = torch.empty(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+        dist.reduce_scatter_tensor(nccl_out, my_input)
+
+        # bf16 summation order differs between our kernel and NCCL,
+        # giving ~1-2 ULP differences. Use generous tolerance.
+        max_diff = (custom_out.float() - nccl_out.float()).abs().max().item()
+        if not torch.allclose(custom_out, nccl_out, atol=0.125, rtol=0.01):
+            mismatches = (
+                ((custom_out.float() - nccl_out.float()).abs() > 0.125).sum().item()
+            )
+            print(
+                f"[{rank}] N_per_rank={N_per_rank} MISMATCH: "
+                f"max_diff={max_diff:.6f}, mismatches={mismatches}"
+            )
+            errors += 1
+        else:
+            if rank == 0:
+                print(f"  N_per_rank={N_per_rank}: PASS (max_diff={max_diff:.6f})")
+
+    # ---- Benchmark ----
+    if rank == 0:
+        print(f"\nworld_size={ws}, max_per_rank={rs.max_per_rank} bytes")
+        print(
+            f"{'config':<12} {'lamport':>10} {'lamp_graph':>10} "
+            f"{'nccl':>10} {'nccl_graph':>10} {'speedup':>8}"
+        )
+        print("-" * 65)
+
+    configs = [
+        ("1tok", 1),
+        ("2tok", 2),
+        ("4tok", 4),
+        ("8tok", 8),
+        ("16tok", 16),
+        ("32tok", 32),
+        ("64tok", 64),
+        ("128tok", 128),
+        ("256tok", 256),
+    ]
+
+    for name, N_per_rank in configs:
+        N_total = N_per_rank * ws
+        input_bytes = N_total * D * 2  # bf16
+        if input_bytes > rs.max_per_rank:
+            if rank == 0:
+                print(f"{name:<12} {'skip (too large)':>40}")
+            continue
+
+        inp = torch.randn(N_total, D, dtype=torch.bfloat16, device=dev)
+        c_out = torch.empty(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+        n_out = torch.empty(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+
+        def run_lamport():
+            rs.reduce_scatter(inp)
+
+        def run_nccl():
+            dist.reduce_scatter_tensor(n_out, inp)
+
+        from bench_moe_allgather import gpu_timer
+
+        lam_us = gpu_timer(run_lamport)
+        try:
+            lam_g_us = gpu_timer_graph(run_lamport)
+        except Exception:
+            lam_g_us = float("nan")
+
+        nccl_us = gpu_timer(run_nccl)
+        try:
+            nccl_g_us = gpu_timer_graph(run_nccl)
+        except Exception:
+            nccl_g_us = float("nan")
+
+        if rank == 0:
+            speedup = nccl_g_us / lam_g_us if lam_g_us > 0 else float("nan")
+            print(
+                f"{name:<12} {lam_us:>9.1f}µ {lam_g_us:>9.1f}µ "
+                f"{nccl_us:>9.1f}µ {nccl_g_us:>9.1f}µ {speedup:>7.2f}x"
+            )
+
+    dist.barrier()
+    print(f"[rank {rank}] {'PASSED' if errors == 0 else f'FAILED ({errors})'}")
+    dist.destroy_process_group()
+    return errors
+
+
+if __name__ == "__main__":
+    sys.exit(main())

csrc/standalone/custom_ar/test_rs_fused.py

@@ -0,0 +1,270 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Test + benchmark: fused RS + residual + RMSNorm vs separate kernels."""
+
+import ctypes
+import os
+import sys
+
+import torch
+import torch.distributed as dist
+
+_cudart = ctypes.CDLL("libcudart.so")
+IPC = 64
+
+
+def _cc(r):
+    if r:
+        raise RuntimeError(f"CUDA err {r}")
+
+
+def ipc_buf(sz, rank, ws):
+    p = ctypes.c_void_p()
+    _cc(_cudart.cudaMalloc(ctypes.byref(p), sz))
+    _cc(_cudart.cudaMemset(p, 0, sz))
+    _cc(_cudart.cudaDeviceSynchronize())
+    h = (ctypes.c_byte * IPC)()
+    _cc(_cudart.cudaIpcGetMemHandle(ctypes.byref(h), p))
+    ah = [None] * ws
+    dist.all_gather_object(ah, bytes(h))
+    ptrs = []
+    for i in range(ws):
+        if i == rank:
+            ptrs.append(p.value)
+        else:
+            hh = (ctypes.c_byte * IPC)(*ah[i])
+            pp = ctypes.c_void_p()
+            _cc(_cudart.cudaIpcOpenMemHandle(ctypes.byref(pp), hh, ctypes.c_uint(1)))
+            ptrs.append(pp.value)
+    return ptrs
+
+
+def rms_norm_ref(x, gamma, eps):
+    """Reference RMSNorm in fp32."""
+    xf = x.float()
+    rms = torch.rsqrt(xf.pow(2).mean(-1, keepdim=True) + eps)
+    return (xf * rms * gamma.float()).to(x.dtype)
+
+
+def gpu_timer(fn, warmup=20, repeats=200):
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    s = torch.cuda.Event(enable_timing=True)
+    e = torch.cuda.Event(enable_timing=True)
+    s.record()
+    for _ in range(repeats):
+        fn()
+    e.record()
+    torch.cuda.synchronize()
+    return s.elapsed_time(e) / repeats * 1000
+
+
+def gpu_timer_graph(fn, warmup=20, repeats=200):
+    for _ in range(warmup):
+        fn()
+    torch.cuda.synchronize()
+    g = torch.cuda.CUDAGraph()
+    with torch.cuda.graph(g):
+        fn()
+    for _ in range(5):
+        g.replay()
+    torch.cuda.synchronize()
+    s = torch.cuda.Event(enable_timing=True)
+    e = torch.cuda.Event(enable_timing=True)
+    s.record()
+    for _ in range(repeats):
+        g.replay()
+    e.record()
+    torch.cuda.synchronize()
+    return s.elapsed_time(e) / repeats * 1000
+
+
+def main():
+    dist.init_process_group("nccl")
+    rank = dist.get_rank()
+    ws = dist.get_world_size()
+    torch.cuda.set_device(rank)
+    dev = f"cuda:{rank}"
+
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+    # Compile fused kernel
+    if rank == 0:
+        from torch.utils.cpp_extension import load
+
+        load(
+            name="moe_rs_fused_kernel",
+            sources=[os.path.join(os.path.dirname(__file__), "moe_rs_fused.cu")],
+            extra_cuda_cflags=["-O3", "--use_fast_math"],
+            verbose=False,
+        )
+    dist.barrier()
+    from torch.utils.cpp_extension import load
+
+    fused_lib = load(
+        name="moe_rs_fused_kernel",
+        sources=[os.path.join(os.path.dirname(__file__), "moe_rs_fused.cu")],
+        extra_cuda_cflags=["-O3", "--use_fast_math"],
+        verbose=False,
+    )
+
+    # Also compile separate RS kernel for comparison
+    from moe_reduce_scatter import MoeReduceScatter
+
+    max_size = 8 * 1024 * 1024
+    bp = ipc_buf(max_size, rank, ws)
+    dist.barrier()
+
+    class FakeCA:
+        pass
+
+    ca = FakeCA()
+    ca.rank = rank
+    ca.world_size = ws
+    ca.device = torch.device(dev)
+    ca.buffer_ptrs = bp
+    ca.max_size = max_size
+
+    rs_separate = MoeReduceScatter(ca)
+
+    # Fused kernel setup (uses same buffer layout as MoeReduceScatter)
+    half_size = (max_size // 2) & ~15
+    buf_offset = half_size
+    fused_seg_cap = (half_size // 2) & ~15
+    fused_rank_stride = (fused_seg_cap // ws) & ~15
+
+    fused_buf_ptrs = torch.zeros(8, dtype=torch.int64, device=dev)
+    for i in range(ws):
+        fused_buf_ptrs[i] = bp[i] + buf_offset
+    fused_counters = torch.zeros(3, dtype=torch.int32, device=dev)
+
+    # Init sentinels for fused kernel's buffer region
+    fused_lib.lamport_init(bp[rank] + buf_offset, half_size)
+    torch.cuda.synchronize()
+    dist.barrier()
+
+    D = 7168
+    eps = 1e-6
+    gamma = torch.randn(D, dtype=torch.bfloat16, device=dev).abs() + 0.5
+    errors = 0
+
+    # ---- Correctness ----
+    for N_per_rank in [1, 4]:
+        N_total = N_per_rank * ws
+        torch.manual_seed(42 + rank)
+        moe_out = torch.randn(N_total, D, dtype=torch.bfloat16, device=dev)
+        residual = torch.randn(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+
+        # Reference: NCCL RS + add + norm
+        rs_ref = torch.empty(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+        dist.reduce_scatter_tensor(rs_ref, moe_out)
+        ref_residual = residual + rs_ref
+        ref_normed = rms_norm_ref(ref_residual, gamma, eps)
+
+        # Fused kernel
+        normed_out = torch.empty(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+        residual_out = torch.empty(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+        fused_lib.moe_rs_fused(
+            fused_buf_ptrs.data_ptr(),
+            fused_counters.data_ptr(),
+            rank,
+            ws,
+            fused_seg_cap,
+            fused_rank_stride,
+            moe_out,
+            residual,
+            gamma,
+            normed_out,
+            residual_out,
+            eps,
+        )
+        torch.cuda.synchronize()
+
+        # Compare
+        max_diff_res = (residual_out.float() - ref_residual.float()).abs().max().item()
+        max_diff_norm = (normed_out.float() - ref_normed.float()).abs().max().item()
+
+        ok = max_diff_res < 0.125 and max_diff_norm < 0.125
+        if rank == 0:
+            print(
+                f"  N_per_rank={N_per_rank}: {'PASS' if ok else 'FAIL'} "
+                f"(res_diff={max_diff_res:.4f}, norm_diff={max_diff_norm:.4f})"
+            )
+        if not ok:
+            errors += 1
+
+    # ---- Benchmark ----
+    if rank == 0:
+        print(f"\nBenchmark: D={D}, world_size={ws}")
+        print(
+            f"{'config':<10} {'fused':>10} {'fused_g':>10} "
+            f"{'RS+norm':>10} {'RS+norm_g':>10} {'speedup':>8}"
+        )
+        print("-" * 58)
+
+    for N_per_rank in [1, 2, 4, 8]:
+        N_total = N_per_rank * ws
+        input_bytes = N_total * D * 2
+        if input_bytes > fused_rank_stride:
+            if rank == 0:
+                print(f"{N_per_rank}tok      skip (too large)")
+            continue
+
+        moe_out = torch.randn(N_total, D, dtype=torch.bfloat16, device=dev)
+        residual = torch.randn(N_per_rank, D, dtype=torch.bfloat16, device=dev)
+        normed_out = torch.empty_like(residual)
+        residual_out = torch.empty_like(residual)
+        rs_out = torch.empty_like(residual)
+
+        # Fused
+        def run_fused():
+            fused_lib.moe_rs_fused(
+                fused_buf_ptrs.data_ptr(),
+                fused_counters.data_ptr(),
+                rank,
+                ws,
+                fused_seg_cap,
+                fused_rank_stride,
+                moe_out,
+                residual,
+                gamma,
+                normed_out,
+                residual_out,
+                eps,
+            )
+
+        # Separate: RS + add + norm (our Lamport RS + triton-like ops)
+        def run_separate():
+            rs_separate.reduce_scatter(moe_out)
+            # Simulate add + RMSNorm (in practice this is a fused triton kernel)
+            tmp = residual + rs_out
+            torch.rsqrt(tmp.float().pow(2).mean(-1, keepdim=True) + eps)
+
+        fused_us = gpu_timer(run_fused)
+        try:
+            fused_g = gpu_timer_graph(run_fused)
+        except Exception:
+            fused_g = float("nan")
+
+        sep_us = gpu_timer(run_separate)
+        try:
+            sep_g = gpu_timer_graph(run_separate)
+        except Exception:
+            sep_g = float("nan")
+
+        if rank == 0:
+            speedup = sep_g / fused_g if fused_g > 0 else float("nan")
+            print(
+                f"{N_per_rank}tok     {fused_us:>9.1f}µ {fused_g:>9.1f}µ "
+                f"{sep_us:>9.1f}µ {sep_g:>9.1f}µ {speedup:>7.2f}x"
+            )
+
+    dist.barrier()
+    print(f"[rank {rank}] {'PASSED' if errors == 0 else f'FAILED ({errors})'}")
+    dist.destroy_process_group()
+    return errors
+
+
+if __name__ == "__main__":
+    sys.exit(main())

vllm/distributed/device_communicators/all2all.py

@@ -11,14 +11,305 @@ from vllm.distributed import get_dp_group, get_ep_group
 from vllm.forward_context import get_forward_context
 from vllm.logger import init_logger
 from vllm.platforms import current_platform
+from vllm.triton_utils import tl, triton
 from vllm.utils.flashinfer import (
     has_flashinfer_nvlink_one_sided,
     has_flashinfer_nvlink_two_sided,
 )
 from vllm.utils.import_utils import has_deep_ep, has_mori
+from vllm.utils.torch_utils import direct_register_custom_op
 
 from .base_device_communicator import All2AllManagerBase, Cache
 
+logger = init_logger(__name__)
+
+# ---- MoE dispatch custom op (torch.compile-safe) -----------------------
+# Registered as torch.ops.vllm.moe_dispatch so torch.compile treats it as
+# opaque.  The real implementation decides custom kernel vs NCCL at runtime
+# based on tensor sizes.
+
+# Global reference to the MoeAllGather/MoeReduceScatter per group, set lazily.
+_moe_ag_instances: dict[str, Any] = {}
+_moe_rs_instances: dict[str, Any] = {}
+
+
+def _get_ca_comm(group_name: str):
+    """Get the CustomAllreduce communicator for a group, or None."""
+    from vllm.distributed.parallel_state import _groups
+
+    group_ref = _groups.get(group_name)
+    if group_ref is None:
+        return None
+    group = group_ref()
+    if group is None:
+        return None
+    dc = getattr(group, "device_communicator", None)
+    if dc is None:
+        return None
+    ca = getattr(dc, "ca_comm", None)
+    if ca is None or ca.disabled or not getattr(ca, "fully_connected", False):
+        return None
+    return ca
+
+
+def _get_or_create_moe_ag(group_name: str):
+    """Get or lazily create the MoeAllGather for this group."""
+    if group_name in _moe_ag_instances:
+        return _moe_ag_instances[group_name]
+
+    ca = _get_ca_comm(group_name)
+    if ca is None:
+        _moe_ag_instances[group_name] = None
+        return None
+
+    try:
+        from vllm.jit_kernels.moe_all_gather.moe_allgather import (
+            MoeAllGather,
+        )
+
+        ag = MoeAllGather(ca)
+        _moe_ag_instances[group_name] = ag
+        logger.info("MoE custom all-gather initialized for group %s", group_name)
+        return ag
+    except ImportError:
+        _moe_ag_instances[group_name] = None
+        return None
+
+
+def _get_or_create_moe_rs(group_name: str):
+    """Get or lazily create the MoeReduceScatter for this group."""
+    if group_name in _moe_rs_instances:
+        return _moe_rs_instances[group_name]
+
+    ca = _get_ca_comm(group_name)
+    if ca is None:
+        _moe_rs_instances[group_name] = None
+        return None
+
+    try:
+        from vllm.jit_kernels.moe_reduce_scatter.moe_reduce_scatter import (
+            MoeReduceScatter,
+        )
+
+        rs = MoeReduceScatter(ca)
+        _moe_rs_instances[group_name] = rs
+        logger.info("MoE custom reduce-scatter initialized for group %s", group_name)
+        return rs
+    except ImportError:
+        _moe_rs_instances[group_name] = None
+        return None
+
+
+@triton.jit
+def _mask_topk_ids_kernel(
+    topk_ids_ptr,
+    topk_ids_stride,
+    topk,
+    num_actual_tokens_ptr,
+    BLOCK_SIZE: tl.constexpr,
+):
+    token_idx = tl.program_id(0)
+    num_actual_tokens = tl.load(num_actual_tokens_ptr)
+    if token_idx < num_actual_tokens:
+        return
+
+    block = tl.arange(0, BLOCK_SIZE)
+    mask = block < topk
+    tl.store(topk_ids_ptr + token_idx * topk_ids_stride + block, -1, mask=mask)
+
+
+def mask_topk_ids(topk_ids: torch.Tensor, num_actual_tokens: torch.Tensor) -> None:
+    num_tokens, topk = topk_ids.shape
+    _mask_topk_ids_kernel[(num_tokens,)](
+        topk_ids,
+        topk_ids.stride(0),
+        topk,
+        num_actual_tokens,
+        BLOCK_SIZE=triton.next_power_of_2(topk),
+    )
+
+
+def moe_dispatch(
+    tensors: list[torch.Tensor],
+    group_name: str,
+) -> list[torch.Tensor]:
+    """All-gather a list of tensors across the EP/DP group.
+
+    Tries the custom one-shot P2P kernel first; falls back to NCCL
+    all_gatherv if the custom kernel is unavailable or tensors are too large.
+    """
+    from vllm.distributed.parallel_state import _groups
+
+    group_ref = _groups.get(group_name)
+    assert group_ref is not None
+    group = group_ref()
+    assert group is not None
+
+    # Try custom kernel for small, uniform-sized batches.
+    n = len(tensors)
+    if 2 <= n <= 4:
+        forward_context = get_forward_context()
+        num_actual_tokens = getattr(forward_context, "num_actual_tokens", None)
+        if num_actual_tokens is not None:
+            topk_ids = tensors[2]
+            mask_topk_ids(topk_ids, num_actual_tokens)
+
+        moe_ag = _get_or_create_moe_ag(group_name)
+        if moe_ag is not None:
+            ws = moe_ag.world_size
+            # Compute packed size.
+            total_bytes = 0
+            ok = True
+            for t in tensors:
+                if not t.is_contiguous():
+                    ok = False
+                    break
+                nbytes = t.numel() * t.element_size()
+                if nbytes % 16 != 0:
+                    ok = False
+                    break
+                total_bytes = (total_bytes + 15) & ~15
+                total_bytes += nbytes
+
+            max_bytes = moe_ag.max_per_rank
+            if ok and total_bytes <= max_bytes:
+                outputs = [
+                    torch.empty(
+                        (t.shape[0] * ws, *t.shape[1:]),
+                        dtype=t.dtype,
+                        device=t.device,
+                    )
+                    for t in tensors
+                ]
+
+                from vllm.jit_kernels.moe_all_gather.moe_allgather import (
+                    _load_lib,
+                )
+
+                lib = _load_lib()
+                lib.moe_all_gather(
+                    moe_ag._buf_ptrs_ptr,
+                    moe_ag._counters_ptr,
+                    moe_ag.rank,
+                    ws,
+                    moe_ag.seg_capacity,
+                    moe_ag.rank_stride,
+                    tensors,
+                    outputs,
+                )
+                return outputs
+
+    # Fallback: NCCL all_gatherv.
+    return group.device_communicator.all_gatherv(tensors, dim=0)
+
+
+def moe_dispatch_fake(
+    tensors: list[torch.Tensor],
+    group_name: str,
+) -> list[torch.Tensor]:
+    """Fake impl for torch.compile tracing — returns empty tensors of the
+    correct gathered shape."""
+    from vllm.distributed.parallel_state import _groups
+
+    group_ref = _groups.get(group_name)
+    assert group_ref is not None
+    group = group_ref()
+    ws = group.world_size if group is not None else 1
+    return [
+        torch.empty((t.shape[0] * ws, *t.shape[1:]), dtype=t.dtype, device=t.device)
+        for t in tensors
+    ]
+
+
+direct_register_custom_op(
+    op_name="moe_dispatch",
+    op_func=moe_dispatch,
+    fake_impl=moe_dispatch_fake,
+    mutates_args=[],
+)
+
+
+def moe_combine(
+    hidden_states: torch.Tensor,
+    group_name: str,
+) -> torch.Tensor:
+    """Reduce-scatter hidden_states across the EP/DP group.
+
+    Tries the custom Lamport kernel first; falls back to NCCL.
+    """
+    from vllm.distributed.parallel_state import _groups
+
+    group_ref = _groups.get(group_name)
+    assert group_ref is not None
+    group = group_ref()
+    assert group is not None
+
+    ws = group.world_size
+    n_total = hidden_states.shape[0]
+
+    # Try custom kernel for small, uniform, bf16 inputs.
+    if (
+        n_total % ws == 0
+        and hidden_states.is_contiguous()
+        and hidden_states.dtype == torch.bfloat16
+    ):
+        moe_rs = _get_or_create_moe_rs(group_name)
+        if moe_rs is not None:
+            input_bytes = hidden_states.numel() * hidden_states.element_size()
+            if input_bytes % 16 == 0 and input_bytes <= moe_rs.max_per_rank:
+                from vllm.jit_kernels.moe_reduce_scatter.moe_reduce_scatter import (  # noqa: E501
+                    _load_lib,
+                )
+
+                lib = _load_lib()
+                n_per_rank = n_total // ws
+                output = torch.empty(
+                    (n_per_rank, *hidden_states.shape[1:]),
+                    dtype=hidden_states.dtype,
+                    device=hidden_states.device,
+                )
+                lib.moe_reduce_scatter(
+                    moe_rs._buf_ptrs_ptr,
+                    moe_rs._counters_ptr,
+                    moe_rs.rank,
+                    ws,
+                    moe_rs.seg_capacity,
+                    moe_rs.rank_stride,
+                    hidden_states,
+                    output,
+                )
+                return output
+
+    # Fallback: NCCL reduce_scatter.
+    return group.device_communicator.reduce_scatter(hidden_states, dim=0)
+
+
+def moe_combine_fake(
+    hidden_states: torch.Tensor,
+    group_name: str,
+) -> torch.Tensor:
+    """Fake impl for torch.compile tracing."""
+    from vllm.distributed.parallel_state import _groups
+
+    group_ref = _groups.get(group_name)
+    assert group_ref is not None
+    group = group_ref()
+    ws = group.world_size if group is not None else 1
+    n_per_rank = hidden_states.shape[0] // ws
+    return torch.empty(
+        (n_per_rank, *hidden_states.shape[1:]),
+        dtype=hidden_states.dtype,
+        device=hidden_states.device,
+    )
+
+
+direct_register_custom_op(
+    op_name="moe_combine",
+    op_func=moe_combine,
+    fake_impl=moe_combine_fake,
+    mutates_args=[],
+)
+
 if has_flashinfer_nvlink_two_sided():
     from flashinfer.comm import Mapping  # type: ignore[import-not-found]
     from flashinfer.comm.mnnvl import MnnvlConfig  # type: ignore[import-not-found]
@@ -155,6 +446,132 @@ class AgRsAll2AllManager(All2AllManagerBase):
 
     def __init__(self, cpu_group, tcp_store_group=None):
         super().__init__(cpu_group, tcp_store_group)
+        self._moe_ag = None  # lazy: MoeAllGather instance
+
+    def _get_moe_allgather(self, dist_group):
+        """Lazily build the standalone MoE all-gather kernel wrapper."""
+        if self._moe_ag is not None:
+            return self._moe_ag
+
+        import logging
+
+        log = logging.getLogger(__name__)
+
+        # Check prerequisites: intra-node, NVLink, ca_comm available.
+        if self.internode:
+            log.warning("MoE AG: skipped (internode)")
+            return None
+        dc = getattr(dist_group, "device_communicator", None)
+        if dc is None:
+            log.warning("MoE AG: skipped (no device_communicator)")
+            return None
+        ca = getattr(dc, "ca_comm", None)
+        if ca is None:
+            log.warning("MoE AG: skipped (ca_comm is None)")
+            return None
+        if ca.disabled:
+            log.warning("MoE AG: skipped (ca_comm disabled)")
+            return None
+        if not getattr(ca, "fully_connected", False):
+            log.warning("MoE AG: skipped (not fully_connected)")
+            return None
+
+        try:
+            from vllm.jit_kernels.moe_all_gather.moe_allgather import (  # type: ignore
+                MoeAllGather,
+            )
+        except ImportError as e:
+            log.warning("MoE AG: skipped (import failed: %s)", e)
+            return None
+
+        self._moe_ag = MoeAllGather(ca)
+        log.info("MoE AG: custom kernel initialized successfully")
+        return self._moe_ag
+
+    # ---- helpers --------------------------------------------------------
+
+    def _custom_all_gather(
+        self,
+        dist_group,
+        tensors: list[torch.Tensor],
+        sizes: list[int],
+    ) -> list[torch.Tensor] | None:
+        """Try the one-shot P2P MoE all-gather kernel.  Returns None on
+        fallback (kernel unavailable, too many tensors, etc.)."""
+        import logging
+
+        _log = logging.getLogger(__name__)
+        _dbg = getattr(self, "_ag_dbg_count", 0)
+
+        n = len(tensors)
+        if n < 2 or n > 4:
+            if _dbg < 10:
+                _log.warning(f"MoE AG: n={n} (need 2-4)")
+                self._ag_dbg_count = _dbg + 1
+            return None
+        if any(s != sizes[0] for s in sizes[1:]):
+            if _dbg < 10:
+                _log.warning(f"MoE AG: non-uniform sizes {sizes}")
+                self._ag_dbg_count = _dbg + 1
+            return None
+        moe_ag = self._get_moe_allgather(dist_group)
+        if moe_ag is None:
+            return None
+
+        ws = moe_ag.world_size
+
+        total_bytes = 0
+        for t in tensors:
+            if not t.is_contiguous():
+                if _dbg < 10:
+                    _log.warning(f"MoE AG: non-contiguous {t.shape} {t.stride()}")
+                    self._ag_dbg_count = _dbg + 1
+                return None
+            nbytes = t.numel() * t.element_size()
+            if nbytes % 16 != 0:
+                if _dbg < 10:
+                    _log.warning(
+                        f"MoE AG: unaligned {nbytes}B shape={t.shape} dtype={t.dtype}"
+                    )
+                    self._ag_dbg_count = _dbg + 1
+                return None
+            total_bytes = (total_bytes + 15) & ~15
+            total_bytes += nbytes
+
+        if total_bytes > moe_ag.max_per_rank:
+            if _dbg < 10:
+                _log.warning(
+                    f"MoE AG: too large {total_bytes} > {moe_ag.max_per_rank} "
+                    f"n={n} shapes={[t.shape for t in tensors]}"
+                )
+                self._ag_dbg_count = _dbg + 1
+            return None
+
+        # Allocate per-tensor outputs.
+        outputs = [
+            torch.empty((t.shape[0] * ws, *t.shape[1:]), dtype=t.dtype, device=t.device)
+            for t in tensors
+        ]
+
+        from vllm.jit_kernels.moe_all_gather.moe_allgather import (  # type: ignore
+            _load_lib,
+        )
+
+        lib = _load_lib()
+        lib.moe_all_gather(
+            moe_ag._buf_ptrs_ptr,
+            moe_ag._counters_ptr,
+            moe_ag.rank,
+            ws,
+            moe_ag.seg_capacity,
+            moe_ag.rank_stride,
+            tensors,
+            outputs,
+        )
+
+        return outputs
+
+    # ---- public API -----------------------------------------------------
 
     def dispatch_router_logits(
         self,
@@ -176,19 +593,22 @@ class AgRsAll2AllManager(All2AllManagerBase):
         dist_group = get_ep_group() if is_sequence_parallel else get_dp_group()
         assert sizes[dist_group.rank_in_group] == hidden_states.shape[0]
 
-        tensors_to_gather = [hidden_states, router_logits]
+        tensors = [hidden_states, router_logits]
         if extra_tensors is not None:
-            tensors_to_gather.extend(extra_tensors)
+            tensors.extend(extra_tensors)
 
-        gathered_tensors = dist_group.all_gatherv(
-            tensors_to_gather,
-            dim=0,
-            sizes=sizes,
-        )
+        # Use custom op so torch.compile treats this as opaque.
+        # Runtime decides custom kernel vs NCCL based on tensor sizes.
+        if all(s == sizes[0] for s in sizes):
+            gathered = torch.ops.vllm.moe_dispatch(
+                tensors, group_name=dist_group.unique_name
+            )
+        else:
+            gathered = dist_group.all_gatherv(tensors, dim=0, sizes=sizes)
 
         if extra_tensors is not None:
-            return (gathered_tensors[0], gathered_tensors[1], gathered_tensors[2:])
-        return gathered_tensors[0], gathered_tensors[1]
+            return (gathered[0], gathered[1], gathered[2:])
+        return gathered[0], gathered[1]
 
     def dispatch(
         self,
@@ -202,7 +622,7 @@ class AgRsAll2AllManager(All2AllManagerBase):
         | tuple[torch.Tensor, torch.Tensor, torch.Tensor, list[torch.Tensor]]
     ):
         """
-        Gather hidden_states and router_logits from all dp ranks.
+        Gather hidden_states, topk_weights, topk_ids from all dp ranks.
         """
         dp_metadata = get_forward_context().dp_metadata
         assert dp_metadata is not None
@@ -211,24 +631,24 @@ class AgRsAll2AllManager(All2AllManagerBase):
         dist_group = get_ep_group() if is_sequence_parallel else get_dp_group()
         assert sizes[dist_group.rank_in_group] == hidden_states.shape[0]
 
-        tensors_to_gather = [hidden_states, topk_weights, topk_ids]
+        tensors = [hidden_states, topk_weights, topk_ids]
         if extra_tensors is not None:
-            tensors_to_gather.extend(extra_tensors)
+            tensors.extend(extra_tensors)
 
-        gathered_tensors = dist_group.all_gatherv(
-            tensors_to_gather,
-            dim=0,
-            sizes=sizes,
-        )
+        if all(s == sizes[0] for s in sizes):
+            gathered = torch.ops.vllm.moe_dispatch(
+                tensors, group_name=dist_group.unique_name
+            )
+        else:
+            gathered = dist_group.all_gatherv(tensors, dim=0, sizes=sizes)
 
-        hidden_states = gathered_tensors[0]
-        topk_weights = gathered_tensors[1]
-        topk_ids = gathered_tensors[2]
+        hidden_states = gathered[0]
+        topk_weights = gathered[1]
+        topk_ids = gathered[2]
 
         if extra_tensors is None:
             return hidden_states, topk_weights, topk_ids
-
-        return hidden_states, topk_weights, topk_ids, gathered_tensors[3:]
+        return hidden_states, topk_weights, topk_ids, gathered[3:]
 
     def combine(
         self, hidden_states: torch.Tensor, is_sequence_parallel: bool = False
@@ -242,11 +662,21 @@ class AgRsAll2AllManager(All2AllManagerBase):
         assert sizes is not None
 
         dist_group = get_ep_group() if is_sequence_parallel else get_dp_group()
-        hidden_states = dist_group.reduce_scatterv(hidden_states, dim=0, sizes=sizes)
+
+        # Use custom op so torch.compile treats this as opaque.
+        if all(s == sizes[0] for s in sizes):
+            hidden_states = torch.ops.vllm.moe_combine(
+                hidden_states, group_name=dist_group.unique_name
+            )
+        else:
+            hidden_states = dist_group.reduce_scatterv(
+                hidden_states, dim=0, sizes=sizes
+            )
         return hidden_states
 
     def destroy(self):
-        pass
+        if self._moe_ag is not None:
+            self._moe_ag = None
 
 
 class DeepEPAll2AllManagerBase(All2AllManagerBase):

vllm/distributed/device_communicators/cuda_communicator.py

@@ -41,8 +41,7 @@ class CudaCommunicator(DeviceCommunicatorBase):
             global_ranks,
             global_world_size,
         )
-        if "tp" not in unique_name:
-            # custom allreduce or torch symm mem can be used only by tp
+        if not any(tag in unique_name for tag in ("tp", "dp", "ep")):
             use_custom_allreduce = False
             use_torch_symm_mem = False
             use_flashinfer_allreduce = False
@@ -50,8 +49,12 @@ class CudaCommunicator(DeviceCommunicatorBase):
             from vllm.distributed.parallel_state import _ENABLE_CUSTOM_ALL_REDUCE
 
             use_custom_allreduce = _ENABLE_CUSTOM_ALL_REDUCE
-            use_torch_symm_mem = envs.VLLM_ALLREDUCE_USE_SYMM_MEM
-            use_flashinfer_allreduce = envs.VLLM_ALLREDUCE_USE_FLASHINFER
+            use_torch_symm_mem = (
+                "tp" in unique_name and envs.VLLM_ALLREDUCE_USE_SYMM_MEM
+            )
+            use_flashinfer_allreduce = (
+                "tp" in unique_name and envs.VLLM_ALLREDUCE_USE_FLASHINFER
+            )
 
         self.use_custom_allreduce = use_custom_allreduce
         self.use_torch_symm_mem = use_torch_symm_mem
@@ -238,8 +241,6 @@ class CudaCommunicator(DeviceCommunicatorBase):
 
     def reduce_scatter(self, input_: torch.Tensor, dim: int = -1):
         world_size = self.world_size
-        pynccl_comm = self.pynccl_comm
-        assert pynccl_comm is not None
         if dim < 0:
             # Convert negative dim to positive.
             dim += input_.dim()
@@ -256,6 +257,8 @@ class CudaCommunicator(DeviceCommunicatorBase):
             output_shape, dtype=input_tensor.dtype, device=input_tensor.device
         )
 
+        pynccl_comm = self.pynccl_comm
+        assert pynccl_comm is not None
         pynccl_comm.reduce_scatter(output, input_tensor)
 
         # Reshape before returning

vllm/forward_context.py

@@ -310,6 +310,7 @@ def override_forward_context(forward_context: ForwardContext | None):
 def set_forward_context(
     attn_metadata: Any,
     vllm_config: VllmConfig,
+    num_actual_tokens: torch.Tensor | None = None,
     num_tokens: int | None = None,
     num_tokens_across_dp: torch.Tensor | None = None,
     cudagraph_runtime_mode: CUDAGraphMode = CUDAGraphMode.NONE,
@@ -377,6 +378,8 @@ def set_forward_context(
         additional_kwargs,
         skip_compiled,
     )
+    if num_actual_tokens is not None:
+        forward_context.num_actual_tokens = num_actual_tokens
 
     try:
         with (

vllm/jit_kernels/moe_all_gather/moe_allgather.cu

@@ -0,0 +1 @@
+/home/woosuk/workspace/vllm/csrc/standalone/custom_ar/moe_allgather.cu

vllm/jit_kernels/moe_all_gather/moe_allgather.py

@@ -0,0 +1 @@
+/home/woosuk/workspace/vllm/csrc/standalone/custom_ar/moe_allgather.py

vllm/jit_kernels/moe_reduce_scatter/moe_reduce_scatter.cu

@@ -0,0 +1 @@
+/home/woosuk/workspace/vllm/csrc/standalone/custom_ar/moe_reduce_scatter.cu

vllm/jit_kernels/moe_reduce_scatter/moe_reduce_scatter.py

@@ -0,0 +1 @@
+/home/woosuk/workspace/vllm/csrc/standalone/custom_ar/moe_reduce_scatter.py

vllm/jit_kernels/moe_reduce_scatter/moe_rs_fused.cu

@@ -0,0 +1 @@
+/home/woosuk/workspace/vllm/csrc/standalone/custom_ar/moe_rs_fused.cu

vllm/jit_kernels/moe_reduce_scatter/moe_rs_fused.py

@@ -0,0 +1 @@
+/home/woosuk/workspace/vllm/csrc/standalone/custom_ar/moe_rs_fused.py

vllm/v1/worker/gpu/cudagraph_utils.py

@@ -335,6 +335,7 @@ class ModelCudaGraphManager(CudaGraphManager):
                 with set_forward_context(
                     attn_metadata if cg_mode != CUDAGraphMode.PIECEWISE else None,
                     self.vllm_config,
+                    num_actual_tokens=input_buffers.num_actual_tokens,
                     num_tokens=num_tokens,
                     cudagraph_runtime_mode=cg_mode,
                     num_tokens_across_dp=num_tokens_across_dp,

vllm/v1/worker/gpu/input_batch.py

@@ -30,6 +30,7 @@ class InputBuffers:
         self.dcp_local_seq_lens = torch.zeros(
             max_num_reqs, dtype=torch.int32, device=device
         )
+        self.num_actual_tokens = torch.zeros(1, dtype=torch.int32, device=device)
 
 
 @dataclass

vllm/v1/worker/gpu/model_runner.py

@@ -971,6 +971,7 @@ class GPUModelRunner(LoRAModelRunnerMixin):
             empty_output = self.kv_connector.no_forward(scheduler_output)
             return empty_output
 
+        self.input_buffers.num_actual_tokens[:1] = num_toks if not dummy_run else 0
         if not dummy_run:
             # Common case.
             # Prepare all the inputs and copy to the input buffers.
@@ -1075,6 +1076,7 @@ class GPUModelRunner(LoRAModelRunnerMixin):
             with set_forward_context(
                 attn_metadata,
                 self.vllm_config,
+                num_actual_tokens=self.input_buffers.num_actual_tokens,
                 num_tokens=input_batch.num_tokens_after_padding,
                 cudagraph_runtime_mode=batch_desc.cg_mode,
                 num_tokens_across_dp=num_tokens_across_dp,