Revert "[WideEP] Integrate DeepEP v2" (#41183)

.buildkite/test_areas/kernels.yaml

@@ -299,4 +299,3 @@ steps:
   - vllm/config
   commands:
     - pytest -v -s kernels/moe/test_moe_layer.py
-    - pytest -v -s kernels/moe/test_deepep_v2_moe.py

tests/distributed/test_mnnvl_alltoall.py

@@ -19,7 +19,6 @@
     has_flashinfer_nvlink_one_sided,
     has_flashinfer_nvlink_two_sided,
 )
-from vllm.utils.import_utils import has_deep_ep_v2
 from vllm.utils.network_utils import get_open_port
 
 from ..utils import init_test_distributed_environment
@@ -195,10 +194,6 @@ class _AttnMeta:
     not _has_sys_ptrace(),
     reason="SYS_PTRACE required (docker run --cap-add=SYS_PTRACE)",
 )
-requires_deep_ep_v2 = pytest.mark.skipif(
-    not has_deep_ep_v2(),
-    reason="DeepEP v2 (ElasticBuffer) not available or NCCL < 2.30.4",
-)
 
 # NOTE: No module-level pytestmark here. The FlashInfer lifecycle tests have
 # their own @requires_two_sided / @requires_one_sided decorators, and
@@ -861,76 +856,3 @@ def _one_sided_data_worker(rank, world_size):
 def test_one_sided_dispatch_combine(world_size):
     """Test FlashInfer one-sided dispatch/combine with actual data flow."""
     _spawn_workers(_one_sided_data_worker, world_size, dp_size=world_size)
-
-
-# ---------------------------------------------------------------------------
-# Test 6: DeepEP v2 (ElasticBuffer) manager lifecycle
-# ---------------------------------------------------------------------------
-#
-# Tests DeepEPV2All2AllManager which wraps DeepEP's ElasticBuffer API using
-# the NCCL GIN backend. Requires DeepEP >= 2.0 and NCCL >= 2.30.4.
-#
-# Uses EP group because the DeepEP v2 manager is constructed with an
-# EP-scoped communicator in production. With tp=world_size the EP group
-# spans all ranks.
-# ---------------------------------------------------------------------------
-
-
-def _deepep_v2_lifecycle_worker(rank, world_size):
-    from vllm.distributed.device_communicators.all2all import (
-        DeepEPV2All2AllManager,
-    )
-
-    cpu_group = get_ep_group().cpu_group
-    manager = DeepEPV2All2AllManager(cpu_group)
-
-    assert manager.rank == rank
-    assert manager.world_size == world_size
-    assert manager._num_sms is None
-
-    hidden_size = 7168
-    num_experts = world_size * 32
-    num_topk = 8
-    max_tokens = 256
-
-    handle_kwargs = dict(
-        num_max_tokens_per_rank=max_tokens,
-        hidden=hidden_size,
-        num_topk=num_topk,
-        num_experts=num_experts,
-        use_fp8_dispatch=False,
-    )
-
-    handle = manager.get_handle(handle_kwargs)
-    assert handle is not None
-    assert manager._num_sms is not None
-    assert manager._num_sms > 0
-
-    torch.distributed.barrier()
-
-    # get_handle again with same args should return cached handle
-    handle2 = manager.get_handle(dict(handle_kwargs))
-    assert handle2 is handle
-
-    torch.distributed.barrier()
-
-    # Destroy clears the cache
-    manager.destroy()
-    assert len(manager.handle_cache._cache) == 0
-
-    torch.distributed.barrier()
-
-    # Re-create after destroy
-    handle3 = manager.get_handle(dict(handle_kwargs))
-    assert handle3 is not None
-
-    torch.distributed.barrier()
-    manager.destroy()
-
-
-@requires_multi_gpu
-@requires_deep_ep_v2
-@pytest.mark.parametrize("world_size", [2])
-def test_deepep_v2_manager_lifecycle(world_size):
-    """Test DeepEP v2 ElasticBuffer manager init, caching, and destroy."""
-    _spawn_workers(_deepep_v2_lifecycle_worker, world_size)

tests/kernels/moe/modular_kernel_tools/common.py

@@ -1,7 +1,6 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 from dataclasses import dataclass
-from types import SimpleNamespace
 from typing import Any
 
 import torch
@@ -44,7 +43,6 @@
 from vllm.utils.import_utils import (
     has_aiter,
     has_deep_ep,
-    has_deep_ep_v2,
     has_deep_gemm,
     has_mori,
 )
