Skip to content

Commit 6ceaa45

Browse files
authored
feat(training): wire generalized tensor parallelism (#5569)
Signed-off-by: yaoyu-33 <yaoyu.094@gmail.com>
1 parent 47afa4f commit 6ceaa45

10 files changed

Lines changed: 467 additions & 16 deletions

File tree

‎src/megatron/bridge/training/config.py‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1423,6 +1423,17 @@ def validate(self) -> None:
14231423
if hasattr(self.model, "finalize"):
14241424
self.model.finalize()
14251425

1426+
from megatron.bridge.training.gtp import is_gtp_remat_active
1427+
1428+
if is_gtp_remat_active(self.model):
1429+
if self.dist.use_decentralized_pg:
1430+
raise ValueError(
1431+
"GTP is not supported with dist.use_decentralized_pg=True. "
1432+
"Set dist.use_decentralized_pg=False to use the standard MCore process-group runtime."
1433+
)
1434+
if self.ddp.average_in_collective:
1435+
raise ValueError("GTP requires ddp.average_in_collective=False.")
1436+
14261437
self.logger.finalize()
14271438
self.train.finalize()
14281439
self.scheduler.finalize()

‎src/megatron/bridge/training/eval.py‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
from megatron.bridge.training.callbacks import CallbackContext, CallbackManager, should_fire
3535
from megatron.bridge.training.config import ConfigContainer
3636
from megatron.bridge.training.forward_step_func_types import ForwardStepCallable
37+
from megatron.bridge.training.gtp import get_data_distribution_group
3738
from megatron.bridge.training.state import GlobalState
3839
from megatron.bridge.training.utils.mlflow_utils import _sanitize_mlflow_metrics
3940
from megatron.bridge.training.utils.pg_utils import get_pg_collection
@@ -141,7 +142,11 @@ def evaluate(
141142
eval_micro_batch_size = state.cfg.validation.eval_micro_batch_size
142143
# MegatronMIMO has heterogeneous per-module DP groups and intentionally owns
143144
# global-batch accounting through the container-level DP size.
144-
eval_data_parallel_size = state.cfg.data_parallel_size if is_multimodule else pg_collection.dp.size()
145+
eval_data_parallel_size = (
146+
state.cfg.data_parallel_size
147+
if is_multimodule
148+
else get_data_distribution_group(pg_collection, state.cfg.model).size()
149+
)
145150
eval_num_microbatches = eval_batch_size // (eval_micro_batch_size * eval_data_parallel_size)
146151

147152
if is_multimodule and not isinstance(p2p_communicator, MultiModulePipelineCommunicator):
@@ -290,7 +295,9 @@ def evaluate(
290295
if is_multimodule:
291296
dp_cp_group = pg_collection.get_language_model_collection().dp_cp
292297
else:
293-
dp_cp_group = pg_collection.dp_cp
298+
dp_cp_group = get_data_distribution_group(
299+
pg_collection, state.cfg.model, with_context_parallel=True
300+
)
294301

295302
for key in loss_dicts[0].keys():
296303
if key not in total_loss_dict:
Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
"""Generalized Tensor Parallelism helpers for the standard Bridge runtime."""
16+
17+
from typing import Any
18+
19+
import torch
20+
from megatron.core import parallel_state
21+
from megatron.core.process_groups_config import ProcessGroupCollection
22+
23+
24+
def get_transformer_config(model_config: Any) -> Any:
25+
"""Return the MCore transformer config nested in a Bridge model config."""
26+
model_fields = getattr(type(model_config), "__dataclass_fields__", {})
27+
if "transformer" in model_fields:
28+
return model_config.transformer
29+
return model_config
30+
31+
32+
def is_gtp_remat_active(model_config: Any) -> bool:
33+
"""Return whether dense or expert GTP weight rematerialization is enabled."""
34+
transformer_config = get_transformer_config(model_config)
35+
dense_size = getattr(transformer_config, "gtp_weight_remat_size", 1)
36+
expert_size = getattr(transformer_config, "expert_gtp_weight_remat_size", 1)
37+
return any(isinstance(size, int) and size > 1 for size in (dense_size, expert_size))
38+
39+
40+
def configure_gtp_remat(model_config: Any) -> None:
41+
"""Configure process-global GTP state before constructing model modules."""
42+
if not is_gtp_remat_active(model_config):
43+
return
44+
45+
transformer_config = get_transformer_config(model_config)
46+
from megatron.core.tensor_parallel import gtp_api
47+
48+
if not gtp_api.HAVE_GTP:
49+
raise RuntimeError("GTP requires TransformerEngine >= 2.19.")
50+
51+
gtp_api.configure_gtp_remat_from_recipe(
52+
fp4=transformer_config.fp4 is not None,
53+
fp8_recipe=transformer_config.fp8_recipe,
54+
fp8=transformer_config.fp8 is not None,
55+
calculate_per_token_loss=transformer_config.calculate_per_token_loss,
56+
)
57+
58+
59+
def classify_gtp_remat_chains(model: list[torch.nn.Module], model_config: Any) -> None:
60+
"""Classify all model chunks after distributed wrapping and before first forward."""
61+
if not is_gtp_remat_active(model_config):
62+
return
63+
64+
transformer_config = get_transformer_config(model_config)
65+
from megatron.core.tensor_parallel import gtp_api
66+
67+
gtp_api.classify_gtp_remat_chains(
68+
model,
69+
cuda_graph_modules=transformer_config.cuda_graph_modules,
70+
moe_shared_expert_overlap=transformer_config.moe_shared_expert_overlap,
71+
cuda_graph_impl=transformer_config.cuda_graph_impl,
72+
)
73+
74+
75+
def get_data_distribution_group(
76+
pg_collection: ProcessGroupCollection,
77+
model_config: Any,
78+
*,
79+
with_context_parallel: bool = False,
80+
) -> torch.distributed.ProcessGroup:
81+
"""Return the group spanning every rank that consumes distinct input data."""
82+
if not is_gtp_remat_active(model_config):
83+
return pg_collection.dp_cp if with_context_parallel else pg_collection.dp
84+
if with_context_parallel:
85+
return pg_collection.dp_cp_gtp_remat
86+
return parallel_state.get_data_parallel_group(with_gtp_remat=True)

‎src/megatron/bridge/training/initialize.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
from megatron.bridge.models.hybrid.hybrid_builder import HybridModelConfig
5151
from megatron.bridge.models.transformer_config import TransformerConfig, _set_moe_expert_tensor_parallel_default
5252
from megatron.bridge.training.config import ConfigContainer, DistributedInitConfig, RerunStateMachineConfig, RNGConfig
53+
from megatron.bridge.training.gtp import is_gtp_remat_active
5354
from megatron.bridge.training.utils.pg_utils import DistTrainProcessGroupCollection
5455
from megatron.bridge.utils.common_utils import (
5556
get_local_rank_preinit,
@@ -759,6 +760,18 @@ def _initialize_distributed(
759760
if dist_config.use_decentralized_pg or dist_config.distributed_backend == "nccl":
760761
raise RuntimeError("Cannot initialize parallel groups with no CUDA devices available (device_count=0)")
761762

763+
if dist_config.use_decentralized_pg and is_gtp_remat_active(model_config):
764+
raise NotImplementedError(
765+
"GTP is not supported with dist.use_decentralized_pg=True. "
766+
"Use the standard MCore process-group runtime by setting dist.use_decentralized_pg=False."
767+
)
768+
769+
if is_gtp_remat_active(model_config):
770+
from megatron.core.tensor_parallel.gtp_api import HAVE_GTP
771+
772+
if not HAVE_GTP:
773+
raise RuntimeError("GTP requires TransformerEngine >= 2.19.")
774+
762775
if dist_config.use_decentralized_pg:
763776
# Use HyperCommGrid to create local parallel groups passed through functions
764777
# instead of relying on mcore's global parallel state (mpu) variables.
@@ -812,6 +825,8 @@ def _initialize_distributed(
812825
expert_model_parallel_size=model_config.expert_model_parallel_size,
813826
num_distributed_optimizer_instances=num_distributed_optimizer_instances,
814827
expert_tensor_parallel_size=model_config.expert_tensor_parallel_size,
828+
gtp_remat_size=model_config.gtp_weight_remat_size,
829+
expert_gtp_remat_size=model_config.expert_gtp_weight_remat_size,
815830
distributed_timeout_minutes=dist_config.distributed_timeout_minutes,
816831
nccl_communicator_config_path=dist_config.nccl_communicator_config_path,
817832
order="tp-cp-ep-dp-pp" if not dist_config.use_tp_pp_dp_mapping else "tp-cp-ep-pp-dp",

‎src/megatron/bridge/training/setup.py‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,11 @@
5555
)
5656
from megatron.bridge.training.config import ConfigContainer
5757
from megatron.bridge.training.fsdp_compat import MEGATRON_FSDP_TYPES
58+
from megatron.bridge.training.gtp import (
59+
classify_gtp_remat_chains,
60+
configure_gtp_remat,
61+
get_data_distribution_group,
62+
)
5863
from megatron.bridge.training.initialize import initialize_megatron, set_jit_fusion_options
5964
from megatron.bridge.training.optim import (
6065
memory_efficient_fp32_optimizer_state_loading,
@@ -500,7 +505,7 @@ def modelopt_pre_wrap_hook(model):
500505
train_state=state.train_state,
501506
model_length=len(model),
502507
train_valid_test_datasets_provider=train_valid_test_datasets_provider,
503-
dp_group=pg_collection.dp,
508+
dp_group=get_data_distribution_group(pg_collection, cfg.model),
504509
eval_dp_group=state._eval_pgs.dp if state._eval_pgs is not None else None,
505510
)
506511
timers("train/valid/test-data-iterators-setup").stop()
@@ -584,10 +589,13 @@ def _register_setup_pre_wrap_hook(
584589
def _build_distributed_model(cfg: ConfigContainer, pg_collection: ProcessGroupCollection) -> list[MegatronModule]:
585590
"""Build distributed model from either ModelConfig or ModelProviderMixin."""
586591
model_config = cfg.model
592+
if not isinstance(model_config, ModelConfig):
593+
model_config.finalize()
594+
configure_gtp_remat(model_config)
587595
if isinstance(model_config, ModelConfig):
588596
builder_cls = model_config.get_builder_cls()
589597
builder = builder_cls(model_config)
590-
return builder.build_distributed_models(
598+
model = builder.build_distributed_models(
591599
pg_collection=pg_collection,
592600
ddp_config=cfg.ddp,
593601
overlap_param_gather_with_optimizer_step=cfg.optimizer.overlap_param_gather_with_optimizer_step,
@@ -596,15 +604,16 @@ def _build_distributed_model(cfg: ConfigContainer, pg_collection: ProcessGroupCo
596604
data_parallel_random_init=cfg.rng.data_parallel_random_init,
597605
)
598606
else:
599-
model_config.finalize()
600-
return model_config.provide_distributed_model(
607+
model = model_config.provide_distributed_model(
601608
ddp_config=cfg.ddp,
602609
use_megatron_fsdp=cfg.dist.use_megatron_fsdp,
603610
use_torch_fsdp2=cfg.dist.use_torch_fsdp2,
604611
overlap_param_gather_with_optimizer_step=cfg.optimizer.overlap_param_gather_with_optimizer_step,
605612
data_parallel_random_init=cfg.rng.data_parallel_random_init,
606613
pg_collection=pg_collection,
607614
)
615+
classify_gtp_remat_chains(model, model_config)
616+
return model
608617

609618

610619
def _update_model_config_funcs(

‎src/megatron/bridge/training/train.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@
7474
from megatron.bridge.training.eval import evaluate_and_print_results
7575
from megatron.bridge.training.forward_step_func_types import ForwardStepCallable
7676
from megatron.bridge.training.fsdp_compat import MEGATRON_FSDP_TYPES
77+
from megatron.bridge.training.gtp import get_data_distribution_group
7778
from megatron.bridge.training.initialize import destroy_global_state
7879
from megatron.bridge.training.nvrx_straggler import (
7980
check_nvrx_straggler_detection,
@@ -336,7 +337,8 @@ def train(
336337
start_iteration = global_state.train_state.step
337338
print_rank_0(f"Starting training loop at iteration {start_iteration}")
338339
p2p_communicator = P2PCommunicator(pp_group=pg_collection.pp, config=model_config)
339-
dp_size = pg_collection.dp.size()
340+
data_distribution_group = get_data_distribution_group(pg_collection, config.model)
341+
dp_size = data_distribution_group.size()
340342
# Anchor for interval-average throughput logging: training_log reports the FLOPS
341343
# performed over each logging interval as the delta of
342344
# floating_point_operations_so_far. Seed it with the current cumulative (0 fresh,
@@ -600,7 +602,7 @@ def train(
600602
global_state,
601603
data_parallel_size=dp_size,
602604
vp_size=config.model.virtual_pipeline_model_parallel_size,
603-
dp_group=pg_collection.dp,
605+
dp_group=data_distribution_group,
604606
include_vision_patch_stats=True,
605607
include_cross_attention_stats=hasattr(
606608
config.model, "_get_num_floating_point_operations_with_runtime_stats"
@@ -1004,7 +1006,7 @@ def train_step(
10041006
# there is one dict per microbatch. in new reporting, we average
10051007
# over the total number of tokens across the global batch.
10061008
val = torch.vstack(val).sum(dim=0)
1007-
dp_cp_group = pg_collection.dp_cp
1009+
dp_cp_group = get_data_distribution_group(pg_collection, cfg.model, with_context_parallel=True)
10081010
torch.distributed.all_reduce(val, group=dp_cp_group)
10091011
loss_reduced[key] = val[0] / val[1]
10101012
elif val[0].numel() == 1:
@@ -1587,7 +1589,7 @@ def _should_skip_and_handle_iteration(
15871589

15881590
# Update step and sample counters
15891591
global_state.train_state.step += 1
1590-
dp_size = pg_collection.dp.size()
1592+
dp_size = get_data_distribution_group(pg_collection, cfg.model).size()
15911593
batch_size = dp_size * cfg.train.micro_batch_size * get_num_microbatches()
15921594
global_state.train_state.consumed_train_samples += batch_size
15931595
global_state.train_state.skipped_train_samples += batch_size

0 commit comments

Comments
 (0)