Skip to content

Commit 0adca65

Browse files
authored
Make partial_grad respect grad_comm_dtype in MFSDPv2 (#8185)
Signed-off-by: Xiaoda Zhang <xiaodaz@nvidia.com>
1 parent 3914483 commit 0adca65

2 files changed

Lines changed: 240 additions & 2 deletions

File tree

‎megatron/core/distributed/fsdp/src/megatron_fsdp/experimental/parameter_group.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,9 @@ class FsdpParameterGroup:
122122
# reduction created a smaller view (e.g. ZeRO-1 or HFSDP), the remaining main_grad
123123
# storage is stale and must be cleared before the next accumulation begins.
124124
_main_grad_is_stale: bool
125+
# dtype of the reduce-scatter input (partial_grad): MixedPrecisionPolicy.grad_comm_dtype,
126+
# or main_grad's dtype when unset.
127+
grad_comm_dtype: torch.dtype
125128
_unsharded_model_weight: "DBuffer | QuantizedDBuffer"
126129
_symm_mem_pool: torch.cuda.MemPool | None
127130
grad_divisor: int
@@ -350,10 +353,11 @@ def _initialize_buffers(
350353
self.main_grad = None
351354
self.pre_optimizer_main_grad = None
352355
self._main_grad_is_stale = False
356+
grad_dtype = mixed_precision_policy.main_grads_dtype or parameters[0].dtype
357+
self.grad_comm_dtype = mixed_precision_policy.grad_comm_dtype or grad_dtype
353358
if not self.requires_grad:
354359
return
355360

356-
grad_dtype = mixed_precision_policy.main_grads_dtype or parameters[0].dtype
357361
# Keep main_grad persistent for the initial implementation. For micro-batch
358362
# size 1, this allocation could be delayed until post_backward and then
359363
# eagerly deallocated right after optimizer.step(), avoiding main_grad
@@ -513,7 +517,7 @@ def allocate_partial_grad_buffer(self) -> DBuffer:
513517
mesh=self.mesh,
514518
placements=[Partial("avg")] * self.mesh.ndim,
515519
layout=self.main_weight.layout,
516-
dtype=grads[0].dtype,
520+
dtype=self.grad_comm_dtype,
517521
device=grads[0].device,
518522
)
519523

Lines changed: 234 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,234 @@
1+
# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
2+
3+
"""Gradient reduction dtype tests for MFSDP v2 (``MixedPrecisionPolicy.grad_comm_dtype``).
4+
5+
The reduce-scatter input (partial_grad) must be allocated in ``grad_comm_dtype``, or in the
6+
main-gradient dtype when it is unset, independently of the compute-gradient dtype, and
7+
training must still track an unsharded baseline in either reduction dtype.
8+
"""
9+
10+
import pytest
11+
import torch
12+
import torch.distributed as dist
13+
import transformer_engine.pytorch as te
14+
from torch import nn
15+
from torch.distributed.device_mesh import init_device_mesh
16+
from torch.distributed.tensor import Shard
17+
from transformer_engine.common.recipe import MXFP8BlockScaling
18+
19+
from megatron.core.distributed.fsdp.src.megatron_fsdp.experimental import (
20+
Placements,
21+
fully_shard,
22+
fully_shard_context,
23+
fully_shard_optimizer,
24+
)
25+
from megatron.core.distributed.fsdp.src.megatron_fsdp.experimental.parameter_group import (
26+
FsdpParameterGroup,
27+
)
28+
from megatron.core.distributed.fsdp.src.megatron_fsdp.experimental.quantized_dbuffer import (
29+
QuantizedDBuffer,
30+
)
31+
from megatron.core.distributed.fsdp.src.megatron_fsdp.mixed_precision import MixedPrecisionPolicy
32+
33+
34+
class TinyModel(nn.Module):
35+
"""Two separately shardable linear layers."""
36+
37+
def __init__(self) -> None:
38+
super().__init__()
39+
self.fc1 = nn.Linear(8, 16)
40+
self.fc2 = nn.Linear(16, 4)
41+
42+
def forward(self, x: torch.Tensor) -> torch.Tensor:
43+
"""Run the tiny model."""
44+
return self.fc2(torch.relu(self.fc1(x)))
45+
46+
47+
def _default_placements() -> Placements:
48+
return Placements(dp_axes=[0], parameter=[Shard(0)], gradient=[Shard(0)], optimizer=[Shard(0)])
49+
50+
51+
@pytest.mark.parametrize(
52+
("main_grads_dtype", "grad_comm_dtype", "expected_comm_dtype"),
53+
[
54+
# Default: reduce in the main-gradient dtype even though compute grads are bf16.
55+
(torch.float32, None, torch.float32),
56+
# Explicit override: reduce in bf16, then accumulate into fp32 main_grad.
57+
(torch.float32, torch.bfloat16, torch.bfloat16),
58+
# No main_grads_dtype: both main_grad and the reduction follow the parameter dtype.
59+
(None, None, torch.bfloat16),
60+
],
61+
)
62+
def test_partial_grad_buffer_follows_grad_comm_dtype(
63+
distributed_setup, monkeypatch, main_grads_dtype, grad_comm_dtype, expected_comm_dtype
64+
):
65+
"""partial_grad is allocated in grad_comm_dtype, defaulting to the main_grad dtype."""
66+
device = distributed_setup.device
67+
mesh = init_device_mesh(device.type, (distributed_setup.world_size,))
68+
model = nn.Linear(8, 8, device=device, dtype=torch.bfloat16)
69+
policy = MixedPrecisionPolicy(
70+
main_params_dtype=torch.float32,
71+
main_grads_dtype=main_grads_dtype,
72+
grad_comm_dtype=grad_comm_dtype,
73+
)
74+
with fully_shard_context(device=device):
75+
fully_shard(
76+
model, mesh=mesh, placements=_default_placements(), mixed_precision_policy=policy
77+
)
78+
79+
# Observe the dtype of the partial-grad buffer FSDP allocates during backward.
80+
allocated_dtypes: list[torch.dtype] = []
81+
original = FsdpParameterGroup.allocate_partial_grad_buffer
82+
83+
def spy(self: FsdpParameterGroup):
84+
buffer = original(self)
85+
allocated_dtypes.append(buffer.dtype)
86+
return buffer
87+
88+
monkeypatch.setattr(FsdpParameterGroup, "allocate_partial_grad_buffer", spy)
89+
90+
x = torch.randn(4, 8, device=device, dtype=torch.bfloat16)
91+
model(x).float().sum().backward()
92+
93+
assert allocated_dtypes == [expected_comm_dtype], allocated_dtypes
94+
for name, parameter in model.named_parameters():
95+
assert parameter.grad.dtype == (main_grads_dtype or torch.bfloat16), name
96+
97+
98+
@pytest.mark.launch_on_gb200
99+
@pytest.mark.skipif(
100+
torch.cuda.get_device_capability()[0] < 10,
101+
reason="MXFP8 requires Blackwell-or-newer CUDA hardware.",
102+
)
103+
@pytest.mark.parametrize(
104+
("main_grads_dtype", "grad_comm_dtype", "expected_comm_dtype"),
105+
[
106+
(torch.float32, None, torch.float32),
107+
(torch.float32, torch.bfloat16, torch.bfloat16),
108+
# MXFP8 weights are stored as uint8 planes, but their gradients are bf16, so the
109+
# fallback must follow the parameter's compute dtype, not the storage dtype.
110+
(None, None, torch.bfloat16),
111+
],
112+
)
113+
def test_mxfp8_partial_grad_buffer_follows_grad_comm_dtype(
114+
distributed_setup, monkeypatch, main_grads_dtype, grad_comm_dtype, expected_comm_dtype
115+
):
116+
"""The quantized (MXFP8) parameter group reduces its gradients in grad_comm_dtype too."""
117+
device = distributed_setup.device
118+
mesh = init_device_mesh(device.type, (distributed_setup.world_size,))
119+
recipe = MXFP8BlockScaling()
120+
with te.quantized_model_init(recipe=recipe, preserve_high_precision_init_val=True):
121+
torch.manual_seed(2026)
122+
model = te.Linear(64, 128, params_dtype=torch.bfloat16, device=device)
123+
policy = MixedPrecisionPolicy(
124+
main_params_dtype=torch.float32,
125+
main_grads_dtype=main_grads_dtype,
126+
grad_comm_dtype=grad_comm_dtype,
127+
)
128+
with fully_shard_context(device=device):
129+
fully_shard(
130+
model, mesh=mesh, placements=_default_placements(), mixed_precision_policy=policy
131+
)
132+
# The weight is an MXFP8 group (QuantizedDBuffer); the bias is a regular bf16 group.
133+
groups_by_kind = {
134+
isinstance(g.model_weight, QuantizedDBuffer): g for g in model.parameter_groups
135+
}
136+
assert set(groups_by_kind) == {True, False}
137+
for group in groups_by_kind.values():
138+
assert group.grad_comm_dtype == expected_comm_dtype
139+
assert group.main_grad.dtype == (main_grads_dtype or torch.bfloat16)
140+
141+
allocated_dtypes: list[torch.dtype] = []
142+
original = FsdpParameterGroup.allocate_partial_grad_buffer
143+
144+
def spy(self: FsdpParameterGroup):
145+
buffer = original(self)
146+
allocated_dtypes.append(buffer.dtype)
147+
return buffer
148+
149+
monkeypatch.setattr(FsdpParameterGroup, "allocate_partial_grad_buffer", spy)
150+
151+
# FP8 GEMMs need the token count divisible by 8 and the feature dim by 16.
152+
x = torch.randn(32, 64, device=device, dtype=torch.bfloat16)
153+
with te.autocast(recipe=recipe):
154+
output = model(x)
155+
output.float().sum().backward()
156+
157+
# One partial buffer per group (weight group and bias group).
158+
assert allocated_dtypes == [expected_comm_dtype] * 2, allocated_dtypes
159+
160+
161+
@pytest.mark.parametrize(
162+
("grad_comm_dtype", "atol"),
163+
[
164+
# fp32 reduction: the only difference from the baseline is the reduction order of
165+
# fp32 sums of bf16-representable values; observed bit-exact on 4 to 256 ranks.
166+
pytest.param(None, 1e-6, id="fp32"),
167+
# bf16 reduction rounds the partial sums of distinct per-rank gradients to bf16.
168+
# Measured max |loss diff| over 5 seeds: 4.5e-4 on 4 ranks, 1.0e-3 on 16 ranks (it
169+
# keeps growing with the DP size: 1.6e-2 on 256 ranks with a 256-wide hidden layer).
170+
pytest.param(torch.bfloat16, 3e-3, id="bf16"),
171+
],
172+
)
173+
def test_grad_comm_dtype_losses_match_baseline(distributed_setup, grad_comm_dtype, atol):
174+
"""Mixed-precision FSDP Adam tracks an unsharded mixed-precision Adam baseline.
175+
176+
The baseline keeps fp32 master weights and computes in bf16 autocast, i.e. the same
177+
scheme as FSDP with fp32 main params/grads, so the only remaining difference is the
178+
gradient reduction. Every rank feeds distinct data so the reduce-scatter really averages
179+
different gradients; the baseline all-reduces its gradients to match.
180+
"""
181+
rank = distributed_setup.rank
182+
world_size = distributed_setup.world_size
183+
device = distributed_setup.device
184+
if world_size < 2:
185+
pytest.skip("This test requires at least 2 ranks.")
186+
187+
mesh = init_device_mesh(device.type, (world_size,))
188+
torch.manual_seed(2026)
189+
model = TinyModel().to(device=device, dtype=torch.bfloat16)
190+
baseline = TinyModel().to(device=device, dtype=torch.float32)
191+
baseline.load_state_dict({name: value.float() for name, value in model.state_dict().items()})
192+
policy = MixedPrecisionPolicy(
193+
main_params_dtype=torch.float32,
194+
main_grads_dtype=torch.float32,
195+
grad_comm_dtype=grad_comm_dtype,
196+
)
197+
with fully_shard_context(device=device):
198+
for layer in (model.fc1, model.fc2):
199+
fully_shard(
200+
layer, mesh=mesh, placements=_default_placements(), mixed_precision_policy=policy
201+
)
202+
203+
baseline_optimizer = torch.optim.Adam(baseline.parameters(), lr=0.01)
204+
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
205+
fully_shard_optimizer(optimizer)
206+
207+
# Keep initialization identical, but exercise reduction of distinct rank-local gradients.
208+
torch.manual_seed(5678 + rank)
209+
x = torch.randn(3, 8, device=device, dtype=torch.bfloat16)
210+
target = torch.randn(3, 4, device=device, dtype=torch.bfloat16)
211+
212+
def train(module, module_optimizer, *, reduce_grads: bool) -> list[torch.Tensor]:
213+
losses = []
214+
for _ in range(3):
215+
module_optimizer.zero_grad()
216+
# The fp32 baseline computes in bf16 like the FSDP model's bf16 compute weights.
217+
with torch.autocast(device.type, dtype=torch.bfloat16, enabled=reduce_grads):
218+
output = module(x)
219+
loss = torch.nn.functional.mse_loss(output.float(), target.float())
220+
losses.append(loss.detach())
221+
loss.backward()
222+
if reduce_grads:
223+
# The unsharded baseline needs the same DP average as FSDP.
224+
for parameter in module.parameters():
225+
dist.all_reduce(parameter.grad, op=dist.ReduceOp.AVG)
226+
module_optimizer.step()
227+
return losses
228+
229+
baseline_losses = train(baseline, baseline_optimizer, reduce_grads=True)
230+
sharded_losses = train(model, optimizer, reduce_grads=False)
231+
232+
torch.testing.assert_close(
233+
torch.stack(sharded_losses), torch.stack(baseline_losses), rtol=0, atol=atol
234+
)

0 commit comments

Comments
 (0)