diff --git a/tests/distributed/test_elastic_ep.py b/tests/distributed/test_elastic_ep.py index 1d0f615d6ea..7c59d9dca5c 100644 --- a/tests/distributed/test_elastic_ep.py +++ b/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", diff --git a/tests/distributed/test_eplb_execute.py b/tests/distributed/test_eplb_execute.py index 0b87477950f..21fa057fd20 100644 --- a/tests/distributed/test_eplb_execute.py +++ b/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, diff --git a/tests/kernels/moe/test_moe_layer.py b/tests/kernels/moe/test_moe_layer.py index f17f5aa4ac3..5935c75a74f 100644 --- a/tests/kernels/moe/test_moe_layer.py +++ b/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, diff --git a/vllm/config/parallel.py b/vllm/config/parallel.py index 2904f40f8a4..a194640f2ec 100644 --- a/vllm/config/parallel.py +++ b/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 " diff --git a/vllm/distributed/elastic_ep/elastic_execute.py b/vllm/distributed/elastic_ep/elastic_execute.py index ac0def77e70..5aff5567d74 100644 --- a/vllm/distributed/elastic_ep/elastic_execute.py +++ b/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, diff --git a/vllm/distributed/eplb/eplb_communicator.py b/vllm/distributed/eplb/eplb_communicator.py index 9cccc05b2ce..6bd20c460e5 100644 --- a/vllm/distributed/eplb/eplb_communicator.py +++ b/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 = ( diff --git a/vllm/distributed/eplb/eplb_state.py b/vllm/distributed/eplb/eplb_state.py index 6208c03c4a8..1eb3a8feac5 100644 --- a/vllm/distributed/eplb/eplb_state.py +++ b/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, diff --git a/vllm/distributed/eplb/eplb_utils.py b/vllm/distributed/eplb/eplb_utils.py index f10891d6cdf..dee19749745 100644 --- a/vllm/distributed/eplb/eplb_utils.py +++ b/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", )