Skip to content

Commit b3cb17b

Browse files
kamran-nvidiaclaude
andcommitted
fix(training): realign the distributed optimizer's parameter index with its group order
Megatron-Core's DistributedOptimizer records each parameter's position within its optimizer group in gradient-buffer order (_build_optimizer_group_ranges), but stores the group's shards as all FP32 shards followed by all 16-bit shards (_build_model_and_main_param_groups). When a group mixes FP32 parameters, for example ones marked with mark_keep_in_fp32, with 16-bit ones and a 16-bit buffer comes first, model_param_group_index_map points at other parameters' shards. Only optimizer checkpointing reads that map, so training is unaffected. Saving optimizer state fails in sharded_param_state_dp_reshardable with "AssertionError: (torch.Size([...]), start, end)", and a load could restore state onto the wrong parameters wherever the sizes happen to agree. Until Megatron-Core fixes the ordering, setup_optimizer rebuilds the map in the stored order for every DistributedOptimizer (including inside a ChainedOptimizer). Already aligned maps are left unchanged, so the step becomes a no-op once the upstream fix lands. Tests: stand-in optimizers cover the realignment, chained and non-distributed optimizers, and unknown parameters. A one-GPU test builds a real DDP model with a BF16 buffer ahead of an FP32 one, takes an Adam step, checks that each parameter's optimizer state matches its own range and values, and builds the dp_reshardable checkpoint state; with the realignment disabled it fails. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Kamran Jafari <kjafarisadeg@nvidia.com>
1 parent 9dfd9f5 commit b3cb17b

2 files changed

Lines changed: 180 additions & 0 deletions

File tree

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020

