Skip to content

Commit 4fdb04c

Browse files
committed
[PyTorch] Scope the quantized-param caching flag to its own graph capture
`make_graphed_callables(cache_quantized_params=True)` creates a process-global flag tensor that gates quantized weight updates, and never scopes it. The modules only check `fp8_graph_capturing()` before reading it, so any *later* capture in the same process picks up the leftover tensor and bakes it in as its own quantize noop flag. Nothing ever writes that flag for a callable graphed without caching, so it keeps whatever the earlier capture left there: if that was "skip", the second module silently reuses a stale quantized weight for the rest of training while its master weight keeps being updated. The flag tensor cannot be cleared or reallocated -- already-captured graphs bake in its address and replay fills it in place -- so gate the read instead. Track whether the capture in progress requested caching and hand out the tensor only then, clearing the scope once capture finishes. Affects every recipe, not a specific one. TE's own graph tests run each case in a fresh process, which is why this never showed up in CI. Signed-off-by: zhihaow6 <zhihaow6@illinois.edu>
1 parent bffde8f commit 4fdb04c

7 files changed

Lines changed: 108 additions & 33 deletions

File tree

tests/pytorch/test_cuda_graphs.py

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1209,3 +1209,69 @@ def test_make_graphed_callables_with_interleaved_pipeline_parallelism_reused_buf
12091209
)
12101210
assert_all_equal(outputs, graph_outputs)
12111211
assert_all_equal(weights, graph_weights)
1212+
1213+
1214+
@pytest.mark.skipif(not fp8_available, reason="FP8 is not supported")
1215+
@pytest.mark.parametrize("fp8_recipe", fp8_recipes, ids=recipe_id)
1216+
def test_make_graphed_callables_weight_caching_does_not_leak(
1217+
*,
1218+
fp8_recipe: recipe.Recipe,
1219+
dtype: torch.dtype = torch.bfloat16,
1220+
) -> None:
1221+
"""A capture with quantized-param caching must not affect later captures.
1222+
1223+
The flag tensor gating quantized weight updates is process-global, so a capture
1224+
without caching that picks it up bakes in a flag nothing ever writes, and then
1225+
silently reuses a stale quantized weight.
1226+
"""
1227+
reset_rng_states()
1228+
FP8GlobalStateManager.reset()
1229+
hidden, seqlen = 32, 32 # MXFP8 needs both dims divisible by 32
1230+
1231+
def build() -> torch.nn.Module:
1232+
torch.manual_seed(0)
1233+
model = Linear(hidden, hidden, bias=False, params_dtype=dtype, device="cuda")
1234+
for param in model.parameters():
1235+
param.grad = torch.empty_like(param)
1236+
return model
1237+
1238+
def data() -> torch.Tensor:
1239+
return torch.randn((seqlen, hidden), dtype=dtype, device="cuda")
1240+
1241+
def capture(model: torch.nn.Module, caching: bool) -> torch.nn.Module:
1242+
return make_graphed_callables(
1243+
model,
1244+
(data(),),
1245+
num_warmup_iters=3,
1246+
enabled=True,
1247+
recipe=fp8_recipe,
1248+
cache_quantized_params=caching,
1249+
)
1250+
1251+
def train(model: torch.nn.Module) -> List[torch.Tensor]:
1252+
outputs = []
1253+
optimizer = torch.optim.SGD(model.parameters(), lr=0.05)
1254+
for step in range(2):
1255+
optimizer.zero_grad(set_to_none=False)
1256+
for microbatch in range(2):
1257+
torch.manual_seed(step * 10 + microbatch)
1258+
with autocast(enabled=True, recipe=fp8_recipe):
1259+
out = model(data())
1260+
out.backward(torch.ones_like(out))
1261+
outputs.append(out.detach().clone())
1262+
optimizer.step()
1263+
return outputs
1264+
1265+
# Leave the global flag asking for the weight update to be skipped. Each
1266+
# microbatch needs its own autocast: the flag is only written for the first
1267+
# module of a context.
1268+
cached_model = capture(build(), caching=True)
1269+
for is_first_microbatch in (True, False):
1270+
with autocast(enabled=True, recipe=fp8_recipe):
1271+
out = cached_model(data(), is_first_microbatch=is_first_microbatch)
1272+
out.backward(torch.ones_like(out))
1273+
1274+
# A later capture without caching must still refresh its quantized weight.
1275+
graphed = train(capture(build(), caching=False))
1276+
ungraphed = train(build())
1277+
assert_all_equal(graphed, ungraphed)

