Skip to content

Commit dc980e4

Browse files
committed
[PyTorch] Harden multi-device FP8 cleanup and ops paths
Signed-off-by: SuperGoodGame <985236470@qq.com>
1 parent a60384c commit dc980e4

7 files changed

Lines changed: 190 additions & 62 deletions

File tree

‎tests/pytorch/test_multi_device_fp8.py‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import torch
1414

1515
import transformer_engine.pytorch as te
16+
import transformer_engine.pytorch.ops as te_ops
1617
from transformer_engine.common.recipe import DelayedScaling, Format
1718
from transformer_engine.pytorch.quantization import FP8GlobalStateManager
1819

@@ -109,6 +110,43 @@ def test_module_off_current_device():
109110
_assert_state_evolved(module)
110111

111112

113+
def test_prepare_forward_exception_restores_current_device(monkeypatch):
114+
"""A failed prepare must not leak the temporary CUDA device guard."""
115+
assert torch.cuda.current_device() == 0
116+
module = te.Linear(16, 16, bias=False, params_dtype=DT, device="cuda:1")
117+
inp = torch.randn(2, 16, device="cuda:1", dtype=DT)
118+
119+
def fail_init(*args, **kwargs):
120+
raise RuntimeError("injected prepare_forward failure")
121+
122+
monkeypatch.setattr(module, "init_fp8_metadata", fail_init)
123+
with pytest.raises(RuntimeError, match="injected prepare_forward failure"):
124+
module.prepare_forward(inp)
125+
126+
assert torch.cuda.current_device() == 0
127+
assert module._forward_device_guards == []
128+
129+
130+
def test_basic_operation_device_awareness():
131+
"""The standalone ops API must use the op's device for state and execution."""
132+
recipe = DelayedScaling(fp8_format=Format.HYBRID)
133+
assert torch.cuda.current_device() == 0
134+
with te.quantized_model_init(enabled=True, recipe=recipe):
135+
op = te_ops.basic.BasicLinear(512, 1024, device="cuda:1", dtype=DT)
136+
inp = torch.randn(128, 512, device="cuda:1", dtype=DT, requires_grad=True)
137+
with te.autocast(enabled=True, recipe=DelayedScaling(fp8_format=Format.HYBRID)):
138+
out = op(inp)
139+
out.sum().backward()
140+
torch.cuda.synchronize()
141+
assert out.device == torch.device("cuda:1")
142+
assert inp.grad is not None and inp.grad.device == torch.device("cuda:1")
143+
assert torch.cuda.current_device() == 0
144+
for mode in ("forward", "backward"):
145+
state = op._fp8_metas[mode][FP8GlobalStateManager.get_meta_tensor_key(mode == "forward")]
146+
assert state.scale.device == torch.device("cuda:1")
147+
assert state.amax_history.device == torch.device("cuda:1")
148+
149+
112150
def test_two_modules_on_different_devices_one_autocast():
113151
"""Two modules on different devices inside one autocast (#3124 case C)."""
114152
torch.manual_seed(1234)

‎transformer_engine/pytorch/module/base.py‎

Lines changed: 70 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -1105,9 +1105,7 @@ def adjust_amax_history_length(self, length: int, fwd: Optional[bool] = None) ->
11051105
qstate.global_amax_buffer[buffer_key][pos] = self.fp8_meta[
11061106
meta_key
11071107
].amax_history[0]
1108-
qstate.global_amax_devices[buffer_key].add(
1109-
self.fp8_meta[meta_key].amax_history.device
1110-
)
1108+
FP8GlobalStateManager._refresh_global_amax_devices(buffer_key)
11111109