2121
import torch
2222
from megatron.core.optimizer import (
23+
DistributedOptimizer,
2324
MegatronOptimizer,
2425
OptimizerConfig,
2526
get_megatron_optimizer,
@@ -110,11 +111,59 @@ def setup_optimizer(
110111
pg_collection=pg_collection,
111112
)
112113

114+
_realign_distributed_optimizer_param_index(optimizer)
113115
scheduler = _get_scheduler(optimizer_config, scheduler_config, optimizer)
114116

115117
return optimizer, scheduler
116118

117119

120+
def _realign_distributed_optimizer_param_index(optimizer: MegatronOptimizer | None) -> int:
121+
"""Point each distributed optimizer's parameter index at the shard its group really holds.
122+
123+
Megatron-Core's ``DistributedOptimizer`` records every parameter's position within its optimizer group in
124+
gradient-buffer order, but stores the group's shards as all FP32 shards followed by all 16-bit shards. When a
125+
group mixes FP32 parameters (for example ones marked with ``mark_keep_in_fp32``) with 16-bit ones and a 16-bit
126+
buffer comes first, ``model_param_group_index_map`` points at other parameters' shards. Only optimizer
127+
checkpointing reads that map, so training is unaffected, but saving optimizer state fails in
128+
``sharded_param_state_dp_reshardable`` and loading could restore state onto the wrong parameters.
129+
130+
Rebuild the map in the stored order. Optimizers whose map already matches are left unchanged.
131+
132+
Args:
133+
optimizer: The optimizer returned by ``setup_optimizer``, possibly a ``ChainedOptimizer``.
134+
135+
Returns:
136+
The number of index entries that were corrected.
137+
"""
138+
corrected = 0
139+
for inner in getattr(optimizer, "chained_optimizers", None) or [optimizer]:
140+
if not isinstance(inner, DistributedOptimizer):
141+
continue
142+
index = getattr(inner, "model_param_group_index_map", None)
143+
fp32_groups = getattr(inner, "model_fp32_groups", None)
144+
float16_groups = getattr(inner, "model_float16_groups", None)
145+
if index is None or fp32_groups is None or float16_groups is None:
146+
continue
147+
stored = {}
148+
for group_index, (fp32_params, float16_params) in enumerate(zip(fp32_groups, float16_groups)):
149+
for group_order, param in enumerate([*fp32_params, *float16_params]):
150+
stored[param] = (group_index, group_order)
151+
if stored.keys() != index.keys():
152+
G_LOGGER.warning(
153+
"Distributed optimizer parameter groups do not match its index map; leaving it unchanged."
154+
)
155+
continue
156+
changed = sum(stored[param] != position for param, position in index.items())
157+
if changed:
158+
inner.model_param_group_index_map = stored
159+
corrected += changed
160+
if corrected:
161+
G_LOGGER.info(
162+
f"Realigned {corrected} distributed-optimizer parameter index entries with the FP32-first group order."
163+
)
164+
return corrected
165+
166+
118167
def _optimizer_state_is_fp32_adam(state_dict: Mapping[str, object]) -> bool:
119168
"""Return whether a state dict contains only FP32 Adam moment tensors."""
120169
states = state_dict.get("state")

‎tests/unit_tests/training/test_optim.py‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
from megatron.bridge.peft.lora import get_lora_plus_config_overrides
2626
from megatron.bridge.training.config import SchedulerConfig
2727
from megatron.bridge.training.optim import (
28+
_realign_distributed_optimizer_param_index,
2829
hybrid_optimizer_state_loading,
2930
memory_efficient_fp32_optimizer_state_loading,
3031
memory_efficient_precision_aware_optimizer_state_checkpointing,
@@ -1031,3 +1032,133 @@ def test_only_two_lora_name_globs_added(self):
10311032
name_keys = [k for k in overrides if getattr(k, "name", None) and not getattr(k, "attr", None)]
10321033
names = {k.name for k in name_keys}
10331034
assert {"*.linear_in.weight", "*.linear_out.weight"}.issubset(names)
1035+
1036+
1037+
def _stand_in_distributed_optimizer(fp32_groups, float16_groups, index):
1038+
"""A DistributedOptimizer carrying only the attributes the realignment reads."""
1039+
from megatron.core.optimizer import DistributedOptimizer
1040+
1041+
optimizer = DistributedOptimizer.__new__(DistributedOptimizer)
1042+
optimizer.model_fp32_groups = fp32_groups
1043+
optimizer.model_float16_groups = float16_groups
1044+
optimizer.model_param_group_index_map = dict(index)
1045+
return optimizer
1046+
1047+
1048+
class TestRealignDistributedOptimizerParamIndex:
1049+
"""The parameter index must follow the FP32-first order in which MCore stores each group's shards."""
1050+
1051+
def _mixed_group(self):
1052+
bf16_a, bf16_b = torch.zeros(4, dtype=torch.bfloat16), torch.zeros(8, dtype=torch.bfloat16)
1053+
fp32 = torch.zeros(2)
1054+
# Gradient-buffer order (BF16 buffer first), as MCore records it.
1055+
index = {bf16_a: (0, 0), bf16_b: (0, 1), fp32: (0, 2)}
1056+
return bf16_a, bf16_b, fp32, index
1057+
1058+
def test_mixed_dtype_group_is_reindexed_fp32_first(self):
1059+
bf16_a, bf16_b, fp32, index = self._mixed_group()
1060+
optimizer = _stand_in_distributed_optimizer([[fp32]], [[bf16_a, bf16_b]], index)
1061+
1062+
assert _realign_distributed_optimizer_param_index(optimizer) == 3
1063+
assert optimizer.model_param_group_index_map == {fp32: (0, 0), bf16_a: (0, 1), bf16_b: (0, 2)}
1064+
# A second pass finds nothing left to correct.
1065+
assert _realign_distributed_optimizer_param_index(optimizer) == 0
1066+
1067+
def test_aligned_index_is_left_unchanged(self):
1068+
bf16_a, bf16_b, fp32, _ = self._mixed_group()
1069+
aligned = {fp32: (0, 0), bf16_a: (0, 1), bf16_b: (0, 2)}
1070+
optimizer = _stand_in_distributed_optimizer([[fp32]], [[bf16_a, bf16_b]], aligned)
1071+
original = optimizer.model_param_group_index_map
1072+
1073+
assert _realign_distributed_optimizer_param_index(optimizer) == 0
1074+
assert optimizer.model_param_group_index_map is original
1075+
1076+
def test_chained_optimizers_and_other_types(self):
1077+
bf16_a, bf16_b, fp32, index = self._mixed_group()
1078+
distributed = _stand_in_distributed_optimizer([[fp32]], [[bf16_a, bf16_b]], index)
1079+
chained = SimpleNamespace(chained_optimizers=[SimpleNamespace(), distributed])
1080+
1081+
assert _realign_distributed_optimizer_param_index(chained) == 3
1082+
assert distributed.model_param_group_index_map[fp32] == (0, 0)
1083+
assert _realign_distributed_optimizer_param_index(None) == 0
1084+
1085+
def test_index_with_unknown_parameters_is_left_unchanged(self):
1086+
bf16_a, bf16_b, fp32, index = self._mixed_group()
1087+
index[torch.zeros(1)] = (0, 3)
1088+
optimizer = _stand_in_distributed_optimizer([[fp32]], [[bf16_a, bf16_b]], index)
1089+
1090+
assert _realign_distributed_optimizer_param_index(optimizer) == 0
1091+
assert optimizer.model_param_group_index_map == index
1092+
1093+
@pytest.mark.run_only_on("gpu")
1094+
def test_real_distributed_optimizer_with_fp32_params_builds_checkpoint_state(self, tmp_path):
1095+
"""A BF16 buffer ahead of an FP32 one: the optimizer state must match every parameter's range."""
1096+
from megatron.core import parallel_state
1097+
from megatron.core.distributed import DistributedDataParallel, DistributedDataParallelConfig
1098+
from megatron.core.optimizer import DistributedOptimizer, get_megatron_optimizer
1099+
from megatron.core.transformer.module import MegatronModule, mark_keep_in_fp32
1100+
from megatron.core.transformer.transformer_config import TransformerConfig
1101+
1102+
if not torch.cuda.is_available():
1103+
pytest.skip("requires CUDA")
1104+
owns_dist = not torch.distributed.is_initialized()
1105+
if owns_dist:
1106+
torch.distributed.init_process_group(
1107+
"nccl", init_method=f"file://{tmp_path}/rendezvous", rank=0, world_size=1
1108+
)
1109+
owns_parallel = not parallel_state.model_parallel_is_initialized()
1110+
if owns_parallel:
1111+
parallel_state.initialize_model_parallel()
1112+
try:
1113+
config = TransformerConfig(
1114+
num_layers=1, hidden_size=16, num_attention_heads=2, bf16=True, params_dtype=torch.bfloat16
1115+
)
1116+
1117+
class _MixedPrecision(MegatronModule):
1118+
def __init__(self):
1119+
super().__init__(config)
1120+
# Declared first, so DDP allocates the BF16 buffer before the FP32 one.
1121+
self.proj = torch.nn.Linear(16, 16, bias=False, device="cuda", dtype=torch.bfloat16)
1122+
# Distinct sizes and values, so state attached to the wrong parameter cannot pass unnoticed.
1123+
self.norm = torch.nn.Parameter(torch.full((16,), 2.0, device="cuda", dtype=torch.bfloat16))
1124+
self.gate = mark_keep_in_fp32(torch.nn.Parameter(torch.full((1,), 0.5, device="cuda")))
1125+
1126+
def forward(self, inputs):
1127+
return (self.proj(inputs) * self.norm).float() * self.gate
1128+
1129+
model = DistributedDataParallel(
1130+
config, DistributedDataParallelConfig(use_distributed_optimizer=True), _MixedPrecision()
1131+
)
1132+
optimizer = get_megatron_optimizer(
1133+
OptimizerConfig(optimizer="adam", lr=1e-3, bf16=True, use_distributed_optimizer=True),
1134+
[model],
1135+
use_gloo_process_groups=False,
1136+
)
1137+
model(torch.randn(4, 16, device="cuda", dtype=torch.bfloat16)).sum().backward()
1138+
model.finish_grad_sync()
1139+
optimizer.step()
1140+
1141+
_realign_distributed_optimizer_param_index(optimizer)
1142+
1143+
inner = [
1144+
o for o in getattr(optimizer, "chained_optimizers", [optimizer]) if isinstance(o, DistributedOptimizer)
1145+
]
1146+
assert inner
1147+
for distributed in inner:
1148+
assert {p.dtype for p in distributed.model_param_group_index_map} == {torch.bfloat16, torch.float32}
1149+
for param in distributed.model_param_group_index_map:
1150+
param_range = distributed._get_model_param_range_map(param)["param"]
1151+
states = distributed._get_main_param_and_optimizer_states(param)
1152+
for key, tensor in states.items():
1153+
if tensor.dim() > 0:
1154+
assert tensor.numel() == param_range.end - param_range.start, key
1155+
# The main-parameter state must hold this parameter's own values (up to BF16 rounding).
1156+
own = param.detach().view(-1)[param_range.start : param_range.end].float()
1157+
torch.testing.assert_close(states["param"].float(), own, rtol=1e-2, atol=1e-2)
1158+
# The step that failed before the realignment: building the dp_reshardable checkpoint state.
1159+
distributed.sharded_param_state_dp_reshardable({})
1160+
finally:
1161+
if owns_parallel and parallel_state.model_parallel_is_initialized():
1162+
parallel_state.destroy_model_parallel()
1163+
if owns_dist and torch.distributed.is_initialized():
1164+
torch.distributed.destroy_process_group()

0 commit comments

Comments
 (0)