diff --git a/deepspeed/runtime/compiler.py b/deepspeed/runtime/compiler.py index 6c605658e2a3..d058088ed92a 100644 --- a/deepspeed/runtime/compiler.py +++ b/deepspeed/runtime/compiler.py @@ -3,11 +3,17 @@ # DeepSpeed Team +from __future__ import annotations + import torch import contextlib import functools +from collections import OrderedDict + +import torch.nn as nn from deepspeed.utils.torch import required_torch_version from deepspeed.accelerator import get_accelerator +from deepspeed.utils import logger try: from torch.compiler import is_compiling as torch_is_compiling @@ -111,3 +117,63 @@ def compile(): return torch.compile else: return dummy_decorator + + +def compile_autoep_non_moe_regions(model: nn.Module, backend, compile_kwargs: dict) -> list[str]: + """Compile decoder regions around AutoEP layers while keeping AutoEP eager.""" + from deepspeed.module_inject.auto_ep_layer import AutoEPMoELayer + + named_modules = dict(model.named_modules()) + autoep_modules = [(name, module) for name, module in named_modules.items() if isinstance(module, AutoEPMoELayer)] + if not autoep_modules: + raise ValueError("compile_mode='autoep_non_moe' requires at least one AutoEPMoELayer. " + "Enable expert_parallel and call compile() after deepspeed.initialize().") + + if "fullgraph" in compile_kwargs and compile_kwargs["fullgraph"] is not False: + raise ValueError("compile_mode='autoep_non_moe' requires fullgraph=False because AutoEP is an eager graph " + "break.") + if "dynamic" in compile_kwargs and compile_kwargs["dynamic"] is not False: + raise ValueError("compile_mode='autoep_non_moe' currently requires dynamic=False.") + + resolved_compile_kwargs = { + "fullgraph": False, + "dynamic": False, + **compile_kwargs, + "backend": backend, + } + regions: OrderedDict[str, nn.Module] = OrderedDict() + for module_name, _ in autoep_modules: + parent_name, separator, _ = module_name.rpartition(".") + if not separator: + raise ValueError("compile_mode='autoep_non_moe' cannot compile an AutoEPMoELayer at the model root.") + parent = named_modules[parent_name] + if type(parent).forward is nn.Module.forward: + raise ValueError(f"AutoEP compile region '{parent_name}' has no forward implementation. " + "The MoE layer must be a direct child of a callable decoder block.") + if getattr(parent, "_compiled_call_impl", None) is not None: + raise ValueError(f"AutoEP compile region '{parent_name}' is already compiled.") + regions.setdefault(parent_name, parent) + + original_forwards = {} + original_compiled_calls = {} + try: + for module_name, module in autoep_modules: + original_forwards[module] = module.__dict__.get("forward") + module.forward = disable(module.forward) + logger.debug("AutoEP regional compile: disabled compiler tracing for '%s'.", module_name) + + for region_name, region in regions.items(): + original_compiled_calls[region] = getattr(region, "_compiled_call_impl", None) + region.compile(**resolved_compile_kwargs) + logger.info("AutoEP regional compile: compiled '%s' with backend=%s.", region_name, backend) + except BaseException: + for module, original_forward in original_forwards.items(): + if original_forward is None: + del module.forward + else: + module.forward = original_forward + for region, compiled_call in original_compiled_calls.items(): + region._compiled_call_impl = compiled_call + raise + + return list(regions) diff --git a/deepspeed/runtime/engine.py b/deepspeed/runtime/engine.py index 1ade9d9ee37a..4705a48c78ee 100644 --- a/deepspeed/runtime/engine.py +++ b/deepspeed/runtime/engine.py @@ -557,6 +557,8 @@ def __init__(self, self.unflatten = _unflatten_dense_tensors self._is_compiled = False + self._compile_mode = None + self._compiled_regions = [] if is_deepcompile_supported(): # Predefined compile passes self.register_compile_pass(zero_1_and_2_compile.NAME_Z1, zero_1_and_2_compile.add_z1_reduce, @@ -5802,9 +5804,13 @@ def compile(self, backend=get_accelerator().get_compile_backend(), compile_kwargs={}, schedule=None, - compiled_autograd_enabled=False) -> None: + compiled_autograd_enabled=False, + compile_mode="model") -> None: """Compile the module using the specified backend and kwargs. - If a compiler_fn is set, it will be used instead of torch.compile(). + + ``compile_mode="model"`` compiles the full module. ``compile_mode="autoep_non_moe"`` + compiles each decoder block containing an AutoEP layer while keeping the AutoEP + router, token movement, expert compute, and collectives eager. """ # Avoid graph breaks deepspeed.utils.nvtx.enable_nvtx = False @@ -5812,13 +5818,47 @@ def compile(self, if not is_compile_supported(): raise RuntimeError("compile is not supported in your version of PyTorch.") + if compile_mode not in ("model", "autoep_non_moe"): + raise ValueError(f"Unknown compile_mode={compile_mode!r}; expected 'model' or 'autoep_non_moe'.") + if self.is_compiled: - return + if self._compile_mode == compile_mode: + return + raise RuntimeError(f"Engine is already compiled with compile_mode={self._compile_mode!r}.") if 'backend' in compile_kwargs: logger.warning("The `backend` in `compile_kwargs` will be overridden. Use the `backend` argument instead.") - logger.info(f"Compiling deepcompile={self.is_deepcompile_enabled()} backend={backend}") + logger.info(f"Compiling mode={compile_mode} deepcompile={self.is_deepcompile_enabled()} backend={backend}") + + if compile_mode == "autoep_non_moe": + if self.is_deepcompile_enabled(): + raise ValueError("compile_mode='autoep_non_moe' uses vanilla torch.compile and cannot be combined " + "with DeepCompile.") + autoep_config = getattr(self._config, "expert_parallel_config", None) + if getattr(autoep_config, "comm_backend", "comm") != "comm": + raise ValueError("compile_mode='autoep_non_moe' supports only expert_parallel.comm_backend='comm'.") + if self.autotp_size() > 1: + raise ValueError("compile_mode='autoep_non_moe' does not support AutoEP+AutoTP folding yet.") + if self._autoep_sequence_parallel_world_size() > 1: + raise ValueError("compile_mode='autoep_non_moe' does not support sequence parallelism yet.") + folding_spec = getattr(self, "_autoep_folding_spec", None) + if getattr(self, "pipeline_parallelism", False) or getattr(folding_spec, "pp_size", 1) > 1: + raise ValueError("compile_mode='autoep_non_moe' does not support pipeline parallelism yet.") + if self.zero_optimization_partition_weights(): + raise ValueError("compile_mode='autoep_non_moe' does not support ZeRO Stage 3 yet.") + if self.zero_offload_optimizer() is not None or self.zero_offload_param() is not None: + raise ValueError("compile_mode='autoep_non_moe' does not support optimizer or parameter offload yet.") + if schedule is not None: + raise ValueError("compile_mode='autoep_non_moe' does not support DeepCompile schedules.") + if compiled_autograd_enabled: + raise ValueError("compile_mode='autoep_non_moe' does not support compiled autograd yet.") + from .compiler import compile_autoep_non_moe_regions + self._compiled_regions = compile_autoep_non_moe_regions(self.module, backend, compile_kwargs) + self._is_compiled = True + self._compile_mode = compile_mode + self._compile_kwargs = compile_kwargs + return resolved_backend = None if self.is_deepcompile_enabled(): @@ -5842,6 +5882,7 @@ def compile(self, raise self._is_compiled = True + self._compile_mode = compile_mode self._compile_kwargs = compile_kwargs if compiled_autograd_enabled: if not self._deepcompile_active: diff --git a/docs/code-docs/source/autoep.rst b/docs/code-docs/source/autoep.rst index b772771dd164..c5b2e371a5dc 100644 --- a/docs/code-docs/source/autoep.rst +++ b/docs/code-docs/source/autoep.rst @@ -74,6 +74,38 @@ Weights-only/module-only Universal Checkpoint loads use the converted } } +Experimental regional ``torch.compile`` +---------------------------------------- + +AutoEP can keep its router, token movement, expert computation, and collectives +in eager mode while compiling the surrounding decoder blocks with vanilla +``torch.compile``. This targets fragmented attention, normalization, residual, +and dense backward work without capturing AutoEP communication in the graph. +The path is opt-in and does not change the default eager execution: + +.. code-block:: python + + engine, optimizer, _, _ = deepspeed.initialize( + model=model, + model_parameters=model.parameters(), + config=ds_config, + ) + engine.compile(compile_mode="autoep_non_moe") + +The call must happen after ``deepspeed.initialize()`` so AutoEP replacement is +complete. DeepSpeed discovers each ``AutoEPMoELayer`` and regionally compiles +its direct parent decoder block with ``fullgraph=False`` and ``dynamic=False``. +The AutoEP layer is an explicit compiler-disabled graph break, so routing, +AllToAll dispatch/combine, and expert execution remain eager. + +The initial experimental path supports vanilla ``torch.compile`` with the +standard ``comm`` backend, sequence and pipeline parallel sizes of one, and +ZeRO stages 0, 1, and 2. Distributed performance and parity validation currently +target ZeRO stage 1. It rejects DeepEP, DeepCompile, AutoEP+AutoTP folding, +sequence or pipeline parallelism, ZeRO stage 3, optimizer or parameter offload, +compiled autograd, DeepCompile schedules, and any ``fullgraph`` or ``dynamic`` +value other than ``False`` instead of silently changing the requested behavior. + **How it works:** 1. During ``deepspeed.initialize()``, AutoEP scans the model for MoE layers diff --git a/tests/unit/v1/moe/test_autoep_grad_parity.py b/tests/unit/v1/moe/test_autoep_grad_parity.py index 38ecd83808ec..544df7f8a774 100644 --- a/tests/unit/v1/moe/test_autoep_grad_parity.py +++ b/tests/unit/v1/moe/test_autoep_grad_parity.py @@ -4,14 +4,23 @@ # DeepSpeed Team """AutoEP gradient parity paths.""" +import copy + import deepspeed import deepspeed.comm as dist +import pytest import torch -from deepspeed.utils import safe_get_full_grad +import torch.nn as nn +from deepspeed.utils import safe_get_full_fp32_param, safe_get_full_grad +from torch.utils.checkpoint import checkpoint from unit.common import DistributedTest from unit.v1.moe.autoep_test_utils import ( + MockHFConfig, + MockMoEBlock, MockMoETransformer, engine_input_dtype as _engine_input_dtype, + h100_tests_enabled, + make_autoep_config, mixed_precision_config as _mixed_precision_config, seed_everything as _seed_everything, ) @@ -151,6 +160,172 @@ def _assert_grad_maps_close(actual, expected, *, lhs_name, rhs_name): f"expected_norm={expected[name].norm().item()}")) +class _CompiledDecoderLayer(nn.Module): + + def __init__(self): + super().__init__() + self.input_layernorm = nn.LayerNorm(128) + self.dense = nn.Linear(128, 128, bias=False) + self.post_attention_layernorm = nn.LayerNorm(128) + self.mlp = MockMoEBlock(num_experts=4, ffn_hidden=256, hidden_size=128) + + def forward(self, hidden_states): + residual = hidden_states + hidden_states = self.input_layernorm(hidden_states) + hidden_states = residual + self.dense(hidden_states) + residual = hidden_states + hidden_states = self.post_attention_layernorm(hidden_states) + return residual + self.mlp(hidden_states) + + +class _CompiledAutoEPModel(nn.Module): + + def __init__(self, checkpoint_enabled): + super().__init__() + self.config = copy.copy(MockHFConfig()) + self.config.hidden_size = 128 + self.config.intermediate_size = 256 + self.model = nn.Module() + self.model.layers = nn.ModuleList([_CompiledDecoderLayer() for _ in range(2)]) + self.output = nn.Linear(128, 64, bias=False) + self.checkpoint_enabled = checkpoint_enabled + with torch.no_grad(): + for name, param in self.named_parameters(): + if "layernorm" in name and param.ndim == 1: + param.fill_(1.0) + else: + param.normal_(mean=0.0, std=0.02) + + def forward(self, hidden_states): + for layer in self.model.layers: + if self.checkpoint_enabled and self.training: + hidden_states = checkpoint(layer, hidden_states, use_reentrant=False) + else: + hidden_states = layer(hidden_states) + return self.output(hidden_states) + + +def _make_compile_config(): + config = make_autoep_config(zero_stage=1, ep_size=2) + config.pop("fp16", None) + config["bf16"] = {"enabled": True} + config["optimizer"] = { + "type": "SGD", + "params": { + "lr": 1e-2 + }, + } + config["zero_allow_untested_optimizer"] = True + return config + + +def _snapshot_dynamo_stats(): + return dict(torch._dynamo.utils.counters["stats"]) + + +def _dynamo_stat_delta(before, after, key): + return after.get(key, 0) - before.get(key, 0) + + +def _snapshot_parameter_data(engine): + snapshot = {} + for name, param in engine.module.named_parameters(): + full_param = safe_get_full_fp32_param(param) + if full_param is None: + full_param = param + snapshot[name] = full_param.detach().float().cpu().clone() + return snapshot + + +def _run_compile_step(engine, batch): + input_tensor = batch.detach().clone().requires_grad_(True) + params_before = _snapshot_parameter_data(engine) + output = engine(input_tensor) + loss = output.float().square().mean() + engine.backward(loss) + + grads = {} + for name, param in engine.module.named_parameters(): + grad = safe_get_full_grad(param) + assert grad is not None, f"Expected gradient for {name}" + grads[name] = grad.detach().float().cpu().clone() + input_grad = input_tensor.grad.detach().float().cpu().clone() + + engine.step() + params_after = _snapshot_parameter_data(engine) + deltas = {name: params_after[name] - params_before[name] for name in params_before} + return { + "output": output.detach().float().cpu(), + "loss": loss.detach().float().cpu(), + "input_grad": input_grad, + "grads": grads, + "deltas": deltas, + } + + +def _warm_compile_step(engine, batch): + input_tensor = batch.detach().clone().requires_grad_(True) + output = engine(input_tensor) + output.float().square().mean().backward() + engine.zero_grad() + engine.optimizer.zero_grad() + + +def _assert_compile_step_close(actual, expected): + for name, rtol, atol in ( + ("output", 5e-3, 2e-2), + ("loss", 5e-3, 2e-3), + ("input_grad", 5e-3, 5e-3), + ): + difference = (actual[name] - expected[name]).abs() + torch.testing.assert_close(actual[name], + expected[name], + rtol=rtol, + atol=atol, + msg=(f"{name} mismatch; max_diff={difference.max().item()}, " + f"actual_norm={actual[name].norm().item()}, " + f"expected_norm={expected[name].norm().item()}")) + assert actual["grads"].keys() == expected["grads"].keys(), "Gradient parameter sets differ" + assert actual["deltas"].keys() == expected["deltas"].keys(), "Optimizer parameter sets differ" + for name in actual["grads"]: + torch.testing.assert_close(actual["grads"][name], + expected["grads"][name], + rtol=5e-3, + atol=5e-3, + msg=f"Gradient mismatch for {name}") + torch.testing.assert_close(actual["deltas"][name], + expected["deltas"][name], + rtol=5e-3, + atol=5e-5, + msg=(f"Optimizer delta max_diff=" + f"{(actual['deltas'][name] - expected['deltas'][name]).abs().max().item()}, " + f"actual_norm={actual['deltas'][name].norm().item()}, " + f"expected_norm={expected['deltas'][name].norm().item()}, name={name}")) + + +def _register_autoep_observers(engine): + from deepspeed.module_inject.auto_ep_layer import AutoEPMoELayer + + eager_calls = [] + routes = [] + handles = [] + for name, module in engine.module.named_modules(): + if not isinstance(module, AutoEPMoELayer): + continue + original_forward = module.forward + + @torch.compiler.disable + def observed_forward(*args, _forward=original_forward, _name=name, **kwargs): + eager_calls.append(_name) + return _forward(*args, **kwargs) + + module.forward = observed_forward + handles.append( + module.router.register_forward_hook(lambda _module, _inputs, output, name=name: routes.append( + (name, output[1].detach().cpu())))) + return eager_calls, routes, handles + + class TestAutoEPGradParity(DistributedTest): world_size = 4 @@ -229,3 +404,79 @@ def test_zero3_autoep_expert_grads_match_zero2_autoep(self): zero2_expert, lhs_name="ZeRO-3 AutoEP expert", rhs_name="ZeRO-2 AutoEP expert") + + +@pytest.mark.skipif(not h100_tests_enabled(), reason="AutoEP regional compile parity requires an H100 test run") +class TestAutoEPRegionalCompileParity(DistributedTest): + world_size = 2 + + @pytest.mark.parametrize("checkpoint_enabled", [True, False]) + def test_regional_compile_matches_eager(self, checkpoint_enabled): + seed = 3456 + _seed_everything(seed) + reference_model = _CompiledAutoEPModel(checkpoint_enabled) + reference_state = copy.deepcopy(reference_model.state_dict()) + + eager_model = _CompiledAutoEPModel(checkpoint_enabled) + compiled_model = _CompiledAutoEPModel(checkpoint_enabled) + eager_model.load_state_dict(reference_state) + compiled_model.load_state_dict(reference_state) + + eager_engine, _, _, _ = deepspeed.initialize(model=eager_model, config=_make_compile_config()) + compiled_engine, _, _, _ = deepspeed.initialize(model=compiled_model, config=_make_compile_config()) + torch._dynamo.reset() + torch._dynamo.utils.counters.clear() + compiled_engine.compile(compile_mode="autoep_non_moe") + + eager_calls, eager_routes, eager_handles = _register_autoep_observers(eager_engine) + compiled_calls, compiled_routes, compiled_handles = _register_autoep_observers(compiled_engine) + generator = torch.Generator().manual_seed(seed + dist.get_rank()) + dtype = _engine_input_dtype(eager_engine) + warmup_batch = torch.randn((1, 16, 128), generator=generator, dtype=dtype).to(eager_engine.device) + measured_batch = torch.randn((1, 16, 128), generator=generator, dtype=dtype).to(eager_engine.device) + + _warm_compile_step(eager_engine, warmup_batch) + _warm_compile_step(compiled_engine, warmup_batch) + + eager_call_start = len(eager_calls) + compiled_call_start = len(compiled_calls) + eager_route_start = len(eager_routes) + compiled_route_start = len(compiled_routes) + dynamo_start = _snapshot_dynamo_stats() + assert dynamo_start.get("unique_graphs", 0) > 0, f"Warmup did not capture graphs: {dynamo_start}" + assert dynamo_start.get("calls_captured", 0) > 0, f"Warmup did not capture calls: {dynamo_start}" + + measured_eager = _run_compile_step(eager_engine, measured_batch) + measured_compiled = _run_compile_step(compiled_engine, measured_batch) + _assert_compile_step_close(measured_compiled, measured_eager) + + dynamo_end = _snapshot_dynamo_stats() + expected_calls = len(compiled_engine._compiled_regions) * (2 if checkpoint_enabled else 1) + eager_call_delta = len(eager_calls) - eager_call_start + compiled_call_delta = len(compiled_calls) - compiled_call_start + assert eager_call_delta == expected_calls, f"Eager AutoEP calls: expected={expected_calls}, got={eager_call_delta}" + assert compiled_call_delta == expected_calls, ( + f"Compiled AutoEP calls: expected={expected_calls}, got={compiled_call_delta}") + unique_graph_delta = _dynamo_stat_delta(dynamo_start, dynamo_end, "unique_graphs") + captured_call_delta = _dynamo_stat_delta(dynamo_start, dynamo_end, "calls_captured") + assert unique_graph_delta == 0, f"Measured unique_graphs delta={unique_graph_delta}" + assert captured_call_delta == 0, f"Measured calls_captured delta={captured_call_delta}" + + measured_eager_routes = eager_routes[eager_route_start:] + measured_compiled_routes = compiled_routes[compiled_route_start:] + assert len(measured_eager_routes) == expected_calls, ( + f"Eager routes: expected={expected_calls}, got={len(measured_eager_routes)}") + assert len(measured_compiled_routes) == expected_calls, ( + f"Compiled routes: expected={expected_calls}, got={len(measured_compiled_routes)}") + for (eager_name, eager_route), (compiled_name, compiled_route) in zip(measured_eager_routes, + measured_compiled_routes): + assert eager_name == compiled_name, f"Route layer mismatch: {eager_name} != {compiled_name}" + assert torch.equal(eager_route, compiled_route), f"Route assignment mismatch for {eager_name}" + + grad_names = measured_compiled["grads"] + assert any(".experts.w1" in name for name in grad_names), "Expert gradients were not checked" + assert any(".router.gate.weight" in name for name in grad_names), "Router gradients were not checked" + assert any(".dense.weight" in name for name in grad_names), "Non-MoE gradients were not checked" + + for handle in eager_handles + compiled_handles: + handle.remove() diff --git a/tests/unit/v1/moe/test_autoep_unit.py b/tests/unit/v1/moe/test_autoep_unit.py index d28eb96c8036..c69db1f2b0cd 100644 --- a/tests/unit/v1/moe/test_autoep_unit.py +++ b/tests/unit/v1/moe/test_autoep_unit.py @@ -46,9 +46,11 @@ from deepspeed.moe.ep_repack import repack_expert_weights from deepspeed.moe.ep_router import TokenChoiceTopKRouter from deepspeed.runtime.engine import DeepSpeedEngine +from deepspeed.runtime.compiler import compile_autoep_non_moe_regions from deepspeed.runtime.zero.stage3 import DeepSpeedZeroOptimizer_Stage3 from deepspeed.utils import groups from unit.v1.moe.autoep_test_utils import ( + MockHFConfig, MockMoEBlock, MockMoETransformer, UNSUPPORTED_LOAD_BALANCE_VALUES, @@ -101,6 +103,41 @@ def _assert_same_dtype_device(actual, expected): assert actual.device == expected.device +class _CallableMoEDecoderLayer(nn.Module): + + def __init__(self): + super().__init__() + self.input_layernorm = nn.LayerNorm(64) + self.dense = nn.Linear(64, 64, bias=False) + self.mlp = MockMoEBlock() + + def forward(self, hidden_states): + residual = hidden_states + hidden_states = self.input_layernorm(hidden_states) + hidden_states = residual + self.dense(hidden_states) + return hidden_states + self.mlp(hidden_states) + + +class _CallableMoETransformer(nn.Module): + + def __init__(self, num_layers=2): + super().__init__() + self.config = MockHFConfig() + self.model = nn.Module() + self.model.layers = nn.ModuleList([_CallableMoEDecoderLayer() for _ in range(num_layers)]) + + def forward(self, hidden_states): + for layer in self.model.layers: + hidden_states = layer(hidden_states) + return hidden_states + + +def _replace_callable_autoep_layers(num_layers=2): + model = _CallableMoETransformer(num_layers=num_layers) + replace_autoep_layers(model, "mixtral", expected_count=num_layers) + return model + + def _mark_fake_zero_param(param, full_data, partition_data=None, ds_id=0, name="param"): param.ds_id = ds_id param.ds_shape = torch.Size(full_data.shape) @@ -833,6 +870,188 @@ def test_invalid_routed_scaling_factor_rejected(self, value): _resolve_route_scale(AutoEPConfig(enabled=True, routed_scaling_factor=value), None) +class TestAutoEPRegionalCompile: + + def test_rejects_model_without_autoep_layers(self): + with pytest.raises(ValueError, match="requires at least one AutoEPMoELayer"): + compile_autoep_non_moe_regions(nn.Linear(4, 4), backend="eager", compile_kwargs={}) + + def test_rejects_non_callable_parent_region(self): + model = MockMoETransformer(num_layers=1) + replace_autoep_layers(model, "mixtral", expected_count=1) + with pytest.raises(ValueError, match="has no forward implementation"): + compile_autoep_non_moe_regions(model, backend="eager", compile_kwargs={}) + + def test_compiles_decoder_parents_and_disables_autoep(self, monkeypatch): + model = _replace_callable_autoep_layers() + compile_calls = [] + + def record_compile(module, **kwargs): + compile_calls.append((module, kwargs)) + module._compiled_call_impl = object() + + monkeypatch.setattr(_CallableMoEDecoderLayer, "compile", record_compile) + + regions = compile_autoep_non_moe_regions(model, backend="eager", compile_kwargs={}) + + assert regions == ["model.layers.0", "model.layers.1"] + assert [module for module, _ in compile_calls] == list(model.model.layers) + assert all(kwargs == {"backend": "eager", "dynamic": False, "fullgraph": False} for _, kwargs in compile_calls) + for layer in model.model.layers: + assert getattr(layer.mlp.forward, "_torchdynamo_disable", False) + + def test_deduplicates_shared_decoder_parent(self, monkeypatch): + model = _replace_callable_autoep_layers(num_layers=1) + model.model.layers[0].second_mlp = AutoEPMoELayer( + spec=_make_spec(moe_module_name="model.layers.0.second_mlp"), + source_module=MockMoEBlock(), + ep_size=1, + ep_rank=0, + config=_runtime_config(), + ) + compile_calls = [] + + def record_compile(module, **kwargs): + compile_calls.append(module) + module._compiled_call_impl = object() + + monkeypatch.setattr(_CallableMoEDecoderLayer, "compile", record_compile) + + regions = compile_autoep_non_moe_regions(model, backend="eager", compile_kwargs={}) + + assert regions == ["model.layers.0"] + assert compile_calls == [model.model.layers[0]] + assert getattr(model.model.layers[0].mlp.forward, "_torchdynamo_disable", False) + assert getattr(model.model.layers[0].second_mlp.forward, "_torchdynamo_disable", False) + + @pytest.mark.parametrize( + "compile_kwargs, match", + [ + ({ + "fullgraph": True + }, "fullgraph=False"), + ({ + "fullgraph": None + }, "fullgraph=False"), + ({ + "dynamic": True + }, "dynamic=False"), + ({ + "dynamic": None + }, "dynamic=False"), + ], + ) + def test_rejects_unsupported_compile_kwargs(self, compile_kwargs, match): + model = _replace_callable_autoep_layers(num_layers=1) + with pytest.raises(ValueError, match=match): + compile_autoep_non_moe_regions(model, backend="eager", compile_kwargs=compile_kwargs) + + def test_rolls_back_partial_compilation(self, monkeypatch): + model = _replace_callable_autoep_layers() + original_forwards = [layer.mlp.forward for layer in model.model.layers] + compile_calls = 0 + + def fail_second_compile(module, **kwargs): + nonlocal compile_calls + compile_calls += 1 + module._compiled_call_impl = object() + if compile_calls == 2: + raise RuntimeError("compile failed") + + monkeypatch.setattr(_CallableMoEDecoderLayer, "compile", fail_second_compile) + + with pytest.raises(RuntimeError, match="compile failed"): + compile_autoep_non_moe_regions(model, backend="eager", compile_kwargs={}) + + for layer, original_forward in zip(model.model.layers, original_forwards): + assert "forward" not in layer.mlp.__dict__ + assert layer.mlp.forward.__func__ is original_forward.__func__ + assert layer._compiled_call_impl is None + + @pytest.mark.parametrize( + "condition, match", + [ + ("deepcompile", "cannot be combined with DeepCompile"), + ("deepep", "comm_backend='comm'"), + ("autotp", "AutoEP\\+AutoTP folding"), + ("sequence_parallel", "sequence parallelism"), + ("pipeline_parallel", "pipeline parallelism"), + ("zero3", "ZeRO Stage 3"), + ("optimizer_offload", "optimizer or parameter offload"), + ("param_offload", "optimizer or parameter offload"), + ("schedule", "DeepCompile schedules"), + ("compiled_autograd", "compiled autograd"), + ], + ) + def test_engine_rejects_unsupported_modes(self, monkeypatch, condition, match): + model = _replace_callable_autoep_layers(num_layers=1) + engine = object.__new__(DeepSpeedEngine) + nn.Module.__init__(engine) + engine.module = model + engine._config = SimpleNamespace( + compile_config=SimpleNamespace(deepcompile=condition == "deepcompile"), + expert_parallel_config=SimpleNamespace(comm_backend="deepep" if condition == "deepep" else "comm"), + ) + engine._is_compiled = False + engine._compile_mode = None + engine._compiled_regions = [] + engine.autotp_size = lambda: 2 if condition == "autotp" else 1 + engine._autoep_sequence_parallel_world_size = lambda: 2 if condition == "sequence_parallel" else 1 + engine.pipeline_parallelism = condition == "pipeline_parallel" + engine._autoep_folding_spec = None + engine.zero_optimization_partition_weights = lambda: condition == "zero3" + engine.zero_offload_optimizer = lambda: object() if condition == "optimizer_offload" else None + engine.zero_offload_param = lambda: object() if condition == "param_offload" else None + monkeypatch.setattr(_CallableMoEDecoderLayer, "compile", lambda module, **kwargs: None) + + with pytest.raises(ValueError, match=match): + engine.compile( + backend="eager", + compile_mode="autoep_non_moe", + schedule=[] if condition == "schedule" else None, + compiled_autograd_enabled=condition == "compiled_autograd", + ) + + def test_engine_rejects_unknown_compile_mode(self): + engine = object.__new__(DeepSpeedEngine) + nn.Module.__init__(engine) + engine._is_compiled = False + with pytest.raises(ValueError, match="Unknown compile_mode"): + engine.compile(backend="eager", compile_mode="unknown") + + def test_engine_tracks_regional_compile_mode(self, monkeypatch): + model = _replace_callable_autoep_layers() + engine = object.__new__(DeepSpeedEngine) + nn.Module.__init__(engine) + engine.module = model + engine._config = SimpleNamespace( + compile_config=SimpleNamespace(deepcompile=False), + expert_parallel_config=SimpleNamespace(comm_backend="comm"), + ) + engine._is_compiled = False + engine._compile_mode = None + engine._compiled_regions = [] + engine._is_compiled_autograd_enabled = False + engine.autotp_size = lambda: 1 + engine._autoep_sequence_parallel_world_size = lambda: 1 + engine.pipeline_parallelism = False + engine._autoep_folding_spec = None + engine.zero_optimization_partition_weights = lambda: False + engine.zero_offload_optimizer = lambda: None + engine.zero_offload_param = lambda: None + monkeypatch.setattr(_CallableMoEDecoderLayer, "compile", + lambda module, **kwargs: setattr(module, "_compiled_call_impl", object())) + + engine.compile(backend="eager", compile_mode="autoep_non_moe") + engine.compile(backend="eager", compile_mode="autoep_non_moe") + + assert engine.is_compiled + assert engine._compile_mode == "autoep_non_moe" + assert engine._compiled_regions == ["model.layers.0", "model.layers.1"] + with pytest.raises(RuntimeError, match="already compiled"): + engine.compile(backend="eager") + + class TestRoutingAndLayerSemantics: def test_router_route_scale_and_group_limited_routing(self):