11121110
def _recipe_state_device(self) -> torch.device:
11131111
"""Device the module's quantization state must be allocated on.
@@ -1127,6 +1125,14 @@ def _recipe_state_device(self) -> torch.device:
11271125
return tensor.device
11281126
return torch.device("cuda", torch.cuda.current_device())
11291127

1128+
def _release_forward_device_guard(self) -> None:
1129+
"""Release the most recently entered forward device guard, if any."""
1130+
if not self._forward_device_guards:
1131+
return
1132+
guard = self._forward_device_guards.pop()
1133+
if guard is not None:
1134+
guard.__exit__(None, None, None)
1135+
11301136
def set_meta_tensor(self, fwd: bool, recipe: Recipe) -> None:
11311137
"""Init scales and amaxes for fwd | bwd."""
11321138
fp8_meta_tensor_key = "scaling_fwd" if fwd else "scaling_bwd"
@@ -1632,68 +1638,81 @@ def prepare_forward(
16321638
guard = torch.cuda.device(inp.device)
16331639
guard.__enter__()
16341640
self._forward_device_guards.append(guard)
1641+
nvtx_pushed = False
16351642

1636-
self.fast_setattr(
1637-
"allow_different_data_and_param_types", allow_different_data_and_param_types
1638-
)
1639-
self.fast_setattr("forwarded_at_least_once", True)
1640-
1641-
# Activation recomputation is used and this is the second forward phase.
1642-
if self.fp8 and in_fp8_activation_recompute_phase():
1643-
delayed_scaling_recipe = _has_delayed_scaling_state(self.fp8_meta)
1644-
FP8GlobalStateManager.get_old_fp8_meta_tensors_for_recompute(self.fp8_meta)
1645-
else:
1646-
if not inp.is_cuda:
1647-
raise RuntimeError(
1648-
f"TransformerEngine needs CUDA. Got input on device: {inp.device}"
1649-
)
1643+
try:
1644+
self.fast_setattr(
1645+
"allow_different_data_and_param_types", allow_different_data_and_param_types
1646+
)
1647+
self.fast_setattr("forwarded_at_least_once", True)
16501648

1651-
if self.tp_size > 1:
1652-
if not self.tp_group_initialized:
1649+
# Activation recomputation is used and this is the second forward phase.
1650+
if self.fp8 and in_fp8_activation_recompute_phase():
1651+
delayed_scaling_recipe = _has_delayed_scaling_state(self.fp8_meta)
1652+
FP8GlobalStateManager.get_old_fp8_meta_tensors_for_recompute(self.fp8_meta)
1653+
else:
1654+
if not inp.is_cuda:
16531655
raise RuntimeError(
1654-
"Tensor parallel group not initialized. Call "
1655-
"set_tensor_parallel_group() before forward pass when tp_size > 1."
1656+
f"TransformerEngine needs CUDA. Got input on device: {inp.device}"
16561657
)
16571658

1658-
self.set_activation_dtype(inp)
1659-
self.init_fp8_metadata(num_gemms=num_gemms)
1660-
self._check_weight_tensor_recipe_correspondence()
1661-
1662-
delayed_scaling_recipe = self.fp8 and _has_delayed_scaling_state(self.fp8_meta)
1663-
if delayed_scaling_recipe:
1664-
if self.sequence_parallel:
1665-
assert (
1666-
self.fp8_meta["recipe"].custom() or self.fp8_meta["recipe"].reduce_amax
1667-
), (
1668-
"Amax reduction across tensor parallel group is "
1669-
"necessary when using sequence parallelism with FP8."
1670-
)
1659+
if self.tp_size > 1:
1660+
if not self.tp_group_initialized:
1661+
raise RuntimeError(
1662+
"Tensor parallel group not initialized. Call "
1663+
"set_tensor_parallel_group() before forward pass when tp_size > 1."
1664+
)
1665+
1666+
self.set_activation_dtype(inp)
1667+
self.init_fp8_metadata(num_gemms=num_gemms)
1668+
self._check_weight_tensor_recipe_correspondence()
1669+
1670+
delayed_scaling_recipe = self.fp8 and _has_delayed_scaling_state(self.fp8_meta)
1671+
if delayed_scaling_recipe:
1672+
if self.sequence_parallel:
1673+
assert (
1674+
self.fp8_meta["recipe"].custom() or self.fp8_meta["recipe"].reduce_amax
1675+
), (
1676+
"Amax reduction across tensor parallel group is "
1677+
"necessary when using sequence parallelism with FP8."
1678+
)
16711679

1672-
if not FP8GlobalStateManager.fp8_graph_capturing():
1673-
FP8GlobalStateManager.add_fp8_tensors_to_global_buffer(self.fp8_meta)
1680+
if not FP8GlobalStateManager.fp8_graph_capturing():
1681+
FP8GlobalStateManager.add_fp8_tensors_to_global_buffer(self.fp8_meta)
16741682

1675-
# Activation recomputation is used and this is the first forward phase.
1676-
if self.training and is_fp8_activation_recompute_enabled():
1677-
FP8GlobalStateManager.copy_forward_fp8_meta_tensors_for_recompute(self.fp8_meta)
1683+
# Activation recomputation is used and this is the first forward phase.
1684+
if self.training and is_fp8_activation_recompute_enabled():
1685+
FP8GlobalStateManager.copy_forward_fp8_meta_tensors_for_recompute(
1686+
self.fp8_meta
1687+
)
16781688

1679-
nvtx_range_push(self.__class__.__name__ + " forward")
1680-
if not allow_non_contiguous and not inp.is_contiguous():
1681-
inp = inp.contiguous()
1682-
return inp
1689+
nvtx_range_push(self.__class__.__name__ + " forward")
1690+
nvtx_pushed = True
1691+
if not allow_non_contiguous and not inp.is_contiguous():
1692+
inp = inp.contiguous()
1693+
return inp
1694+
except BaseException:
1695+
try:
1696+
if nvtx_pushed:
1697+
nvtx_range_pop()
1698+
finally:
1699+
self._release_forward_device_guard()
1700+
raise
16831701

16841702
def end_forward(self):
16851703
"""
16861704
Required to be called at the end of the forward function to properly handle
16871705
DelayedScaling metadata handling and the NVTX ranges.
16881706
"""
1689-
delayed_scaling_recipe = self.fp8 and _has_delayed_scaling_state(self.fp8_meta)
1690-
if delayed_scaling_recipe and self.fp8 and in_fp8_activation_recompute_phase():
1691-
FP8GlobalStateManager.restore_fp8_meta_tensors(self.fp8_meta)
1692-
nvtx_range_pop()
1693-
if self._forward_device_guards:
1694-
guard = self._forward_device_guards.pop()
1695-
if guard is not None:
1696-
guard.__exit__(None, None, None)
1707+
try:
1708+
try:
1709+
delayed_scaling_recipe = self.fp8 and _has_delayed_scaling_state(self.fp8_meta)
1710+
if delayed_scaling_recipe and self.fp8 and in_fp8_activation_recompute_phase():
1711+
FP8GlobalStateManager.restore_fp8_meta_tensors(self.fp8_meta)
1712+
finally:
1713+
nvtx_range_pop()
1714+
finally:
1715+
self._release_forward_device_guard()
16971716

16981717
@contextmanager
16991718
def prepare_forward_ctx(

‎transformer_engine/pytorch/ops/basic/basic_linear.py‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,10 @@ def __init__(
146146
# Initialize recipe state if needed for natively quantized weight
147147
self._with_quantized_weight: bool = FP8GlobalStateManager.with_fp8_parameters()
148148
if self._with_quantized_weight:
149-
self.reset_recipe_state(recipe=FP8GlobalStateManager.get_fp8_recipe())
149+
self.reset_recipe_state(
150+
recipe=FP8GlobalStateManager.get_fp8_recipe(),
151+
device=device if device.type == "cuda" else None,
152+
)
150153

151154
# Initialize parameters if needed
152155
weight = torch.empty(
@@ -333,7 +336,10 @@ def reset_parameters(self) -> None:
333336
columnwise=torch.is_grad_enabled(),
334337
)
335338
quantizer.internal = False
336-
with torch.no_grad():
339+
device_context = (
340+
torch.cuda.device(weight.device) if weight.is_cuda else contextlib.nullcontext()
341+
)
342+
with device_context, torch.no_grad():
337343
weight = quantizer(weight)
338344

339345
# Save updated parameter
@@ -361,8 +367,10 @@ def pre_fuser_forward(self, *, requires_grad: bool) -> None:
361367
weight_quantizer.set_usage(rowwise=True, columnwise=requires_grad)
362368
grad_output_quantizer.set_usage(rowwise=True, columnwise=columnwise_usage)
363369

364-
def reset_recipe_state(self, *, recipe: Optional[Recipe]) -> None:
365-
super().reset_recipe_state(recipe=recipe)
370+
def reset_recipe_state(
371+
self, *, recipe: Optional[Recipe], device: Optional[torch.device] = None
372+
) -> None:
373+
super().reset_recipe_state(recipe=recipe, device=device)
366374

367375
# Configure input/grad output quantizers
368376
# Note: These tensors are only used internally. If there is no

‎transformer_engine/pytorch/ops/basic/grouped_linear.py‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -242,7 +242,10 @@ def __init__(
242242
# Initialize recipe state if needed for natively quantized weight
243243
self._with_quantized_weight: bool = FP8GlobalStateManager.with_fp8_parameters()
244244
if self._with_quantized_weight:
245-
self.reset_recipe_state(recipe=FP8GlobalStateManager.get_fp8_recipe())
245+
self.reset_recipe_state(
246+
recipe=FP8GlobalStateManager.get_fp8_recipe(),
247+
device=device if device.type == "cuda" else None,
248+
)
246249

247250
# RNG state tracker
248251
self._rng_state_tracker_function: Optional[Callable[[], CudaRNGStatesTracker]]
@@ -585,7 +588,10 @@ def _quantize_weights(
585588
return self._quantize_weights_mxfp8(weights, quantizers)
586589

587590
# Use quantizers to construct quantized weights
588-
with torch.no_grad():
591+
device_context = (
592+
torch.cuda.device(weights[0].device) if weights[0].is_cuda else contextlib.nullcontext()
593+
)
594+
with device_context, torch.no_grad():
589595
return [quantizer(weight) for quantizer, weight in zip(quantizers, weights)]
590596

591597
def _quantize_weights_mxfp8(
@@ -775,8 +781,10 @@ def pre_fuser_forward(self, *, requires_grad: bool) -> None:
775781
weight_quantizer.set_usage(rowwise=True, columnwise=requires_grad)
776782
grad_output_quantizer.set_usage(rowwise=True, columnwise=weight_requires_grad)
777783

778-
def reset_recipe_state(self, *, recipe: Optional[Recipe]) -> None:
779-
super().reset_recipe_state(recipe=recipe)
784+
def reset_recipe_state(
785+
self, *, recipe: Optional[Recipe], device: Optional[torch.device] = None
786+
) -> None:
787+
super().reset_recipe_state(recipe=recipe, device=device)
780788

781789
for group_idx in range(self.num_groups):
782790
# Input/grad output quantizers use internal tensors

‎transformer_engine/pytorch/ops/fuser.py‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -266,6 +266,25 @@ def backward(
266266
func_ctx: Any,
267267
grad_output: torch.Tensor,
268268
*grad_extra_outputs: torch.Tensor,
269+
) -> tuple[Optional[torch.Tensor], ...]:
270+
"""Run backward under the output tensor's CUDA device context."""
271+
guard = None
272+
if grad_output.is_cuda and grad_output.device.index != torch.cuda.current_device():
273+
guard = torch.cuda.device(grad_output.device)
274+
guard.__enter__()
275+
try:
276+
return _OperationFuserAutogradFunction._backward_impl(
277+
func_ctx, grad_output, *grad_extra_outputs
278+
)
279+
finally:
280+
if guard is not None:
281+
guard.__exit__(None, None, None)
282+
283+
@staticmethod
284+
def _backward_impl(
285+
func_ctx: Any,
286+
grad_output: torch.Tensor,
287+
*grad_extra_outputs: torch.Tensor,
269288
) -> tuple[Optional[torch.Tensor], ...]:
270289
"""Backward pass"""
271290

