diff --git a/tests/diffusion/distributed/test_expert_parallel_layout.py b/tests/diffusion/distributed/test_expert_parallel_layout.py index d7b66be5d..bfbbca588 100644 --- a/tests/diffusion/distributed/test_expert_parallel_layout.py +++ b/tests/diffusion/distributed/test_expert_parallel_layout.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project from types import SimpleNamespace @@ -194,6 +194,47 @@ def test_moe_ep_maps_diffusion_sp_cfg_dp_to_vllm_groups(monkeypatch): assert ep_groups == [parallel_state.vllm_parallel_state._EP.local_group] +@pytest.mark.cpu +@pytest.mark.core_model +@pytest.mark.parametrize( + ("fully_shard_degree", "error_message"), + [(0, "fully_shard_degree must be positive"), (3, "must be divisible by fully_shard_degree")], +) +def test_invalid_hsdp_shard_size_fails_before_group_creation(monkeypatch, fully_shard_degree: int, error_message: str): + """Reject invalid HSDP shard sizes before allocating any process groups.""" + world_size = 4 + created_groups: list[object] = [] + + def fail_if_group_created(*args, **kwargs): + del args, kwargs + created_groups.append(object()) + raise AssertionError("a process group was created before HSDP validation") + + fake_world_group = SimpleNamespace( + rank_in_group=0, + local_rank=0, + device_group=object(), + ) + monkeypatch.setattr(parallel_state.torch.distributed, "is_initialized", lambda: True) + monkeypatch.setattr(parallel_state.torch.distributed, "get_world_size", lambda: world_size) + monkeypatch.setattr(parallel_state, "get_world_group", lambda: fake_world_group) + monkeypatch.setattr(parallel_state, "init_model_parallel_group", fail_if_group_created) + + for name in ("_DP", "_CFG", "_SP", "_PP", "_FS", "_HSDP_REPLICATE", "_EXPERT_PARALLEL_GROUP_RANKS"): + monkeypatch.setattr(parallel_state, name, None) + for name in ("_TP", "_PCP", "_DP", "_EP", "_PP"): + monkeypatch.setattr(parallel_state.vllm_parallel_state, name, None, raising=False) + + with pytest.raises(ValueError, match=error_message): + parallel_state.initialize_model_parallel( + fully_shard_degree=fully_shard_degree, + use_hsdp=True, + backend="gloo", + ) + + assert created_groups == [] + + @pytest.mark.cpu @pytest.mark.core_model def test_cfg_parallel_keeps_diffusion_dp_without_ep(monkeypatch): diff --git a/vllm_omni/diffusion/distributed/parallel_state.py b/vllm_omni/diffusion/distributed/parallel_state.py index 7d219ebbe..85ba9cd6c 100644 --- a/vllm_omni/diffusion/distributed/parallel_state.py +++ b/vllm_omni/diffusion/distributed/parallel_state.py @@ -782,6 +782,12 @@ def _initialize_model_parallel( raise ValueError("HSDP (FSDP2) requires data_parallel_size to be 1") if non_dp_size not in (1, world_size): raise ValueError(f"HSDP non-DP parallel size must be 1 or WORLD size ({world_size}), but got {non_dp_size}") + if fully_shard_degree <= 0: + raise ValueError(f"fully_shard_degree must be positive, got {fully_shard_degree}") + if world_size % fully_shard_degree != 0: + raise ValueError( + f"WORLD size ({world_size}) must be divisible by fully_shard_degree ({fully_shard_degree})" + ) data_parallel_size = 1 else: inferred_data_parallel_size = world_size // non_dp_size @@ -901,13 +907,6 @@ def _initialize_model_parallel( ) if use_hsdp: - if fully_shard_degree <= 0: - raise ValueError(f"fully_shard_degree must be positive, got {fully_shard_degree}") - if world_size % fully_shard_degree != 0: - raise ValueError( - f"WORLD size ({world_size}) must be divisible by fully_shard_degree ({fully_shard_degree})" - ) - assert _FS is None, "fully shard group is already initialized" # HSDP builds its mesh from arange(world_size).reshape(replicate, shard), # so each consecutive rank run is one fully-sharded group.