@@ -152,10 +150,6 @@ def make_env_data(self) -> tuple[VllmConfig, dict[Any, Any]]:
         make env data for vllm launch.
         """
         vllm_config = VllmConfig()
-        vllm_config.model_config = SimpleNamespace(
-            enforce_eager=True,
-            is_moe=True,
-        )
         vllm_config.parallel_config.data_parallel_size = self.world_size
         vllm_config.parallel_config.enable_expert_parallel = True
 
@@ -245,10 +239,6 @@ def needs_deep_ep(self):
             or info.backend == "deepep_low_latency"
         )
 
-    def needs_deep_ep_v2(self):
-        info = prepare_finalize_info(self.prepare_finalize_type)
-        return info.backend == "deepep_v2"
-
     def needs_aiter(self):
         info = expert_info(self.fused_experts_type)
         return info.needs_aiter
@@ -325,8 +315,6 @@ def is_valid(self) -> tuple[bool, str | None]:
         # Check dependencies (turn into asserts?)
         if self.needs_deep_ep() and not has_deep_ep():
             return False, "Needs DeepEP, but DeepEP not available."
-        if self.needs_deep_ep_v2() and not has_deep_ep_v2():
-            return False, "Needs DeepEP v2, but DeepEP v2 not available."
         if self.needs_deep_gemm() and not has_deep_gemm():
             return False, "Needs DeepGEMM, but DeepGEMM not available."
         if self.needs_aiter() and not has_aiter():  # noqa: SIM103
@@ -669,58 +657,6 @@ def make_modular_kernel(
     return modular_kernel
 
 
-def _maybe_convert_weights_for_experts(
-    config: Config,
-    rank_weights: WeightTensors,
-) -> WeightTensors:
-    """Convert weights to expert-specific format (e.g., TrtLLM BlockMajorK)."""
-    from vllm.model_executor.layers.fused_moe.oracle.fp8 import (
-        Fp8MoeBackend,
-        convert_to_fp8_moe_kernel_format,
-    )
-
-    fe_type = config.fused_experts_type
-    fe_name = getattr(fe_type, "__name__", "")
-
-    backend: Fp8MoeBackend | None = None
-    if fe_name == "TrtLlmFp8ExpertsModular":
-        backend = Fp8MoeBackend.FLASHINFER_TRTLLM
-    elif fe_name == "FlashInferExperts":
-        backend = Fp8MoeBackend.FLASHINFER_CUTLASS
-
-    if backend is None or not rank_weights.is_quantized():
-        return rank_weights
-
-    mock_layer = SimpleNamespace(
-        weight_block_size=config.quant_block_shape,
-        moe_config=SimpleNamespace(
-            is_act_and_mul=True,
-            intermediate_size_per_partition=config.N,
-        ),
-        activation=SimpleNamespace(is_gated=True),
-    )
-
-    w1, w2, w1_scale, w2_scale = convert_to_fp8_moe_kernel_format(
-        fp8_backend=backend,
-        layer=mock_layer,
-        w13=rank_weights.w1,
-        w2=rank_weights.w2,
-        w13_scale=rank_weights.w1_scale,
-        w2_scale=rank_weights.w2_scale,
-        w13_input_scale=None,
-        w2_input_scale=None,
-    )
-
-    return WeightTensors(
-        w1=w1,
-        w2=w2,
-        w1_scale=w1_scale,
-        w2_scale=w2_scale,
-        w1_gs=rank_weights.w1_gs,
-        w2_gs=rank_weights.w2_gs,
-    )
-
-
 def run_modular_kernel(
     pgi: ProcessGroupInfo,
     vllm_config: VllmConfig,
@@ -733,7 +669,6 @@ def run_modular_kernel(
 
     # weights for rank
     rank_weights = weights.slice_weights(pgi.rank, config.num_local_experts)
-    rank_weights = _maybe_convert_weights_for_experts(config, rank_weights)
 
     if config.quant_dtype == "nvfp4":
         gscale = _make_gscale(config.num_local_experts)
@@ -779,8 +714,6 @@ def run_modular_kernel(
         [num_tokens] * config.world_size, device="cuda", dtype=torch.int
     )
 
-    torch.distributed.barrier()
-
     with set_forward_context(
         None,
         vllm_config,

tests/kernels/moe/modular_kernel_tools/mk_objects.py

@@ -36,12 +36,10 @@
 from vllm.utils.flashinfer import (
     has_flashinfer_cutlass_fused_moe,
     has_flashinfer_nvlink_one_sided,
-    has_flashinfer_trtllm_fused_moe,
 )
 from vllm.utils.import_utils import (
     has_aiter,
     has_deep_ep,
-    has_deep_ep_v2,
     has_deep_gemm,
     has_mori,
 )
@@ -218,19 +216,6 @@ def expert_info(kind) -> ExpertInfo:
         backend="deepep_low_latency",
     )
 
-if has_deep_ep_v2() and current_platform.has_device_capability(100):
-    from vllm.model_executor.layers.fused_moe.prepare_finalize.deepep_v2 import (
-        DeepEPV2PrepareAndFinalize,
-    )
-
-    register_prepare_and_finalize(
-        DeepEPV2PrepareAndFinalize,
-        standard_format,
-        common_float_types,
-        blocked_quantization_support=True,
-        backend="deepep_v2",
-    )
-
 if has_mori():
     from vllm.model_executor.layers.fused_moe.prepare_finalize.mori import (
         MoriPrepareAndFinalize,
@@ -304,18 +289,6 @@ def expert_info(kind) -> ExpertInfo:
         blocked_quantization_support=False,
     )
 
-if has_flashinfer_trtllm_fused_moe() and current_platform.has_device_capability(100):
-    from vllm.model_executor.layers.fused_moe.experts.trtllm_fp8_moe import (
-        TrtLlmFp8ExpertsModular,
-    )
-
-    register_experts(
-        TrtLlmFp8ExpertsModular,
-        standard_format,
-        fp8_types,
-        blocked_quantization_support=True,
-    )
-
 if has_aiter():
     from vllm.model_executor.layers.fused_moe.experts.rocm_aiter_moe import (
         AiterExperts,

tests/kernels/moe/modular_kernel_tools/parallel_utils.py

@@ -106,7 +106,7 @@ def _worker_parallel_launch(
     if vllm_config is not None:
         cpu_group = _set_vllm_config(vllm_config, world_size, rank, local_rank)
 
-    def _run_worker():
+    try:
         worker(
             ProcessGroupInfo(
                 world_size=world_size,
@@ -121,13 +121,6 @@ def _run_worker():
             *args,
             **worker_kwargs,
         )
-
-    try:
-        if vllm_config is not None:
-            with set_current_vllm_config(vllm_config):
-                _run_worker()
-        else:
-            _run_worker()
     except Exception as ex:
         print(ex)
         traceback.print_exc()

tests/kernels/moe/parallel_utils.py

@@ -15,7 +15,7 @@
 from torch.multiprocessing import spawn  # pyright: ignore[reportPrivateImportUsage]
 from typing_extensions import ParamSpec
 
-from vllm.utils.import_utils import has_deep_ep, has_deep_ep_v2
+from vllm.utils.import_utils import has_deep_ep
 from vllm.utils.network_utils import get_open_port
 
 if has_deep_ep():
@@ -26,11 +26,6 @@
         DeepEPLLPrepareAndFinalize,
     )
 
-if has_deep_ep_v2():
-    from vllm.model_executor.layers.fused_moe.prepare_finalize.deepep_v2 import (
-        DeepEPV2PrepareAndFinalize,
-    )
-
 ## Parallel Processes Utils
 
 P = ParamSpec("P")
@@ -60,10 +55,11 @@ def _worker_parallel_launch(
     torch.accelerator.set_device_index(local_rank)
     device = torch.device("cuda", local_rank)
     torch.distributed.init_process_group(
-        backend="nccl",
+        backend="cpu:gloo,cuda:nccl",
         init_method=init_method,
         rank=rank,
         world_size=world_size,
+        device_id=device,
     )
     barrier = torch.tensor([rank], device=device)
     torch.distributed.all_reduce(barrier)
@@ -204,42 +200,3 @@ def make_deepep_a2a(
 
     assert deepep_ll_args is not None
     return make_deepep_ll_a2a(pg, pgi, deepep_ll_args, q_dtype, block_shape)
-
-
-@dataclasses.dataclass
-class DeepEPV2Args:
-    num_local_experts: int
-    num_experts: int
-    num_topk: int
-    hidden_size: int
-    max_tokens_per_rank: int
-    use_fp8_dispatch: bool
-
-
-def make_deepep_v2_a2a(
-    pg: ProcessGroup,
-    pgi: ProcessGroupInfo,
-    dp_size: int,
-    v2_args: DeepEPV2Args,
-    use_cudagraph: bool = False,
-):
-    import deep_ep
-
-    buffer = deep_ep.ElasticBuffer(
-        group=pg,
-        num_max_tokens_per_rank=v2_args.max_tokens_per_rank,
-        hidden=v2_args.hidden_size,
-        num_topk=v2_args.num_topk,
-        use_fp8_dispatch=v2_args.use_fp8_dispatch,
-        explicitly_destroy=True,
-    )
-    return DeepEPV2PrepareAndFinalize(
-        buffer=buffer,
-        num_dispatchers=pgi.world_size,
-        dp_size=dp_size,
-        rank_expert_offset=pgi.rank * v2_args.num_local_experts,
-        num_experts=v2_args.num_experts,
-        num_topk=v2_args.num_topk,
-        use_fp8_dispatch=v2_args.use_fp8_dispatch,
-        use_cudagraph=use_cudagraph,
-    )