@@ -697,6 +716,23 @@ def __call__(
697716
*extra_inputs: torch.Tensor,
698717
basic_op_kwargs: Optional[list[dict[str, Any]]] = None,
699718
) -> torch.Tensor | tuple[torch.Tensor, ...]:
719+
"""Apply this fuser under the input tensor's CUDA device context."""
720+
guard = None
721+
if input.is_cuda and input.device.index != torch.cuda.current_device():
722+
guard = torch.cuda.device(input.device)
723+
guard.__enter__()
724+
try:
725+
return self._call_impl(input, *extra_inputs, basic_op_kwargs=basic_op_kwargs)
726+
finally:
727+
if guard is not None:
728+
guard.__exit__(None, None, None)
729+
730+
def _call_impl(
731+
self,
732+
input: torch.Tensor, # pylint: disable=redefined-builtin
733+
*extra_inputs: torch.Tensor,
734+
basic_op_kwargs: Optional[list[dict[str, Any]]] = None,
735+
) -> torch.Tensor | tuple[torch.Tensor, ...]:
700736

701737
# Verify extra input count
702738
if len(extra_inputs) != self.num_extra_inputs:

‎transformer_engine/pytorch/ops/op.py‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -315,10 +315,21 @@ def get_grad_output_quantizer(self) -> Optional[Quantizer]:
315315
return self.get_quantizer("backward", 0)
316316
return None
317317

