diff --git a/gpt_builders.py b/gpt_builders.py index 37a23e3434b..d393961bb04 100644 --- a/gpt_builders.py +++ b/gpt_builders.py @@ -132,7 +132,6 @@ def _get_transformer_layer_spec(use_te, config): args.qk_layernorm, args.multi_latent_attention, args.experimental_attention_variant, - moe_use_legacy_grouped_gemm=args.moe_use_legacy_grouped_gemm, qk_l2_norm=args.qk_l2_norm, use_kitchen=config.use_kitchen, use_te_activation_func=config.use_te_activation_func, @@ -152,7 +151,6 @@ def _get_transformer_layer_spec(use_te, config): args.qk_layernorm, args.multi_latent_attention, args.experimental_attention_variant, - moe_use_legacy_grouped_gemm=args.moe_use_legacy_grouped_gemm, normalization=args.normalization, use_kitchen=config.use_kitchen, use_kitchen_attention=config.use_kitchen_attention, diff --git a/megatron/core/extensions/transformer_engine_spec_provider.py b/megatron/core/extensions/transformer_engine_spec_provider.py index a445eacdfe0..4ac6a061552 100644 --- a/megatron/core/extensions/transformer_engine_spec_provider.py +++ b/megatron/core/extensions/transformer_engine_spec_provider.py @@ -20,7 +20,6 @@ from megatron.core.tensor_parallel.layers import ColumnParallelLinear, RowParallelLinear from megatron.core.transformer.mlp import MLPSubmodules, TEActivationFunctionBuilder from megatron.core.transformer.moe.experts import ( - GroupedMLP, SequentialMLP, TEGroupedMLP, TEGroupedMLPSubmodules, @@ -66,27 +65,16 @@ def core_attention(self) -> type: return TEDotProductAttention def grouped_mlp_modules( - self, moe_use_grouped_gemm: bool, moe_use_legacy_grouped_gemm: bool + self, moe_use_grouped_gemm: bool ) -> ( tuple[type[TEGroupedMLP], TEGroupedMLPSubmodules] | tuple[type[SequentialMLP], MLPSubmodules] - | tuple[type[GroupedMLP], None] ): """Which module and submodules to use for grouped mlp""" - if ( - moe_use_grouped_gemm - and TEColumnParallelGroupedLinear is not None - and not moe_use_legacy_grouped_gemm - ): + if moe_use_grouped_gemm and TEColumnParallelGroupedLinear is not None: return TEGroupedMLP, TEGroupedMLPSubmodules( linear_fc1=TEColumnParallelGroupedLinear, linear_fc2=TERowParallelGroupedLinear ) - elif moe_use_grouped_gemm: - warnings.warn( - 'The legacy GroupedMLP will be deprecated in Megatron-Core v0.12.0. ' - 'Please update the TransformerEngine to version>=1.7.0 and use TEGroupedMLP.' - ) - return GroupedMLP, None else: if not is_te_min_version("1.7.0.dev0"): warnings.warn( diff --git a/megatron/core/models/backends.py b/megatron/core/models/backends.py index ebb979772f0..f3995519595 100644 --- a/megatron/core/models/backends.py +++ b/megatron/core/models/backends.py @@ -13,7 +13,6 @@ from megatron.core.transformer.dot_product_attention import DotProductAttention from megatron.core.transformer.mlp import MLPSubmodules, TEActivationFunctionBuilder from megatron.core.transformer.moe.experts import ( - GroupedMLP, InferenceGroupedMLP, SequentialMLP, TEGroupedMLPSubmodules, @@ -84,7 +83,7 @@ def core_attention(self) -> type: @abstractmethod def grouped_mlp_modules( - self, moe_use_grouped_gemm: bool, moe_use_legacy_grouped_gemm: bool + self, moe_use_grouped_gemm: bool ) -> tuple[type, MLPSubmodules | TEGroupedMLPSubmodules | None]: """Which module and submodules to use for grouped mlp""" ... @@ -128,19 +127,12 @@ def core_attention(self) -> type: return DotProductAttention def grouped_mlp_modules( - self, moe_use_grouped_gemm: bool, moe_use_legacy_grouped_gemm: bool - ) -> tuple[type[GroupedMLP], None] | tuple[type[SequentialMLP], MLPSubmodules]: + self, moe_use_grouped_gemm: bool + ) -> tuple[type[SequentialMLP], MLPSubmodules]: """Which module and submodules to use for grouped mlp""" - if moe_use_grouped_gemm: - warnings.warn( - "The legacy GroupedMLP will be deprecated in Megatron-Core v0.12.0. " - "Please update the TransformerEngine to version>=1.7.0 and use TEGroupedMLP." - ) - return GroupedMLP, None - else: - return SequentialMLP, MLPSubmodules( - linear_fc1=ColumnParallelLinear, linear_fc2=RowParallelLinear - ) + return SequentialMLP, MLPSubmodules( + linear_fc1=ColumnParallelLinear, linear_fc2=RowParallelLinear + ) def activation_func(self) -> TEActivationFunctionBuilder | None: """Which module to use for activation function""" @@ -190,7 +182,7 @@ def activation_func(self) -> TEActivationFunctionBuilder | None: return cast(TEActivationFunctionBuilder, TEActivationOp) def grouped_mlp_modules( - self, moe_use_grouped_gemm: bool, moe_use_legacy_grouped_gemm: bool + self, moe_use_grouped_gemm: bool ) -> Tuple[type, Optional[MLPSubmodules]]: """Which module and submodules to use for grouped mlp""" return InferenceGroupedMLP, MLPSubmodules( diff --git a/megatron/core/models/gpt/experimental_attention_variant_module_specs.py b/megatron/core/models/gpt/experimental_attention_variant_module_specs.py index a7cc7cc0a55..e20f88ee4d1 100644 --- a/megatron/core/models/gpt/experimental_attention_variant_module_specs.py +++ b/megatron/core/models/gpt/experimental_attention_variant_module_specs.py @@ -392,7 +392,6 @@ def _get_self_attention_module_spec( moe_grouped_gemm=config.moe_grouped_gemm, qk_layernorm=config.qk_layernorm, multi_latent_attention=config.multi_latent_attention, - moe_use_legacy_grouped_gemm=config.moe_use_legacy_grouped_gemm, qk_l2_norm=config.qk_l2_norm, use_kitchen=config.use_kitchen, use_te_activation_func=config.use_te_activation_func, @@ -444,7 +443,6 @@ def _get_moe_module_spec( backend=backend, num_experts=config.num_moe_experts, moe_grouped_gemm=config.moe_grouped_gemm, - moe_use_legacy_grouped_gemm=config.moe_use_legacy_grouped_gemm, use_te_activation_func=config.use_te_activation_func, ) moe_spec.metainfo["fuse_pre_mlp_layernorm"] = False diff --git a/megatron/core/models/gpt/gpt_layer_specs.py b/megatron/core/models/gpt/gpt_layer_specs.py index aae2d5f3e81..103601a3be0 100755 --- a/megatron/core/models/gpt/gpt_layer_specs.py +++ b/megatron/core/models/gpt/gpt_layer_specs.py @@ -93,7 +93,6 @@ def get_gpt_layer_with_inference_submodules( backend=backend, num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, use_te_op_fuser=False, use_te_activation_func=False, ) @@ -179,7 +178,6 @@ def get_gpt_layer_with_transformer_engine_submodules( qk_layernorm: Optional[bool] = False, multi_latent_attention: Optional[bool] = False, fp8: Optional[str] = None, # pylint: disable=unused-argument - moe_use_legacy_grouped_gemm: Optional[bool] = False, qk_l2_norm: Optional[bool] = False, use_te_op_fuser: Optional[bool] = False, use_kitchen: bool = False, @@ -197,8 +195,6 @@ def get_gpt_layer_with_transformer_engine_submodules( qk_layernorm (bool, optional): To use layernorm for queries/keys. Defaults to False. multi_latent_attention (bool, optional): To use MLA. Defaults to False. fp8 (str, optional): Deprecated. For temporary Nemo compatibility. - moe_use_legacy_grouped_gemm (bool, optional): Force use the legacy GroupedMLP. - Defaults to False. qk_l2_norm (bool, optional): To use l2 norm for queries/keys. Defaults to False. use_te_op_fuser (bool, optional): Use Transformer Engine's operation-based API, which may enable certain operation fusions. Defaults to False. @@ -231,7 +227,6 @@ def get_gpt_layer_with_transformer_engine_submodules( backend=backend, num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, use_te_op_fuser=use_te_op_fuser, use_te_activation_func=use_te_activation_func, ) @@ -318,7 +313,6 @@ def get_gpt_layer_local_submodules( qk_layernorm: Optional[bool] = False, multi_latent_attention: Optional[bool] = False, fp8: Optional[str] = None, # pylint: disable=unused-argument - moe_use_legacy_grouped_gemm: Optional[bool] = False, normalization: Optional[str] = None, qk_l2_norm: Optional[bool] = False, use_kitchen: bool = False, @@ -334,8 +328,6 @@ def get_gpt_layer_local_submodules( qk_layernorm (bool, optional): To use layernorm for queries/keys. Defaults to False. multi_latent_attention (bool, optional): To use MLA. Defaults to False. fp8 (str, optional): Deprecated. For temporary Nemo compatibility. - moe_use_legacy_grouped_gemm (bool, optional): Force use the legacy GroupedMLP. - Defaults to False. qk_l2_norm (bool, optional): To use l2 norm for queries/keys. Defaults to False. Returns: @@ -366,10 +358,7 @@ def get_gpt_layer_local_submodules( ) mlp = get_mlp_module_spec_for_backend( - backend=backend, - num_experts=num_experts, - moe_grouped_gemm=moe_grouped_gemm, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, + backend=backend, num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm ) if multi_latent_attention: @@ -438,7 +427,6 @@ def _get_mlp_module_spec( num_experts: Optional[int] = None, moe_grouped_gemm: Optional[bool] = False, fp8: Optional[str] = None, # pylint: disable=unused-argument - moe_use_legacy_grouped_gemm: Optional[bool] = False, ): warnings.warn( """This private function is on a deprecation track. Please switch to `get_mlp_module_spec` @@ -446,11 +434,7 @@ def _get_mlp_module_spec( ) return get_mlp_module_spec( - use_te=use_te, - num_experts=num_experts, - moe_grouped_gemm=moe_grouped_gemm, - fp8=fp8, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, + use_te=use_te, num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm, fp8=fp8 ) @@ -459,7 +443,6 @@ def get_mlp_module_spec( num_experts: Optional[int] = None, moe_grouped_gemm: Optional[bool] = False, fp8: Optional[str] = None, # pylint: disable=unused-argument - moe_use_legacy_grouped_gemm: Optional[bool] = False, use_te_op_fuser: Optional[bool] = False, ) -> ModuleSpec: """Helper function to get module spec for MLP/MoE""" @@ -482,7 +465,6 @@ def get_mlp_module_spec( backend=TESpecProvider() if use_te else LocalSpecProvider(), num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, use_te_op_fuser=use_te_op_fuser, ) @@ -491,7 +473,6 @@ def get_mlp_module_spec_for_backend( backend: BackendSpecProvider, num_experts: Optional[int] = None, moe_grouped_gemm: Optional[bool] = False, - moe_use_legacy_grouped_gemm: Optional[bool] = False, use_te_op_fuser: Optional[bool] = False, use_te_activation_func: bool = False, ) -> ModuleSpec: @@ -520,7 +501,6 @@ def get_mlp_module_spec_for_backend( backend=backend, num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, use_te_activation_func=use_te_activation_func, ) @@ -541,7 +521,6 @@ def get_gpt_decoder_layer_specs( moe_grouped_gemm=False, qk_layernorm=config.qk_layernorm, multi_latent_attention=config.multi_latent_attention, - moe_use_legacy_grouped_gemm=config.moe_use_legacy_grouped_gemm, qk_l2_norm=qk_l2_norm, use_kitchen=config.use_kitchen, use_te_activation_func=config.use_te_activation_func, @@ -553,7 +532,6 @@ def get_gpt_decoder_layer_specs( moe_grouped_gemm=config.moe_grouped_gemm, qk_layernorm=config.qk_layernorm, multi_latent_attention=config.multi_latent_attention, - moe_use_legacy_grouped_gemm=config.moe_use_legacy_grouped_gemm, qk_l2_norm=qk_l2_norm, use_kitchen=config.use_kitchen, use_te_activation_func=config.use_te_activation_func, @@ -582,7 +560,6 @@ def get_gpt_decoder_layer_specs( moe_grouped_gemm=False, qk_layernorm=config.qk_layernorm, multi_latent_attention=config.multi_latent_attention, - moe_use_legacy_grouped_gemm=config.moe_use_legacy_grouped_gemm, normalization=normalization, qk_l2_norm=qk_l2_norm, use_kitchen=config.use_kitchen, @@ -594,7 +571,6 @@ def get_gpt_decoder_layer_specs( moe_grouped_gemm=config.moe_grouped_gemm, qk_layernorm=config.qk_layernorm, multi_latent_attention=config.multi_latent_attention, - moe_use_legacy_grouped_gemm=config.moe_use_legacy_grouped_gemm, normalization=normalization, qk_l2_norm=qk_l2_norm, use_kitchen=config.use_kitchen, diff --git a/megatron/core/models/gpt/moe_module_specs.py b/megatron/core/models/gpt/moe_module_specs.py index 4b0d5640b46..f67f08331e6 100755 --- a/megatron/core/models/gpt/moe_module_specs.py +++ b/megatron/core/models/gpt/moe_module_specs.py @@ -19,7 +19,6 @@ def get_moe_module_spec( use_te: Optional[bool] = True, num_experts: Optional[int] = None, moe_grouped_gemm: Optional[bool] = False, - moe_use_legacy_grouped_gemm: Optional[bool] = False, ) -> ModuleSpec: """Helper function to get module spec for MoE. @@ -37,10 +36,7 @@ def get_moe_module_spec( else: backend = LocalSpecProvider() return get_moe_module_spec_for_backend( - backend=backend, - num_experts=num_experts, - moe_grouped_gemm=moe_grouped_gemm, - moe_use_legacy_grouped_gemm=moe_use_legacy_grouped_gemm, + backend=backend, num_experts=num_experts, moe_grouped_gemm=moe_grouped_gemm ) @@ -48,7 +44,6 @@ def get_moe_module_spec_for_backend( backend: BackendSpecProvider, num_experts: Optional[int] = None, moe_grouped_gemm: Optional[bool] = False, - moe_use_legacy_grouped_gemm: Optional[bool] = False, use_te_activation_func: bool = False, ) -> ModuleSpec: """Helper function to get module spec for MoE""" @@ -63,8 +58,7 @@ def get_moe_module_spec_for_backend( ) expert_module, expert_submodule = backend.grouped_mlp_modules( - moe_grouped_gemm is not None and moe_grouped_gemm, - moe_use_legacy_grouped_gemm is not None and moe_use_legacy_grouped_gemm, + moe_grouped_gemm is not None and moe_grouped_gemm ) if expert_submodule is not None: expert_submodule.activation_func = activation_func @@ -95,7 +89,7 @@ def get_inference_optimized_moe_spec() -> ModuleSpec: backend = InferenceSpecProvider() activation_func = backend.activation_func() - expert_module, expert_submodule = backend.grouped_mlp_modules(True, False) + expert_module, expert_submodule = backend.grouped_mlp_modules(True) if expert_submodule is not None: expert_submodule.activation_func = activation_func diff --git a/megatron/core/models/mamba/mamba_layer_specs.py b/megatron/core/models/mamba/mamba_layer_specs.py index 957f20847fc..39f7fab0266 100755 --- a/megatron/core/models/mamba/mamba_layer_specs.py +++ b/megatron/core/models/mamba/mamba_layer_specs.py @@ -41,7 +41,6 @@ use_te=True, num_experts=8, # Can be any positive integer (must not be None). moe_grouped_gemm=True, - moe_use_legacy_grouped_gemm=False, ) # Inference-optimized MoE spec diff --git a/megatron/core/transformer/moe/experts.py b/megatron/core/transformer/moe/experts.py index 0fc954db4af..2401276ac38 100644 --- a/megatron/core/transformer/moe/experts.py +++ b/megatron/core/transformer/moe/experts.py @@ -1,41 +1,26 @@ # Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. from __future__ import annotations -import copy import logging from collections.abc import Callable from copy import deepcopy from dataclasses import dataclass -from functools import partial from math import ceil from typing import Optional, Protocol, Tuple import torch import torch.nn.functional as F -from torch.nn.parameter import Parameter from megatron.core import tensor_parallel from megatron.core.activations import squared_relu -from megatron.core.dist_checkpointing import ShardedTensor -from megatron.core.dist_checkpointing.mapping import ( - LocalNonpersistentObject, - ReplicaId, - ShardedStateDict, - ShardedTensorFactory, -) +from megatron.core.dist_checkpointing.mapping import ShardedStateDict from megatron.core.dist_checkpointing.utils import replace_prefix_for_sharding from megatron.core.fusions.fused_bias_geglu import quick_gelu, weighted_bias_quick_geglu_impl from megatron.core.fusions.fused_bias_swiglu import weighted_bias_swiglu_impl from megatron.core.fusions.fused_weighted_squared_relu import weighted_squared_relu_impl -from megatron.core.jit import jit_fuser from megatron.core.pipeline_parallel.fine_grained_activation_offload import ( FineGrainedActivationOffloadingInterface as off_interface, ) -from megatron.core.tensor_parallel.layers import ( - _initialize_affine_weight_cpu, - _initialize_affine_weight_gpu, -) -from megatron.core.tensor_parallel.utils import divide from megatron.core.transformer.mlp import ( MLP, MLPSubmodules, @@ -43,7 +28,6 @@ apply_swiglu_sharded_factory, ) from megatron.core.transformer.module import MegatronModule -from megatron.core.transformer.moe import grouped_gemm_util as gg from megatron.core.transformer.moe.moe_utils import ( ProcessGroupCollection, get_align_size_for_quantization, @@ -51,7 +35,6 @@ from megatron.core.transformer.transformer_config import TransformerConfig from megatron.core.transformer.utils import ( ensure_metadata_has_dp_cp_group, - make_sharded_object_for_checkpoint, sharded_state_dict_default, ) from megatron.core.typed_torch import apply_module, not_none @@ -79,453 +62,6 @@ logger = logging.getLogger(__name__) -class GroupedMLP(MegatronModule): - """An efficient implementation of the Experts layer using GroupedGEMM. - - Executes multiple experts in parallel to maximize computational efficiency. - """ - - # TODO(M4): breaking api, switched from pass in tp_group to pass in pg_collection. - def __init__( - self, - num_local_experts: int, - config: TransformerConfig, - pg_collection: Optional[ProcessGroupCollection] = None, - ): - super().__init__(config=config) - self.config: TransformerConfig = config - self.num_local_experts = num_local_experts - gg.assert_grouped_gemm_is_available() - assert ( - config.add_bias_linear == False - ), "bias not supported in Grouped GEMM yet, please set '--disable-bias-linear' instead." - assert ( - config.moe_latent_size is None - ), "MoE latent projection not supported in GroupedMLP yet." - - self.expert_parallel = config.expert_model_parallel_size > 1 - if self.config.gated_linear_unit: - if self.config.activation_func not in (F.silu, F.gelu): - raise ValueError("Activation function must be silu or gelu when using GroupedMLP.") - - @jit_fuser - def glu(x): - x = torch.chunk(x, 2, dim=-1) - return self.config.activation_func(x[0]) * x[1] - - self.activation_func = glu - else: - self.activation_func = self.config.activation_func - self.activation_recompute = ( - self.config.recompute_granularity == 'selective' - and "moe_act" in self.config.recompute_modules - ) - if self.activation_recompute and (self.config.fp8 or self.config.fp4): - raise ValueError( - "moe_act recompute for fp8 or fp4 cannot work with the legacy GroupedMLP." - ) - - @jit_fuser - def activation_func_with_probs(x, probs): - dtype = x.dtype - res = self.activation_func(x) * probs - return res.to(dtype) - - self.activation_func_with_probs = activation_func_with_probs - - self.ep_group = pg_collection.ep - # use pg_collection.expt_tp_group as tensor parallel group in this module. - self.tp_group = pg_collection.expt_tp - # use pg_collection.expt_dp_group as data parallel group in this module. - self.dp_group = pg_collection.expt_dp - # How many feature each rank holds for fc1 and fc2, respectively. - tp_size = self.tp_group.size() - tp_rank = self.tp_group.rank() - - fc1_output_size = self.config.moe_ffn_hidden_size * self.num_local_experts - if config.gated_linear_unit: - # Project to 4h. If using swiglu double the output width, - # see https://arxiv.org/pdf/2002.05202.pdf - fc1_output_size *= 2 - fc1_output_size_per_partition = divide(fc1_output_size, tp_size) - - fc2_input_size = self.config.moe_ffn_hidden_size * self.num_local_experts - fc2_input_size_per_partition = divide(fc2_input_size, tp_size) - - # Note: The current kernel implementations of grouped_gemm - # does not support transposition with CUTLASS grouped GEMM - # (https://github.com/fanshiqing/grouped_gemm/blob/main/csrc/grouped_gemm.cu#L355-L358) - # and as a result we avoid allocate the transpose of weights. - # Initialize weight. - if config.use_cpu_initialization: - self.weight1 = Parameter( - torch.empty( - self.config.hidden_size, - fc1_output_size_per_partition, - dtype=config.params_dtype, - ) - ) - self.weight2 = Parameter( - torch.empty( - fc2_input_size_per_partition, self.config.hidden_size, dtype=config.params_dtype - ) - ) - if config.perform_initialization: - _initialize_affine_weight_cpu( - self.weight1, - self.config.hidden_size, - fc1_output_size, - fc1_output_size_per_partition, - partition_dim=1, - init_method=config.init_method, - params_dtype=config.params_dtype, - rank=tp_rank, - world_size=tp_size, - ) - _initialize_affine_weight_cpu( - self.weight2, - fc2_input_size, - self.config.hidden_size, - fc2_input_size_per_partition, - partition_dim=0, - init_method=config.output_layer_init_method, - params_dtype=config.params_dtype, - rank=tp_rank, - world_size=tp_size, - ) - else: - self.weight1 = Parameter( - torch.empty( - self.config.hidden_size, - fc1_output_size_per_partition, - device=torch.cuda.current_device(), - dtype=config.params_dtype, - ) - ) - self.weight2 = Parameter( - torch.empty( - fc2_input_size_per_partition, - self.config.hidden_size, - device=torch.cuda.current_device(), - dtype=config.params_dtype, - ) - ) - if config.perform_initialization: - _initialize_affine_weight_gpu( - self.weight1, config.init_method, partition_dim=1, is_expert=True - ) - _initialize_affine_weight_gpu( - self.weight2, config.output_layer_init_method, partition_dim=0, is_expert=True - ) - setattr(self.weight1, 'allreduce', not self.expert_parallel) - setattr(self.weight2, 'allreduce', not self.expert_parallel) - - def remove_extra_states_check(self, incompatible_keys): - """ - Remove _extra_state from unexpected keys. - These keys are for dist ckpt compatibility with SequentialMLP. - """ - keys = deepcopy(incompatible_keys.unexpected_keys) - for key in keys: - if '_extra_state' in key: - incompatible_keys.unexpected_keys.remove(key) - - self.register_load_state_dict_post_hook(remove_extra_states_check) - - def forward( - self, - permuted_local_hidden_states: torch.Tensor, - tokens_per_expert: torch.Tensor, - permuted_probs: torch.Tensor, - ): - """Forward step of the GroupedMLP.""" - assert self.config.bf16, "Currently GroupedMLP for MoE only supports bf16." - if self.activation_recompute: - self.activation_checkpoint = tensor_parallel.CheckpointWithoutOutput() - - if self.config.moe_apply_probs_on_input: - assert ( - self.config.moe_router_topk == 1 - ), "`moe_apply_probs_on_input` only works with `moe_router_topk`=1." - original_dtype = permuted_local_hidden_states.dtype - permuted_local_hidden_states = ( - permuted_probs.unsqueeze(-1) * permuted_local_hidden_states - ) - permuted_local_hidden_states = permuted_local_hidden_states.to(original_dtype) - # Probs already applied, so reset to 1. - permuted_probs = torch.ones_like(permuted_probs) - - if permuted_local_hidden_states.nelement() != 0: - # Reshape the weights for the grouped GEMMs. - w1 = self.weight1.view(self.num_local_experts, self.config.hidden_size, -1) - w2 = self.weight2.view(self.num_local_experts, -1, self.config.hidden_size) - - fc1_output = gg.ops.gmm( - permuted_local_hidden_states, w1, tokens_per_expert, trans_b=False - ) - if self.activation_recompute: - intermediate_parallel = self.activation_checkpoint.checkpoint( - self.activation_func_with_probs, fc1_output, permuted_probs.unsqueeze(-1) - ) - fc2_output = gg.ops.gmm(intermediate_parallel, w2, tokens_per_expert, trans_b=False) - self.activation_checkpoint.discard_output_and_register_recompute(fc2_output) - else: - intermediate_parallel = self.activation_func_with_probs( - fc1_output, permuted_probs.unsqueeze(-1) - ) - fc2_output = gg.ops.gmm(intermediate_parallel, w2, tokens_per_expert, trans_b=False) - else: - # No token is allocated for local experts. - assert torch.count_nonzero(tokens_per_expert) == 0 - - # Make sure params of experts still have gradients even given zero tokens. - w1 = self.weight1.view(self.config.hidden_size, -1) - w2 = self.weight2.view(-1, self.config.hidden_size) - h = torch.matmul(permuted_local_hidden_states, w1) - if self.activation_recompute: - h = self.activation_checkpoint.checkpoint( - self.activation_func_with_probs, h, permuted_probs.unsqueeze(-1) - ) - fc2_output = torch.matmul(h, w2) - self.activation_checkpoint.discard_output_and_register_recompute(fc2_output) - else: - h = self.activation_func_with_probs(h, permuted_probs.unsqueeze(-1)) - fc2_output = torch.matmul(h, w2) - - return fc2_output, None - - def sharded_state_dict(self, prefix='', sharded_offsets=(), metadata=None): - """ - Maps local expert to global experts. - The sharded_state_dict for the weight parts are compatible with the SequentialMLP, - whereas the optimizer states are not due to the limitation from weight transposing. - That is, for finetuning scenario, the checkpoint is compatible with the SequentialMLP. - - When `singleton_local_shards` metadata flag is True, experts are broken down into - separate tensors and stored under separate global keys. Additionally, similarly to MLP, - layers with GLU activations are broken down into separate `w` and `v` tensors. - """ - singleton_local_shards = (metadata or {}).get('singleton_local_shards', False) - sharded_state_dict = {} - ep_size = self.ep_group.size() - ep_rank = self.ep_group.rank() - tp_size = self.tp_group.size() - tp_rank = self.tp_group.rank() - dp_rank = self.dp_group.rank() - num_global_experts = ep_size * self.num_local_experts - local_expert_indices_offset = ep_rank * self.num_local_experts - - prepend_axis_num = len(sharded_offsets) - replica_id = (0, 0, dp_rank) - - local_ffn_dim_size = ( - self.weight2.numel() // self.num_local_experts // self.config.hidden_size - ) - - def _break_into_individual_experts( - experts_ten: torch.Tensor, - key: str, - tp_offset: Tuple[int, int, int], - replica_id: ReplicaId, - ): - """Breaks experts into individual tensors and stores them under separate global keys""" - experts_state = [] - assert len(experts_ten) == self.num_local_experts, ( - experts_ten.shape, - self.num_local_experts, - ) - for local_expert_idx, expert_ten in enumerate(experts_ten): - global_expert_idx = local_expert_indices_offset + local_expert_idx - expert_key = key.replace( - f'{prefix}experts.', f'{prefix}experts.{global_expert_idx}.' - ) - experts_state.append( - ShardedTensor.from_rank_offsets( - expert_key, - expert_ten.contiguous(), - *sharded_offsets, - tp_offset, - replica_id=replica_id, - prepend_axis_num=prepend_axis_num, - ) - ) - return experts_state - - @torch.no_grad() - def sh_ten_build_fn( - key: str, - t: torch.Tensor, - replica_id: ReplicaId, - flattened_range: Optional[slice], - tp_axis: int, - with_glu: bool, - ): - # TODO: write a generic implementation to cover both cases with and without GLU - if tp_axis == 1: - # weight1 - if with_glu: - last_dim_size = local_ffn_dim_size * 2 - else: - last_dim_size = local_ffn_dim_size - real_shape = (self.num_local_experts, self.config.hidden_size, last_dim_size) - elif tp_axis == 0: - # weight2 - real_shape = (self.num_local_experts, local_ffn_dim_size, self.config.hidden_size) - assert with_glu == False - else: - raise ValueError("tp_axis should be 0 or 1.") - if flattened_range is None: - # weights - t = t.view(real_shape).transpose(-1, -2) - # change tp_axis due to the transposing - tp_axis = 1 - tp_axis - if with_glu: - assert tp_axis == 0, tp_axis - if singleton_local_shards: - w_tensor, v_tensor = torch.chunk(t, 2, -2) - w_key = f'{key}_w' - v_key = f'{key}_v' - sub_states = { - 'singleton_local_shards': LocalNonpersistentObject(True), - 'data': { - 'w': _break_into_individual_experts( - w_tensor, - w_key, - (prepend_axis_num, tp_rank, tp_size), - replica_id, - ), - 'v': _break_into_individual_experts( - v_tensor, - v_key, - (prepend_axis_num, tp_rank, tp_size), - replica_id, - ), - }, - } - else: - local_tensors = torch.chunk(t, 2, -2) - sub_states = [ - ShardedTensor.from_rank_offsets( - key, - local_tensors[0].contiguous(), - *sharded_offsets, - (prepend_axis_num, ep_rank, ep_size), - (prepend_axis_num + 1, tp_rank, tp_size * 2), - replica_id=replica_id, - prepend_axis_num=prepend_axis_num, - ), - ShardedTensor.from_rank_offsets( - key, - local_tensors[1].contiguous(), - *sharded_offsets, - (prepend_axis_num, ep_rank, ep_size), - (prepend_axis_num + 1, tp_size + tp_rank, tp_size * 2), - replica_id=replica_id, - prepend_axis_num=prepend_axis_num, - ), - ] - else: - if singleton_local_shards: - sub_states = { - 'singleton_local_shards': LocalNonpersistentObject(True), - 'data': _break_into_individual_experts( - t, key, (prepend_axis_num + tp_axis, tp_rank, tp_size), replica_id - ), - } - else: - sub_states = ShardedTensor.from_rank_offsets( - key, - t.contiguous(), - *sharded_offsets, - (prepend_axis_num, ep_rank, ep_size), - (prepend_axis_num + 1 + tp_axis, tp_rank, tp_size), - replica_id=replica_id, - prepend_axis_num=prepend_axis_num, - ) - return sub_states # pylint: disable=possibly-used-before-assignment - - @torch.no_grad() - def sh_ten_merge_fn(sub_state_dict, tp_axis: int, with_glu: bool): - if tp_axis == 1: - # weight1 - weight_shape = (self.config.hidden_size, -1) - elif tp_axis == 0: - # weight2 - weight_shape = (-1, self.config.hidden_size) - assert with_glu == False - else: - raise ValueError("tp_axis should be 0 or 1.") - if isinstance(sub_state_dict, dict): - assert sub_state_dict['singleton_local_shards'] - if with_glu: - assert isinstance(sub_state_dict['data'], dict) - sub_state_dict = torch.cat( - ( - torch.stack(sub_state_dict['data']['w']), - torch.stack(sub_state_dict['data']['v']), - ), - dim=-2, - ) - else: - assert isinstance(sub_state_dict['data'], list) - sub_state_dict = torch.stack(sub_state_dict['data']) - else: - if with_glu: - sub_state_dict = torch.cat(sub_state_dict, -2) - return sub_state_dict.transpose(-1, -2).reshape(weight_shape) - - state_dict = self.state_dict(prefix='', keep_vars=True) - for name, tensor in state_dict.items(): - if name == 'weight1': - tp_axis = 1 - with_glu = self.config.gated_linear_unit - wkey = f'{prefix}experts.linear_fc1.weight' - else: - tp_axis = 0 - with_glu = False - wkey = f'{prefix}experts.linear_fc2.weight' - - this_replica_id = list(copy.deepcopy(replica_id)) - - sharded_state_dict[f'{prefix}{name}'] = ShardedTensorFactory( - wkey, - tensor, - partial(sh_ten_build_fn, tp_axis=tp_axis, with_glu=with_glu), - partial(sh_ten_merge_fn, tp_axis=tp_axis, with_glu=with_glu), - tuple(this_replica_id), - ) - - replica_id = (0, tp_rank, dp_rank) - # Add fake _extra_state to be compatible with SequentialMLP - for expert_local_idx in range(self.num_local_experts): - expert_global_idx = local_expert_indices_offset + expert_local_idx - if singleton_local_shards: - expert_sharded_offsets = sharded_offsets - else: - expert_sharded_offsets = ( - *sharded_offsets, - (len(sharded_offsets), expert_global_idx, num_global_experts), - ) - for mod in ['linear_fc1', 'linear_fc2']: - if singleton_local_shards: - expert_key = f'{prefix}experts.{expert_global_idx}.{mod}._extra_state' - else: - expert_key = f'{prefix}experts.{mod}._extra_state' - sharded_state_dict[f'{prefix}expert{expert_global_idx}.{mod}._extra_state'] = ( - make_sharded_object_for_checkpoint( - None, expert_key, expert_sharded_offsets, replica_id - ) - ) - - return sharded_state_dict - - def backward_dw(self): - """Performs backward pass for weight gradients in Experts. - Empty implementation for compatibility with SequentialMLP and TEGroupedMLP. - """ - pass - - class GroupedLinearFc1Interface(Protocol): """Interface for linear_fc1 module in TEGroupedMLP.""" diff --git a/megatron/core/transformer/moe/grouped_gemm_util.py b/megatron/core/transformer/moe/grouped_gemm_util.py deleted file mode 100644 index 5dd344816bd..00000000000 --- a/megatron/core/transformer/moe/grouped_gemm_util.py +++ /dev/null @@ -1,22 +0,0 @@ -# Copyright (c) 2023, NVIDIA CORPORATION. All rights reserved. - -try: - import grouped_gemm -except ImportError: - grouped_gemm = None - - -def grouped_gemm_is_available(): - """Check if grouped_gemm is available.""" - return grouped_gemm is not None - - -def assert_grouped_gemm_is_available(): - """Assert that grouped_gemm is available.""" - assert grouped_gemm_is_available(), ( - "Grouped GEMM is not available. Please run " - "`pip install git+https://github.com/fanshiqing/grouped_gemm@v1.1.4`." - ) - - -ops = grouped_gemm.ops if grouped_gemm_is_available() else None diff --git a/megatron/core/transformer/transformer_config.py b/megatron/core/transformer/transformer_config.py index 559f4226af2..642af8415d3 100644 --- a/megatron/core/transformer/transformer_config.py +++ b/megatron/core/transformer/transformer_config.py @@ -721,10 +721,6 @@ class TransformerConfig(ModelParallelConfig): GEMM feature introduced since CUTLASS 2.8 (https://github.com/fanshiqing/grouped_gemm). """ - moe_use_legacy_grouped_gemm: bool = False - """Use legacy GroupedMLP rather than TEGroupedMLP. - Note: The legacy one will be deprecated soon.""" - moe_aux_loss_coeff: Union[float, List[float]] = 0.0 """Scaling coefficient for the aux loss. A starting value of 1e-2 is recommended. If a list of load balancing types is provided for `moe_router_load_balancing_type`, @@ -2090,9 +2086,6 @@ def __post_init__(self): assert ( self.overlap_moe_expert_parallel_comm ), 'overlap_moe_expert_parallel_comm must be enabled when enabling delay_wgrad_compute' - assert ( - not self.moe_use_legacy_grouped_gemm - ), 'delay_wgrad_compute is not supported with legacy groupedgemm implementation' if self.cuda_graph_impl == "transformer_engine": assert is_te_min_version("2.10.0"), ( 'TE version >= 2.10.0 is required for delay_wgrad_compute with ' diff --git a/megatron/core/transformer/transformer_layer.py b/megatron/core/transformer/transformer_layer.py index 73be3496876..d032cb7d6b5 100644 --- a/megatron/core/transformer/transformer_layer.py +++ b/megatron/core/transformer/transformer_layer.py @@ -360,7 +360,7 @@ def __init__( additional_mlp_kwargs = {} # import here to avoid circular import from megatron.core.extensions.transformer_engine import TEFusedMLP - from megatron.core.transformer.moe.experts import GroupedMLP, SequentialMLP, TEGroupedMLP + from megatron.core.transformer.moe.experts import SequentialMLP, TEGroupedMLP from megatron.core.transformer.moe.moe_layer import MoELayer # MLP expects tp_group but MoELayer expects pg_collection to be passed in. @@ -368,7 +368,7 @@ def __init__( # The conditional below is to make the logic explicit # if submodules.mlp is not a ModuleSpec,we dont have to handle passing additional kwargs if isinstance(submodules.mlp, ModuleSpec): - if submodules.mlp.module in (MoELayer, GroupedMLP, TEGroupedMLP, SequentialMLP): + if submodules.mlp.module in (MoELayer, TEGroupedMLP, SequentialMLP): additional_mlp_kwargs["pg_collection"] = pg_collection # Pass is_mtp_layer flag to MoELayer to distinguish MTP MoE layers. if submodules.mlp.module == MoELayer: diff --git a/megatron/core/utils.py b/megatron/core/utils.py index 2094d60ae68..42b82f8d00f 100644 --- a/megatron/core/utils.py +++ b/megatron/core/utils.py @@ -333,8 +333,11 @@ def get_te_version_str(): return version("transformer-engine") global _te_version - if _te_version is None and HAVE_TE: - _te_version = PkgVersion(get_te_version_str()) + if _te_version is None: + if HAVE_TE: + _te_version = PkgVersion(get_te_version_str()) + else: + _te_version = PkgVersion("0.0.0") return _te_version diff --git a/megatron/training/arguments.py b/megatron/training/arguments.py index 6c1cf20c5ac..91e26af99c6 100644 --- a/megatron/training/arguments.py +++ b/megatron/training/arguments.py @@ -406,19 +406,6 @@ def validate_args(args, defaults={}): "installed. See https://github.com/fzyzcjy/torch_memory_saver." ) - # When using different EP sizes for inference and training (EP refit), the legacy - # GroupedMLP is not supported. Only SequentialMLP or TEGroupedMLP can be used. - if ( - args.rl_inference_expert_model_parallel_size is not None - and args.rl_inference_expert_model_parallel_size != args.expert_model_parallel_size - ): - assert not args.moe_use_legacy_grouped_gemm, ( - "Legacy GroupedMLP (--moe-use-legacy-grouped-gemm) is not supported when using " - "different expert parallelism sizes for inference and training. " - "Use SequentialMLP (default when --moe-grouped-gemm is not set) or " - "TEGroupedMLP (--moe-grouped-gemm without --moe-use-legacy-grouped-gemm)." - ) - args.grpo_samples_per_iteration = args.grpo_prompts_per_step * args.grpo_group_size num_generated_samples_per_inference_iteration = ( args.grpo_samples_per_iteration * args.grpo_iterations) @@ -1545,7 +1532,6 @@ def validate_args(args, defaults={}): assert args.moe_latent_size > 0, "MoE latent projection dimension has to be greater than zero." assert args.num_experts is not None, "MoE latent projections are applicable only for MoE models." assert not args.use_legacy_models, "MoE latent projections are only supported for mcore models." - assert not args.moe_use_legacy_grouped_gemm, "MoE latent projection is not supported yet with legacy grouped GEMM." if args.tiktoken_special_tokens and not args.tokenizer_special_tokens: warn_rank_0( diff --git a/tests/unit_tests/dist_checkpointing/models/test_moe_experts.py b/tests/unit_tests/dist_checkpointing/models/test_moe_experts.py index 5d8153f49d7..f341af44754 100644 --- a/tests/unit_tests/dist_checkpointing/models/test_moe_experts.py +++ b/tests/unit_tests/dist_checkpointing/models/test_moe_experts.py @@ -21,7 +21,6 @@ from megatron.core.tensor_parallel.random import model_parallel_cuda_manual_seed from megatron.core.transformer.mlp import MLPSubmodules from megatron.core.transformer.moe.experts import ( - GroupedMLP, SequentialMLP, TEGroupedMLP, TEGroupedMLPSubmodules, @@ -56,9 +55,7 @@ def initialize_expert_layer(seed, glu=True, expert_type='sequential', fp8=False, ) default_config_kwargs.update(**config_kwargs) transformer_config = TransformerConfig(**default_config_kwargs) - if expert_type == 'grouped': - model = GroupedMLP(num_local_experts, transformer_config, pg_collection) - elif expert_type == 'te_grouped': + if expert_type == 'te_grouped': layer_submodules = get_gpt_layer_with_transformer_engine_submodules( num_experts=num_moe_experts, moe_grouped_gemm=True ) @@ -98,14 +95,13 @@ def initialize_expert_layer(seed, glu=True, expert_type='sequential', fp8=False, ) else: raise ValueError( - 'expert_type can only be one of ["sequential", "te_sequential", "grouped",' - ' "te_grouped"]' + 'expert_type can only be one of ["sequential", "te_sequential", "te_grouped"]' ) return model -expert_type = ['sequential', 'grouped'] -src_dest_expert_type = [('sequential', 'grouped'), ('grouped', 'sequential')] +expert_type = ['sequential'] +src_dest_expert_type = [] if is_te_min_version("1.7.0.dev0"): expert_type.append('te_sequential') src_dest_expert_type.append(('sequential', 'te_sequential')) diff --git a/tests/unit_tests/models/test_gpt_model.py b/tests/unit_tests/models/test_gpt_model.py index 9518ad8668d..336ff0552b0 100644 --- a/tests/unit_tests/models/test_gpt_model.py +++ b/tests/unit_tests/models/test_gpt_model.py @@ -123,7 +123,6 @@ def test_get_mlp_module_spec_interface(): "num_experts": inspect.Parameter.POSITIONAL_OR_KEYWORD, "moe_grouped_gemm": inspect.Parameter.POSITIONAL_OR_KEYWORD, "fp8": inspect.Parameter.POSITIONAL_OR_KEYWORD, - "moe_use_legacy_grouped_gemm": inspect.Parameter.POSITIONAL_OR_KEYWORD, "use_te_op_fuser": inspect.Parameter.POSITIONAL_OR_KEYWORD, } @@ -132,7 +131,6 @@ def test_get_mlp_module_spec_interface(): "num_experts": None, "moe_grouped_gemm": False, "fp8": None, - "moe_use_legacy_grouped_gemm": False, "use_te_op_fuser": False, } diff --git a/tests/unit_tests/models/test_gpt_model_quantization.py b/tests/unit_tests/models/test_gpt_model_quantization.py index e993c9be8d2..73f1e20558f 100644 --- a/tests/unit_tests/models/test_gpt_model_quantization.py +++ b/tests/unit_tests/models/test_gpt_model_quantization.py @@ -267,7 +267,6 @@ def test_kitchen_config_resolution_moe(self) -> None: moe_router_load_balancing_type="sinkhorn", moe_router_topk=1, moe_grouped_gemm=True, - moe_use_legacy_grouped_gemm=False, num_layers=2, hidden_size=12, num_attention_heads=4, diff --git a/tests/unit_tests/models/test_mamba_moe_model.py b/tests/unit_tests/models/test_mamba_moe_model.py index 98c6ac63e0e..f8d1cde7028 100644 --- a/tests/unit_tests/models/test_mamba_moe_model.py +++ b/tests/unit_tests/models/test_mamba_moe_model.py @@ -188,7 +188,6 @@ "moe_token_dispatcher_type": "alltoall", "moe_token_drop_policy": "probs", "moe_token_dropping": False, - "moe_use_legacy_grouped_gemm": False, "moe_z_loss_coeff": None, "moe_enable_routing_replay": False, "mrope_section": None, diff --git a/tests/unit_tests/pipeline_parallel/test_fine_grained_activation_offloading.py b/tests/unit_tests/pipeline_parallel/test_fine_grained_activation_offloading.py index 558c6934a0c..524b22e2de5 100644 --- a/tests/unit_tests/pipeline_parallel/test_fine_grained_activation_offloading.py +++ b/tests/unit_tests/pipeline_parallel/test_fine_grained_activation_offloading.py @@ -73,7 +73,6 @@ def _build_gpt_model( transformer_layer_spec=get_gpt_layer_with_transformer_engine_spec( num_experts=num_experts, moe_grouped_gemm=num_experts is not None, - moe_use_legacy_grouped_gemm=False, multi_latent_attention=is_mla, ), vocab_size=vocab_size, @@ -435,10 +434,7 @@ def _build_overlap_moe_gpt( GPTModel( config=transformer_config, transformer_layer_spec=get_gpt_layer_with_transformer_engine_spec( - num_experts=num_experts, - moe_grouped_gemm=True, - moe_use_legacy_grouped_gemm=False, - multi_latent_attention=is_mla, + num_experts=num_experts, moe_grouped_gemm=True, multi_latent_attention=is_mla ), vocab_size=vocab_size, max_sequence_length=seq_length, diff --git a/tests/unit_tests/transformer/moe/test_grouped_mlp.py b/tests/unit_tests/transformer/moe/test_grouped_mlp.py index 67543168480..58ca0cf1e3f 100644 --- a/tests/unit_tests/transformer/moe/test_grouped_mlp.py +++ b/tests/unit_tests/transformer/moe/test_grouped_mlp.py @@ -9,7 +9,6 @@ get_gpt_layer_with_transformer_engine_submodules, ) from megatron.core.transformer.module import Float16Module -from megatron.core.transformer.moe import grouped_gemm_util as gg from megatron.core.transformer.moe.experts import TEGroupedMLP from megatron.core.transformer.moe.moe_layer import MoELayer from megatron.core.transformer.transformer_config import TransformerConfig @@ -18,208 +17,6 @@ from megatron.training.initialize import _set_random_seed from tests.unit_tests.test_utilities import Utils -DEVICE_CAPABILITY = None -if torch.cuda.is_available(): - DEVICE_CAPABILITY = torch.cuda.get_device_capability() - - -@pytest.mark.skipif(is_te_min_version("1.9.0.dev0"), reason="Switch to TEGroupedMLP when TE>1.9.") -class TestParallelGroupedMLP: - - def setup_method(self, method, use_cpu_initialization=False, swiglu=True): - print("============") - print( - "Test for use_cpu_initilization={} and swiglu={}.".format( - use_cpu_initialization, swiglu - ) - ) - print("============") - Utils.initialize_model_parallel(1, 1) - num_layers = 1 # 2 - self.hidden_size = ( - 16 # must be an multiple of 16, otherwise trigger CUTLASS misaligned issue - ) - self.num_experts = 2 - self.gated_linear_unit = swiglu - self.activation_func = F.silu if swiglu else F.gelu - self.use_cpu_initialization = use_cpu_initialization - - tf_config = TransformerConfig( - num_layers=num_layers, - hidden_size=self.hidden_size, - num_attention_heads=4, - num_moe_experts=self.num_experts, - use_cpu_initialization=self.use_cpu_initialization, - add_bias_linear=False, - gated_linear_unit=self.gated_linear_unit, - activation_func=self.activation_func, - bias_activation_fusion=False, - bf16=True, - params_dtype=torch.bfloat16, - moe_router_load_balancing_type="sinkhorn", - moe_router_topk=1, - ) - - self.fc1_ffn_hidden_size = tf_config.ffn_hidden_size - self.fc2_ffn_hidden_size = tf_config.ffn_hidden_size - # If using swiglu double the output width, see https://arxiv.org/pdf/2002.05202.pdf - if self.gated_linear_unit: - self.fc1_ffn_hidden_size *= 2 - - ## Vanilla sequential GEMM - # Set random seed for reproducability - _set_random_seed(seed_=123, data_parallel_random_init=False) - submodules = get_gpt_layer_local_submodules(self.num_experts, moe_grouped_gemm=False) - self.sequential_mlp = MoELayer(tf_config, submodules.mlp.submodules) - - self.args = parse_args(ignore_unknown_args=True) - self.args.bf16 = True - # Bias is not supported in grouped gemm currently, thus we disable the - # bias in the linear layer. - self.args.add_bias_linear = False - self.sequential_mlp = Float16Module(self.sequential_mlp.config, self.sequential_mlp).module - print("done intializing for sequential gemm") - - ## Grouped GEMM - _set_random_seed(seed_=123, data_parallel_random_init=False) - tf_config.moe_grouped_gemm = True - self.grouped_mlp = MoELayer( - tf_config, - get_gpt_layer_with_transformer_engine_submodules( - self.num_experts, moe_grouped_gemm=True - ).mlp.submodules, - ) - self.grouped_mlp = Float16Module(self.grouped_mlp.config, self.grouped_mlp).module - print("done intializing for grouped gemm") - - def teardown_method(self, method): - Utils.destroy_model_parallel() - - @pytest.mark.internal - def test_constructor(self): - assert isinstance(self.sequential_mlp, MoELayer) - assert isinstance(self.grouped_mlp, MoELayer) - - num_weights_smm = sum([p.numel() for p in self.sequential_mlp.parameters()]) - num_weights_gmm = sum([p.numel() for p in self.grouped_mlp.parameters()]) - - # For the same hyper-parm model configs except the `moe_grouped_gemm`, - # GroupedGEMM and sequential GEMMs should hold the same number of parms. - assert num_weights_smm == num_weights_gmm - # expected num weights: router linear weights+bias + MLP weights(no bias) of all experts - expected_num_weights = ( - self.hidden_size * self.num_experts - + self.hidden_size - * (self.fc1_ffn_hidden_size + self.fc2_ffn_hidden_size) - * self.num_experts - ) - assert num_weights_smm == expected_num_weights - - assert torch.equal(self.sequential_mlp.router.weight, self.grouped_mlp.router.weight) - - # weight1: [h, num_experts*4h] - # weight2: [num_experts*4h, h] - assert self.grouped_mlp.experts.weight1.shape[0] == self.hidden_size - assert ( - self.grouped_mlp.experts.weight1.shape[1] == self.num_experts * self.fc1_ffn_hidden_size - ) - if self.gated_linear_unit: - assert ( - self.grouped_mlp.experts.weight2.shape[0] - == self.num_experts * self.fc2_ffn_hidden_size - ) - assert self.grouped_mlp.experts.weight2.shape[1] == self.hidden_size - else: - assert ( - self.grouped_mlp.experts.weight1.shape == self.grouped_mlp.experts.weight2.t().shape - ) - - @pytest.mark.internal - def test_weight_init_value_the_same(self): - gmm_w1 = self.grouped_mlp.experts.weight1.view(self.num_experts, -1, self.hidden_size) - gmm_w2 = self.grouped_mlp.experts.weight2.view(self.num_experts, self.hidden_size, -1) - gmm_expert1_fc1 = gmm_w1[0] - gmm_expert1_fc2 = gmm_w2[0] - gmm_expert2_fc1 = gmm_w1[1] - gmm_expert2_fc2 = gmm_w2[1] - - smm_expert1_fc1 = self.sequential_mlp.experts.local_experts[0].linear_fc1.weight - smm_expert1_fc2 = self.sequential_mlp.experts.local_experts[0].linear_fc2.weight - smm_expert2_fc1 = self.sequential_mlp.experts.local_experts[1].linear_fc1.weight - smm_expert2_fc2 = self.sequential_mlp.experts.local_experts[1].linear_fc2.weight - - assert torch.equal(gmm_expert1_fc1, smm_expert1_fc1) - if not self.use_cpu_initialization: - assert torch.equal(gmm_expert1_fc2, smm_expert1_fc2) - # the param init value is not exactly the same between gmm and smm (refer to test_weight_init_value_the_same.) - # TODO: is it necessary to keep smm and gmm share exactly the same init params? - # assert torch.equal(gmm_expert2_fc1, smm_expert2_fc1) - if self.use_cpu_initialization: - assert torch.equal(gmm_expert2_fc2, smm_expert2_fc2) - - @pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available") - @pytest.mark.internal - @pytest.mark.skipif( - not DEVICE_CAPABILITY or DEVICE_CAPABILITY[0] < 8, - reason='GroupedGEMM kernels are not supported on this device.', - ) - def test_gpu_forward(self): - self.sequential_mlp.cuda() - self.grouped_mlp.cuda() - # [sequence length, batch size, hidden size] - seq_len = 3 # 32 - batch_size = 2 - hidden_states = torch.rand( - (seq_len, batch_size, self.sequential_mlp.config.hidden_size), dtype=torch.bfloat16 - ) - hidden_states = hidden_states.cuda() - output_smm, _ = self.sequential_mlp(hidden_states) - output_gmm, _ = self.grouped_mlp(hidden_states) - - # The following assert fails due to the param init value is not exactly - # the same between gmm and smm (refer to test_weight_init_value_the_same.) - # assert torch.equal(output_smm, output_gmm) - - @pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available") - @pytest.mark.internal - @pytest.mark.skipif( - not DEVICE_CAPABILITY or DEVICE_CAPABILITY[0] < 8, - reason='GroupedGEMM kernels are not supported on this device.', - ) - def test_gpu_forward_with_no_tokens_allocated(self): - """Test the case when no token is allocated for groupedGEMM kernels.""" - w1 = self.grouped_mlp.experts.weight1.view(self.num_experts, -1, self.hidden_size) - num_allocated_tokens = 0 - tokens_per_expert = torch.zeros(self.num_experts) - hidden_states = torch.rand((num_allocated_tokens, self.hidden_size), dtype=torch.bfloat16) - hidden_states = hidden_states.cuda() - try: - gg.ops.gmm(hidden_states, w1, tokens_per_expert, trans_b=False) - except Exception as e: - print("Expected error message from groupedGEMM:", e) - assert str(e) == "Input batch_sizes should not be all zeros!" - - @pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA not available") - @pytest.mark.internal - @pytest.mark.skipif( - not DEVICE_CAPABILITY or DEVICE_CAPABILITY[0] < 8, - reason='GroupedGEMM kernels are not supported on this device.', - ) - def test_gradient_with_no_tokens_allocated(self): - """Test that when no token is passed in, the parameters of the grouped MLP will also have gradients.""" - self.grouped_mlp.cuda() - num_allocated_tokens = 0 - tokens_per_expert = torch.zeros(self.num_experts) - hidden_states = torch.rand((num_allocated_tokens, self.hidden_size), dtype=torch.bfloat16) - hidden_states = hidden_states.cuda() - probs = torch.rand((num_allocated_tokens,), dtype=torch.float32) - probs = probs.cuda() - output_gmm, _ = self.grouped_mlp.experts( - hidden_states, tokens_per_expert=tokens_per_expert, permuted_probs=probs - ) - output_gmm.mean().backward() - assert self.grouped_mlp.experts.weight1.grad is not None - @pytest.mark.skipif( not is_te_min_version("1.9.0.dev0"), @@ -384,17 +181,3 @@ def test_gpu_forward_backward_with_no_tokens_allocated(self): for i in range(self.num_experts): assert getattr(self.grouped_mlp.experts.linear_fc1, f"weight{i}").grad is not None assert getattr(self.grouped_mlp.experts.linear_fc2, f"weight{i}").grad is not None - - -if __name__ == "__main__": - for use_cpu_unitilization in [True, False]: - for swiglu in [True, False]: - GMLP_test = TestParallelGroupedMLP() - GMLP_test.setup_method( - method=None, use_cpu_initialization=use_cpu_unitilization, swiglu=swiglu - ) - GMLP_test.test_constructor() - GMLP_test.test_weight_init_value_the_same() - GMLP_test.test_gpu_forward() - GMLP_test.test_gpu_forward_with_no_tokens_allocated() - GMLP_test.teardown_method(method=None)