transformer_engine/pytorch/graph.py

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -371,6 +371,9 @@ def _make_graphed_callables(
371371
consumed_sample_q[sample_keys].append(per_callable_fwd_idx)
372372
fwd_sample_qs[m_chunk] = fwd_sample_qs[m_chunk][num_consumed_samples:]
373373

374+
# Scope caching to this capture: the flag tensor below outlives it (replay fills
375+
# it in place), so a later capture must not bake in this one's leftover flag.
376+
FP8GlobalStateManager.set_caching_quantized_params(cache_quantized_params)
374377
if cache_quantized_params:
375378
# Initialize flag that controls FP8 weight updates
376379
FP8GlobalStateManager.set_skip_fp8_weight_update_tensor(False)
@@ -1609,23 +1612,26 @@ def call_func(self, *args, **kwargs):
16091612
else:
16101613
original_rng_states = torch.cuda.get_rng_state()
16111614

1612-
graphed_callables = _make_graphed_callables(
1613-
forward_funcs,
1614-
sample_args,
1615-
num_warmup_iters=num_warmup_iters,
1616-
allow_unused_input=allow_unused_input,
1617-
cache_quantized_params=cache_quantized_params,
1618-
sample_kwargs=sample_kwargs,
1619-
_order=_order,
1620-
_num_layers_per_chunk=_num_layers_per_chunk,
1621-
pool=pool,
1622-
retain_graph_in_backward=retain_graph_in_backward,
1623-
_reuse_graph_input_output_buffers=_reuse_graph_input_output_buffers,
1624-
clone_param_grads_on_return=clone_param_grads_on_return,
1625-
pre_warmup_hook=pre_warmup_hook,
1626-
post_warmup_hook=post_warmup_hook,
1627-
capture_time_hooks=capture_time_hooks,
1628-
)
1615+
try:
1616+
graphed_callables = _make_graphed_callables(
1617+
forward_funcs,
1618+
sample_args,
1619+
num_warmup_iters=num_warmup_iters,
1620+
allow_unused_input=allow_unused_input,
1621+
cache_quantized_params=cache_quantized_params,
1622+
sample_kwargs=sample_kwargs,
1623+
_order=_order,
1624+
_num_layers_per_chunk=_num_layers_per_chunk,
1625+
pool=pool,
1626+
retain_graph_in_backward=retain_graph_in_backward,
1627+
_reuse_graph_input_output_buffers=_reuse_graph_input_output_buffers,
1628+
clone_param_grads_on_return=clone_param_grads_on_return,
1629+
pre_warmup_hook=pre_warmup_hook,
1630+
post_warmup_hook=post_warmup_hook,
1631+
capture_time_hooks=capture_time_hooks,
1632+
)
1633+
finally:
1634+
FP8GlobalStateManager.set_caching_quantized_params(False)
16291635

16301636
# Ensures warmup does not affect numerics for ops such as dropout.
16311637
if graph_safe_rng_available():

transformer_engine/pytorch/module/grouped_linear.py

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1872,9 +1872,7 @@ def forward(
18721872
num_gemms = self.num_gemms
18731873

18741874
if FP8GlobalStateManager.fp8_graph_capturing():
1875-
skip_fp8_weight_update = (
1876-
FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor
1877-
)
1875+
skip_fp8_weight_update = FP8GlobalStateManager.get_skip_fp8_weight_update_tensor()
18781876
else:
18791877
skip_fp8_weight_update = None
18801878
if skip_fp8_weight_update is not None:
@@ -1893,9 +1891,7 @@ def forward(
18931891
)
18941892

18951893
if FP8GlobalStateManager.fp8_graph_capturing():
1896-
skip_fp8_weight_update = (
1897-
FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor
1898-
)
1894+
skip_fp8_weight_update = FP8GlobalStateManager.get_skip_fp8_weight_update_tensor()
18991895
else:
19001896
skip_fp8_weight_update = None
19011897
if skip_fp8_weight_update is not None:

transformer_engine/pytorch/module/layernorm_linear.py

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1706,9 +1706,7 @@ def forward(
17061706
debug = self.is_debug_iter()
17071707

17081708
if FP8GlobalStateManager.fp8_graph_capturing():
1709-
skip_fp8_weight_update = (
1710-
FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor
1711-
)
1709+
skip_fp8_weight_update = FP8GlobalStateManager.get_skip_fp8_weight_update_tensor()
17121710
else:
17131711
skip_fp8_weight_update = None
17141712
if skip_fp8_weight_update is not None:

transformer_engine/pytorch/module/layernorm_mlp.py

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2291,9 +2291,7 @@ def forward(
22912291
debug = self.is_debug_iter()
22922292

22932293
if FP8GlobalStateManager.fp8_graph_capturing():
2294-
skip_fp8_weight_update = (
2295-
FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor
2296-
)
2294+
skip_fp8_weight_update = FP8GlobalStateManager.get_skip_fp8_weight_update_tensor()
22972295
else:
22982296
skip_fp8_weight_update = None
22992297
if skip_fp8_weight_update is not None:

transformer_engine/pytorch/module/linear.py

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1881,9 +1881,7 @@ def forward(
18811881
debug = self.is_debug_iter()
18821882

18831883
if FP8GlobalStateManager.fp8_graph_capturing():
1884-
skip_fp8_weight_update = (
1885-
FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor
1886-
)
1884+
skip_fp8_weight_update = FP8GlobalStateManager.get_skip_fp8_weight_update_tensor()
18871885
else:
18881886
skip_fp8_weight_update = None
18891887
if skip_fp8_weight_update is not None:

transformer_engine/pytorch/quantization.py

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -400,6 +400,9 @@ class FP8GlobalState:
400400
default_factory=dict
401401
)
402402
skip_fp8_weight_update_tensor: Optional[torch.Tensor] = None
403+
# Whether the graph capture in progress caches quantized params. Scopes the flag
404+
# tensor above, which outlives capture and would otherwise leak into later ones.
405+
caching_quantized_params: bool = False
403406

404407

405408
class FP8GlobalStateManager:
@@ -418,9 +421,19 @@ def set_skip_fp8_weight_update_tensor(cls, skip: bool) -> None:
418421
)
419422
cls.quantization_state.skip_fp8_weight_update_tensor.fill_(skip)
420423

424+
@classmethod
425+
def set_caching_quantized_params(cls, caching: bool) -> None:
426+
"""Mark whether the graph capture in progress caches quantized params"""
427+
cls.quantization_state.caching_quantized_params = caching
428+
421429
@classmethod
422430
def get_skip_fp8_weight_update_tensor(cls) -> Optional[torch.Tensor]:
423-
"""Get the skip fp8 weight update tensor"""
431+
"""Get the skip fp8 weight update tensor
432+
433+
``None`` unless the capture in progress caches quantized params.
434+
"""
435+
if not cls.quantization_state.caching_quantized_params:
436+
return None
424437
return cls.quantization_state.skip_fp8_weight_update_tensor
425438

426439
@classmethod

0 commit comments

Comments
 (0)