[EPLB] Reject NCCL-based EPLB communicators with async EPLB (#44978)

Signed-off-by: Markov Ilya <[email protected]>

tests/distributed/test_elastic_ep.py

@@ -78,6 +78,8 @@ def test_elastic_ep_scaling():
         "--enable-eplb",
         "--eplb-config.num_redundant_experts",
         "0",
+        "--eplb-config.use_async",
+        "false",
         "--data-parallel-backend",
         "ray",
         "--data-parallel-size",
@@ -151,6 +153,8 @@ def test_elastic_ep_scaling_uneven():
         "--enable-eplb",
         "--eplb-config.num_redundant_experts",
         "0",
+        "--eplb-config.use_async",
+        "false",
         "--data-parallel-backend",
         "ray",
         "--data-parallel-size",

tests/distributed/test_eplb_execute.py

@@ -644,9 +644,7 @@ def _test_rearrange_expert_weights_no_change(env, world_size) -> None:
         (2, 2, 2, 3),
     ],
 )
-@pytest.mark.parametrize(
-    "eplb_communicator", ["torch_nccl", "torch_gloo", "pynccl", "nixl"]
-)
+@pytest.mark.parametrize("eplb_communicator", ["torch_gloo", "nixl"])
 def test_async_transfer_layer_without_mtp(
     world_size: int,
     num_layers: int,

tests/kernels/moe/test_moe_layer.py

@@ -1285,6 +1285,9 @@ def _test_body_eplb(
     expert_weights = [list(eplb_moe_layer.get_expert_weights())]
 
     expert_buffer = [torch.empty_like(w) for w in expert_weights[0]]
+    assert vllm_config.parallel_config.eplb_config.communicator is not None, (
+        "EPLB communicator backend must be set by ParallelConfig"
+    )
     communicator = create_eplb_communicator(
         group_coordinator=get_eplb_group(),
         backend=vllm_config.parallel_config.eplb_config.communicator,

vllm/config/parallel.py

@@ -94,13 +94,20 @@ class EPLBConfig:
     - "torch_gloo": Use torch.distributed gloo with CPU staging
     - "nixl": Use NIXL/ RIXL with staged send/recv buffers
     - "pynccl": Use PyNccl send/recv
-    - None: Auto-select backend ("torch_gloo" for async, "torch_nccl" for sync)
+    - None: Auto-select backend (prefers "nixl", falls back to "torch_gloo")
     """
 
     @model_validator(mode="after")
     def _validate_eplb_config(self) -> Self:
         if self.use_async and self.policy != "default":
             raise ValueError("Async EPLB is only supported with the default policy.")
+        if self.use_async and self.communicator in ("torch_nccl", "pynccl"):
+            raise ValueError(
+                f"{self.communicator} communicator is incompatible with "
+                "async EPLB due to NCCL multi-stream conflicts. Use "
+                "'torch_gloo' or 'nixl' instead, or leave communicator "
+                "unset for automatic selection."
+            )
         if self.log_balancedness and self.log_balancedness_interval <= 0:
             raise ValueError("log_balancedness_interval must be greater than 0.")
         return self
@@ -787,6 +794,13 @@ class ParallelConfig:
         if self.enable_elastic_ep:
             if not self.enable_eplb:
                 raise ValueError("Elastic EP is only supported with enable_eplb=True.")
+            if self.eplb_config.use_async:
+                raise ValueError(
+                    "Elastic EP requires the pynccl communicator, which is "
+                    "incompatible with async EPLB due to NCCL multi-stream "
+                    "conflicts. Disable async EPLB (eplb_config.use_async=False) "
+                    "to use elastic EP."
+                )
             if self.pipeline_parallel_size > 1:
                 raise ValueError(
                     "Elastic EP is not supported with pipeline parallelism "

vllm/distributed/elastic_ep/elastic_execute.py

@@ -470,6 +470,9 @@ class ElasticEPScalingExecutor:
         eplb_model_state.expert_buffer = [
             torch.empty_like(w) for w in model.expert_weights[0]
         ]
+        assert parallel_config.eplb_config.communicator is not None, (
+            "EPLB communicator backend must be set by ParallelConfig"
+        )
         eplb_model_state.communicator = create_eplb_communicator(
             group_coordinator=get_eplb_group(),
             backend=parallel_config.eplb_config.communicator,

vllm/distributed/eplb/eplb_communicator.py

@@ -617,7 +617,7 @@ class PyNcclEplbCommunicator(EplbCommunicator):
 
 def create_eplb_communicator(
     group_coordinator: GroupCoordinator,
-    backend: str | None,
+    backend: str,
     expert_weights: Sequence[Sequence[torch.Tensor]],
     expert_buffer: Sequence[torch.Tensor],
 ) -> EplbCommunicator:
@@ -628,7 +628,6 @@ def create_eplb_communicator(
             device and CPU communication groups.
         backend: Communicator backend name (``"torch_nccl"``,
             ``"torch_gloo"``, ``"pynccl"``, or ``"nixl"``).
-            Falls back to ``"torch_nccl"`` when *None*.
             Stateless (elastic EP) groups only support ``"torch_nccl"``
             and ``"pynccl"``; ``"torch_nccl"`` is silently promoted to
             ``"pynccl"`` in that case.  When tensors reside on CPU,
@@ -641,9 +640,6 @@ def create_eplb_communicator(
         expert_buffer: Pre-allocated receive buffer tensors (one per
             weight tensor in a single layer).
     """
-    if backend is None:
-        backend = "torch_nccl"
-
     first_layer = expert_weights[0] if expert_weights else []
     tensor_device_type = first_layer[0].device.type if first_layer else "cpu"
     torch_group = (

vllm/distributed/eplb/eplb_state.py

@@ -447,6 +447,9 @@ class EplbState:
         self._init_should_record_tensor(model)
         expert_buffer = [torch.empty_like(w) for w in model.expert_weights[0]]
 
+        assert self.parallel_config.eplb_config.communicator is not None, (
+            "EPLB communicator backend must be set by ParallelConfig"
+        )
         communicator = create_eplb_communicator(
             group_coordinator=get_eplb_group(),
             backend=self.parallel_config.eplb_config.communicator,

vllm/distributed/eplb/eplb_utils.py

@@ -74,8 +74,6 @@ def override_envs_for_eplb(
     """
     is_data_parallel = parallel_config.data_parallel_size > 1
     is_eplb_enabled = parallel_config.enable_eplb
-    async_eplb = parallel_config.eplb_config.use_async
-    is_deepep_ll = parallel_config.all2all_backend == "deepep_low_latency"
     is_mega_moe = moe_backend == "deep_gemm_mega_moe"
     is_nccl_based_eplb_communicator = parallel_config.eplb_config.communicator in (
         "torch_nccl",
@@ -85,29 +83,16 @@ def override_envs_for_eplb(
     # Override NCCL_MAX_CTAS to avoid hangs when EPLB's NCCL weight exchange
     # contends with MoE backend's cooperative-launch on GPU SMs.
     #
-    # DeepEP low-latency:
-    # The hang happens when two ranks interleave kernel launches differently
-    # between NCCL collectives (used by async EPLB weight exchange) and DeepEP
-    # low-latency (LL) kernels. DeepEP LL uses a cooperative launch and tries
-    # to reserve a large fraction of the GPU's SMs; if those SMs are currently
-    # occupied by NCCL, the DeepEP LL launch blocks until enough SMs are
-    # freed.
-    #
-    # If rank A enters DeepEP LL in main thread while rank B is still executing
-    # NCCL in async thread, rank A can block waiting for SMs, while rank B can
-    # block inside NCCL waiting for rank A to participate in the collective.
-    # This circular wait causes a deadlock.
-    # Limiting NCCL occupancy via NCCL_MAX_CTAS leaves space for the DeepEP
-    # cooperative kernel to launch and complete, breaking the deadlock.
-    # See: https://github.com/deepseek-ai/DeepEP/issues/496
-    #
-    # DeepGEMM Mega MoE also uses cooperative launch and will cause hang even
-    # with sync EPLB.
+    # DeepGEMM Mega MoE uses cooperative launch, which tries to reserve a
+    # large fraction of the GPU's SMs. If those SMs are occupied by NCCL,
+    # the cooperative launch blocks until enough SMs are freed, causing a
+    # deadlock. Limiting NCCL occupancy via NCCL_MAX_CTAS leaves space for
+    # the cooperative kernel to launch and complete.
     if (
         is_data_parallel
         and is_eplb_enabled
         and is_nccl_based_eplb_communicator
-        and ((is_deepep_ll and async_eplb) or is_mega_moe)
+        and is_mega_moe
     ):
         current_value_str = os.getenv("NCCL_MAX_CTAS")
 
@@ -116,10 +101,9 @@ def override_envs_for_eplb(
 
         override_value = 8
         os.environ["NCCL_MAX_CTAS"] = str(override_value)
-        backend = "deepep_low_latency" if is_deepep_ll else "deep_gemm_mega_moe"
         logger.info_once(
             f"EPLB: Setting NCCL_MAX_CTAS={override_value} "
             f"for expert parallel with NCCL-based EPLB communicator and "
-            f"cooperative MoE backend ({backend})",
+            f"cooperative MoE backend (deep_gemm_mega_moe)",
             scope="global",
         )