318+
def _recipe_state_device(self) -> torch.device:
319+
"""Device on which this operation's quantization state must be allocated."""
320+
for tensor in self.parameters(recurse=False):
321+
if tensor is not None and tensor.device.type == "cuda":
322+
return tensor.device
323+
for tensor in self.buffers(recurse=False):
324+
if tensor is not None and tensor.device.type == "cuda":
325+
return tensor.device
326+
return torch.device("cuda", torch.cuda.current_device())
327+
318328
def reset_recipe_state(
319329
self,
320330
*,
321331
recipe: Optional[Recipe],
332+
device: Optional[torch.device] = None,
322333
) -> None:
323334
"""Construct state for quantization recipe"""
324335

@@ -369,6 +380,7 @@ def reset_recipe_state(
369380
recipe,
370381
mode=mode,
371382
num_quantizers=num_quantizers,
383+
device=device if device is not None else self._recipe_state_device(),
372384
roles=roles,
373385
)
374386
fp8_meta_key = FP8GlobalStateManager.get_meta_tensor_key(
@@ -431,9 +443,7 @@ def reset_recipe_state(
431443
qstate.global_amax_history_buffer[buffer_key][
432444
pos
433445
] = recipe_state.amax_history
434-
qstate.global_amax_devices[buffer_key].add(
435-
recipe_state.amax_history.device
436-
)
446+
FP8GlobalStateManager._refresh_global_amax_devices(buffer_key)
437447

438448
# Add meta tensors to global buffer to participate in reduction
439449
for mode in ("forward", "backward"):

‎transformer_engine/pytorch/quantization.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -573,6 +573,15 @@ def add_fp8_tensors_to_global_buffer(
573573
fp8_meta[index_in_buffer].append(len(qstate.global_amax_buffer[key]) - 1)
574574
fp8_meta[index_in_buffer].append(key)
575575

576+
@classmethod
577+
def _refresh_global_amax_devices(cls, buffer_key: str) -> None:
578+
"""Rebuild the device index after registered amax tensors are replaced."""
579+
qstate = cls.quantization_state
580+
if buffer_key in qstate.global_amax_buffer:
581+
qstate.global_amax_devices[buffer_key] = {
582+
tensor.device for tensor in qstate.global_amax_buffer[buffer_key]
583+
}
584+
576585
@classmethod
577586
def is_fp8_enabled(cls) -> bool:
578587
"""Is FP8 enabled"""

0 commit comments

Comments
 (0)