diff --git a/tests/pytorch/test_fusible_ops.py b/tests/pytorch/test_fusible_ops.py index d4455c09d68..21d2672f105 100644 --- a/tests/pytorch/test_fusible_ops.py +++ b/tests/pytorch/test_fusible_ops.py @@ -4,7 +4,9 @@ from __future__ import annotations + from collections.abc import Iterable, Sequence +from unittest.mock import Mock import io import os import math @@ -22,7 +24,9 @@ from transformer_engine.pytorch.ops.basic.grouped_linear import ( OUTPUT_BUFFER_KEY, GRAD_INPUT_BUFFER_KEY, + is_op_fuser_grouped_tensor_path_supported, ) +from transformer_engine.pytorch.quantization import FP8GlobalStateManager from transformer_engine.pytorch.ops.fuser import OperationFuser from transformer_engine.pytorch._extra_state import UNSAFE_PICKLE_EXTRA_STATE_ENV @@ -5661,3 +5665,88 @@ def test_linear_inference_loop( quantization=quantization, recipe=recipe, ) + + +@pytest.mark.parametrize("split_alignment", [64, 128]) +def test_basic_grouped_pre_fuser_split_alignment(monkeypatch, split_alignment): + """Split fallback configures every expert before adjacent ops consume the quantizers.""" + fp8_recipe = transformer_engine.common.recipe.MXFP8BlockScaling() + if not is_op_fuser_grouped_tensor_path_supported(fp8_recipe, torch.bfloat16): + pytest.skip("Native MXFP8 grouped GEMM is unavailable.") + monkeypatch.delenv("NVTE_CUTEDSL_FUSED_GROUPED_MLP", raising=False) + FP8GlobalStateManager.reset() + op = te_ops.GroupedLinear( + 3, + 128, + 128, + bias=False, + dtype=torch.bfloat16, + device="cuda", + split_alignment=split_alignment, + ) + with te.autocast(enabled=True, recipe=fp8_recipe): + op.reset_recipe_state(recipe=fp8_recipe) + for quantizers in (op._input_quantizers, op._grad_output_quantizers): + for quantizer in quantizers: + quantizer.set_usage(rowwise=True, columnwise=False) + op.pre_fuser_forward(requires_grad=True) + for quantizers in (op._input_quantizers, op._grad_output_quantizers): + for index, quantizer in enumerate(quantizers): + assert quantizer.columnwise_usage == (split_alignment == 64 or index == 0) + + +@pytest.mark.parametrize( + "fp8_recipe", + [ + transformer_engine.common.recipe.MXFP8BlockScaling(), + transformer_engine.common.recipe.Float8CurrentScaling(), + ], +) +def test_grouped_discrete_weight_quantizers(fp8_recipe): + """Op usages may change between calls without disabling existing parameter usages.""" + if not is_op_fuser_grouped_tensor_path_supported(fp8_recipe, torch.bfloat16): + pytest.skip("Native grouped GEMM is unavailable.") + FP8GlobalStateManager.reset() + op = te_ops.GroupedLinear(3, 128, 128, bias=False, dtype=torch.bfloat16, device="cuda") + with te.autocast(enabled=True, recipe=fp8_recipe): + op.reset_recipe_state(recipe=fp8_recipe) + quantizers = [op.get_quantizer("forward", 2 * i + 1) for i in range(op.num_groups)] + assert len({id(q) for q in quantizers}) == 3 + for q in quantizers: + q.set_usage(rowwise=True, columnwise=True) + weights = [torch.randn(128, 128, device="cuda", dtype=torch.bfloat16) for _ in range(3)] + reference = [q(w) for w, q in zip(weights, quantizers)] + weights[1] = reference[1] + param_quantizer = weights[1]._quantizer + assert param_quantizer is not quantizers[1] + # The data already has both layouts; its quantizer must adopt the op's requirement. + param_quantizer.set_usage(rowwise=True, columnwise=False) + for q in quantizers: + q.set_usage(rowwise=False, columnwise=False) + + # Enable columnwise usage, stop requesting it, then request it again. + for columnwise_usage in (True, False, True): + quantizer_spies = [Mock(wraps=q) for q in quantizers] + outputs = op._get_discrete_weights_for_gemm( + weights, + quantizer_spies, + columnwise_usage=columnwise_usage, + with_quantized_compute=True, + dtype=torch.bfloat16, + ) + for i in (0, 2): + quantizer_spies[i].assert_called_once_with(weights[i]) + for spy, quantizer in zip(quantizer_spies, quantizers): + spy.set_usage.assert_called_once_with(rowwise=True, columnwise=columnwise_usage) + assert quantizer.rowwise_usage + assert quantizer.columnwise_usage == columnwise_usage + quantizer_spies[1].assert_not_called() + assert outputs[1] is weights[1] + assert weights[1]._quantizer is param_quantizer + assert param_quantizer.rowwise_usage + assert param_quantizer.columnwise_usage + assert weights[1].get_usages()["columnwise"] + for output, ref in zip(outputs, reference): + torch.testing.assert_close(output.dequantize(), ref.dequantize(), rtol=0, atol=0) + scales = [w._rowwise_scale_inv if fp8_recipe.mxfp8() else w._scale_inv for w in outputs] + assert len({s.data_ptr() for s in scales}) == 3 diff --git a/tests/pytorch/test_grouped_linear.py b/tests/pytorch/test_grouped_linear.py index 774e758a637..4bf74d11ddd 100644 --- a/tests/pytorch/test_grouped_linear.py +++ b/tests/pytorch/test_grouped_linear.py @@ -4,6 +4,7 @@ import os import random +from unittest.mock import Mock from typing import Dict, List, Optional, Sequence import pytest @@ -60,6 +61,35 @@ seed = 1234 reset_rng_states() + +@pytest.mark.parametrize("has_skip", [False, True]) +@pytest.mark.parametrize("capturing", [False, True]) +def test_grouped_linear_skip_update_capture_short_circuit(monkeypatch, has_skip, capturing): + """Only an active capture skip tensor may override first-microbatch accumulation.""" + FP8GlobalStateManager.reset() + module = GroupedLinear(2, 128, 128, bias=False, params_dtype=torch.bfloat16) + x = torch.randn(256, 128, device="cuda", dtype=torch.bfloat16) + skip = torch.ones(1, device="cuda") if has_skip else None + monkeypatch.setattr( + FP8GlobalStateManager.quantization_state, "skip_fp8_weight_update_tensor", skip + ) + capture = Mock(return_value=capturing) + monkeypatch.setattr(FP8GlobalStateManager, "fp8_graph_capturing", capture) + original = _GroupedLinear.forward + seen = [] + + def forward(ctx, inp, fwd_args, *args): + seen.append((fwd_args.is_first_microbatch, fwd_args.skip_fp8_weight_update)) + return original(ctx, inp, fwd_args, *args) + + monkeypatch.setattr(_GroupedLinear, "forward", forward) + with torch.no_grad(): + module(x, [128, 128], is_first_microbatch=True) + assert capture.call_count == int(has_skip) + assert seen[0][0] == (not (has_skip and capturing)) + assert seen[0][1] is (skip if capturing else None) + + NVTE_TEST_NVINSPECT_ENABLED = int(os.environ.get("NVTE_TEST_NVINSPECT_ENABLED", "0")) if NVTE_TEST_NVINSPECT_ENABLED: @@ -1986,6 +2016,198 @@ def test_single_grouped_primary_mxfp8_bypasses_weight_workspace(): assert "weight" not in grouped_linear._fp8_workspaces +@pytest.mark.skipif(not _mxfp8_available, reason=_reason_for_no_mxfp8) +@pytest.mark.parametrize("primary_fp8", [False, True]) +@pytest.mark.parametrize("save_original_input", [False, True]) +def test_grouped_runtime_usage_transitions(primary_fp8, save_original_input): + """Preserve primary usages and rebuild missing columnwise workspaces after inference.""" + from transformer_engine.pytorch.distributed import activation_recompute_forward + + fp8_recipe = recipe.MXFP8BlockScaling() + if not is_module_grouped_tensor_path_supported(fp8_recipe, torch.bfloat16): + pytest.skip("Native MXFP8 grouped GEMM is unavailable.") + FP8GlobalStateManager.reset() + with quantized_model_init(enabled=primary_fp8, recipe=fp8_recipe): + module = GroupedLinear( + 2, + 128, + 128, + bias=False, + params_dtype=torch.bfloat16, + single_grouped_weight=True, + use_grouped_tensor=True, + save_original_input=save_original_input, + ) + splits = torch.tensor([128, 128], device="cuda") + x = torch.randn(256, 128, device="cuda", dtype=torch.bfloat16, requires_grad=True) + + def run(*, grad_enabled=True, first=False): + with torch.set_grad_enabled(grad_enabled), autocast(enabled=True, recipe=fp8_recipe): + return module(x, splits, is_first_microbatch=first) + + run(first=True).sum().backward() + reference_grad = x.grad.clone() + reference_wgrad = module.weight.grad.clone() + weight_q = module.weight.quantizer if primary_fp8 else module._get_weight_quantizers()[0] + assert weight_q.columnwise_usage + module.eval() + run(grad_enabled=False, first=True) + # Persistent parameters retain columnwise usage; temporary workspaces follow this call. + assert weight_q.columnwise_usage == primary_fp8 + # Force a rowwise-only workspace, as on a first inference microbatch. + if not primary_fp8: + module._fp8_workspaces.clear() + run(grad_enabled=False, first=True) + assert module._fp8_workspaces["weight"].columnwise_data is None + module.train() + module.weight.grad = None + x.grad = None + # Keep two forwards alive before either backward; quantizers are shared references. + y1, y2 = run(), run() + assert weight_q.columnwise_usage + y1.sum().backward() + y2.sum().backward() + torch.testing.assert_close(x.grad, 2 * reference_grad, rtol=1e-2, atol=5e-3) + torch.testing.assert_close(module.weight.grad, 2 * reference_wgrad, rtol=1e-2, atol=5e-3) + with activation_recompute_forward(activation_recompute=True, recompute_phase=False): + run(grad_enabled=False, first=True) + assert weight_q.columnwise_usage + with activation_recompute_forward(activation_recompute=True, recompute_phase=True): + run().sum().backward() + assert weight_q.columnwise_usage + module.weight.requires_grad_(False) + run().sum().backward() + input_q = module.quantizers["scaling_fwd"][module._offsets["input"]] + assert not input_q.columnwise_usage + + +@pytest.mark.skipif(not _mxfp8_available, reason=_reason_for_no_mxfp8) +def test_grouped_runtime_recipe_rebuild(): + """Recipe rebuilds replace cached quantizers while preserving saved forwards.""" + recipes = [ + recipe.MXFP8BlockScaling(), + recipe.Float8CurrentScaling(), + recipe.MXFP8BlockScaling(), + ] + if not all(is_module_grouped_tensor_path_supported(r, torch.bfloat16) for r in recipes): + pytest.skip("Required native grouped GEMM recipes are unavailable.") + FP8GlobalStateManager.reset() + module = GroupedLinear( + 2, + 128, + 128, + bias=False, + params_dtype=torch.bfloat16, + single_grouped_weight=True, + use_grouped_tensor=True, + ) + reference = GroupedLinear( + 2, + 128, + 128, + bias=False, + params_dtype=torch.bfloat16, + use_grouped_tensor=True, + ) + reference.load_state_dict(module.state_dict()) + x = torch.randn(256, 128, device="cuda", dtype=torch.bfloat16, requires_grad=True) + ref_x = x.detach().clone().requires_grad_(True) + splits = torch.tensor([128, 128], device="cuda") + pending = [] + workspaces = [] + for fp8_recipe in recipes: + with autocast(enabled=True, recipe=fp8_recipe): + y = module(x, splits, is_first_microbatch=True) + ref_y = reference(ref_x, splits, is_first_microbatch=True) + pending.append((y, ref_y)) + workspace = module._fp8_workspaces["weight"] + assert workspace.quantizer is module._get_weight_quantizers()[0] + assert all(workspace.quantizer is not old.quantizer for old in workspaces) + workspaces.append(workspace) + for y, ref_y in reversed(pending): + torch.testing.assert_close(y, ref_y, rtol=1e-2, atol=5e-3) + y.sum().backward() + ref_y.sum().backward() + torch.testing.assert_close(x.grad, ref_x.grad, rtol=1e-2, atol=5e-3) + ref_wgrad = torch.stack([reference.weight0.grad, reference.weight1.grad]) + torch.testing.assert_close(module.weight.grad, ref_wgrad, rtol=1e-2, atol=5e-3) + + +@pytest.mark.parametrize("single_weight", [False, True]) +@pytest.mark.parametrize("restriction", ["debug", "calibration", "offload", "output", "override"]) +def test_grouped_runtime_rechecks_support(monkeypatch, single_weight, restriction): + """Recheck runtime restrictions without allowing a single weight to use the split path.""" + module = GroupedLinear( + 2, + 128, + 128, + bias=False, + params_dtype=torch.bfloat16, + single_grouped_weight=single_weight, + use_grouped_tensor=True, + ) + module.activation_dtype = torch.bfloat16 + module.fp8 = True + active_recipe = recipe.Float8CurrentScaling() + if restriction == "override": + active_recipe.backward_override = "high_precision" + if not is_module_grouped_tensor_path_supported(recipe.Float8CurrentScaling(), torch.bfloat16): + pytest.skip("Native current-scaling grouped GEMM is unavailable.") + monkeypatch.setattr(FP8GlobalStateManager, "get_fp8_recipe", lambda: active_recipe) + module.fp8_calibration = restriction == "calibration" + args = ( + restriction == "debug", + restriction == "offload", + [object()] if restriction == "output" else (), + ) + if single_weight: + with pytest.raises(RuntimeError, match="Single grouped parameters require"): + module._grouped_tensor_path_supported(*args) + else: + assert not module._grouped_tensor_path_supported(*args) + module.fp8_calibration = False + active_recipe.backward_override = None + assert module._grouped_tensor_path_supported(False, False, ()) + + +@pytest.mark.parametrize( + "use_grouped_tensor,single_bias", [(False, False), (True, False), (True, True)] +) +@pytest.mark.parametrize("use_bias", [False, True]) +def test_grouped_linear_return_bias_with_frozen_weights(use_grouped_tensor, single_bias, use_bias): + """A trainable returned bias keeps the output differentiable when input/weight are frozen.""" + if use_grouped_tensor and not is_module_grouped_tensor_path_supported(None, torch.bfloat16): + pytest.skip("Native grouped GEMM is unavailable.") + module = GroupedLinear( + 2, + 128, + 128, + bias=use_bias, + return_bias=True, + params_dtype=torch.bfloat16, + single_grouped_weight=single_bias, + single_grouped_bias=single_bias, + use_grouped_tensor=use_grouped_tensor, + ) + for weight in _grouped_linear_weight_params(module): + weight.requires_grad_(False) + x = torch.randn(256, 128, device="cuda", dtype=torch.bfloat16) + splits = torch.tensor([128, 128], device="cuda" if use_grouped_tensor else "cpu") + y, biases = module(x, splits) + assert y.requires_grad == use_bias + if not use_bias: + assert isinstance(biases, list) and len(biases) == 2 + assert all(b.numel() == 0 for b in biases) + return + params = _grouped_linear_bias_params(module) + returned = [biases] if single_bias else biases + assert len(returned) == len(params) + assert all(value is param for value, param in zip(returned, params)) + (y.sum() + sum(value.sum() for value in returned)).backward() + for param in params: + torch.testing.assert_close(param.grad, torch.ones_like(param.grad), rtol=0, atol=0) + + def _clone_outputs(outputs): return [None if out is None else out.detach().clone() for out in outputs] @@ -2821,8 +3043,200 @@ def test_grouped_linear_grouped_tensor_path_skips_non_rht_nvfp4(): ) -def test_grouped_linear_delay_wgrad_rejects_implicit_fallback(monkeypatch): +@pytest.mark.parametrize( + "use_grouped_tensor,single_weight,single_bias", + [ + (False, False, False), + (True, False, False), + (True, True, False), + (True, False, True), + (True, True, True), + ], + ids=["split", "grouped-discrete", "single-weight", "single-bias", "single-both"], +) +@pytest.mark.parametrize( + "recipe_name", ["bf16", "delayed", "current", "mxfp8", "high_precision", "dequantized"] +) +@pytest.mark.parametrize("freeze_weight", [False, True]) +@pytest.mark.parametrize("input_requires_grad", [False, True]) +@pytest.mark.parametrize("delay_wgrad", [False, True]) +@pytest.mark.parametrize("grad_storage", ["grad", "preallocated_grad", "main_grad"]) +def test_grouped_linear_bias_gradient_accumulation( + use_grouped_tensor, + single_weight, + single_bias, + recipe_name, + freeze_weight, + input_requires_grad, + delay_wgrad, + grad_storage, +): + """Bias gradients accumulate during main backward, independently of deferred/frozen weights.""" + if recipe_name == "bf16": + fp8_recipe = None + elif recipe_name == "delayed": + fp8_recipe = recipe.DelayedScaling() + elif recipe_name == "current": + fp8_recipe = recipe.Float8CurrentScaling() + # Use exact power-of-two scales for the analytical FP32 main_grad reference. + fp8_recipe.fp8_quant_fwd_inp = recipe.QParams(power_2_scale=True) + fp8_recipe.fp8_quant_fwd_weight = recipe.QParams(power_2_scale=True) + fp8_recipe.fp8_quant_bwd_grad = recipe.QParams(power_2_scale=True) + else: + if not mxfp8_available: + pytest.skip(reason_for_no_mxfp8) + fp8_recipe = recipe.MXFP8BlockScaling( + backward_override=None if recipe_name == "mxfp8" else recipe_name + ) + if fp8_recipe is not None and not fp8_available: + pytest.skip(reason_for_no_fp8) + if use_grouped_tensor and not is_module_grouped_tensor_path_supported( + fp8_recipe, torch.bfloat16 + ): + pytest.skip("Recipe is unsupported by the native grouped-tensor path") + + FP8GlobalStateManager.reset() + module = GroupedLinear( + 2, + 128, + 128, + bias=True, + params_dtype=torch.bfloat16, + use_grouped_tensor=use_grouped_tensor, + single_grouped_weight=single_weight, + single_grouped_bias=single_bias, + delay_wgrad_compute=delay_wgrad, + fuse_wgrad_accumulation=grad_storage == "main_grad", + ) + weights = _grouped_linear_weight_params(module) + biases = _grouped_linear_bias_params(module) + with torch.no_grad(): + for param in weights: + data = param.rowwise_data if isinstance(param, GroupedTensor) else param + data.fill_(0.125) + param.requires_grad_(not freeze_weight) + for param in biases: + data = param.rowwise_data if isinstance(param, GroupedTensor) else param + data.fill_(0.5) + + hook_calls = {id(param): 0 for param in module.parameters()} + flat_main_grad = torch.zeros(2 * 128 * 128, device="cuda", dtype=torch.float32) + weight_views = flat_main_grad.view(2, 128, 128) + weight_views = [weight_views] if single_weight else list(weight_views.unbind()) + main_grad_ptrs = [] + if grad_storage == "main_grad": + # Model MCore's ownership: weight GEMM accumulates into an external flat buffer, + # while parameter hooks move unfused grads to main_grad and clear the handoff. + for param, view in zip(weights, weight_views): + param.main_grad = view + param.grad_added_to_main_grad = False + param.overwrite_main_grad = False + main_grad_ptrs.append(view.data_ptr()) + for param in biases: + param.main_grad = torch.zeros(param.shape, device="cuda", dtype=torch.float32) + param.grad_added_to_main_grad = False + + def accumulate(param): + hook_calls[id(param)] += 1 + if param.grad is not None and not param.grad_added_to_main_grad: + param.main_grad.add_(param.grad) + param.grad = None + + for param in module.parameters(): + if not param.requires_grad: + continue + if getattr(param, "skip_backward_post_hook", False): + module.register_wgrad_accumulation_and_reduce_hooks( + lambda param=param: accumulate(param) + ) + else: + param.register_post_accumulate_grad_hook(accumulate) + elif grad_storage == "preallocated_grad": + for param in biases: + param.grad = torch.zeros(param.shape, device="cuda", dtype=param.dtype) + + splits = torch.tensor([128, 128], dtype=torch.int64, device="cuda") + total_scale = 0 + try: + # Decreasing dY avoids delayed-scaling overflow from a larger second amax. + # Powers of two make output and gradients exactly representable in all tested recipes. + for step, scale in enumerate((2, 1)): + total_scale += scale + for param in weights: + param.grad = None + x = torch.full( + (256, 128), + 0.25, + device="cuda", + dtype=torch.bfloat16, + requires_grad=input_requires_grad, + ) + with autocast(enabled=fp8_recipe is not None, recipe=fp8_recipe): + y = module(x, splits, is_first_microbatch=step == 0) + torch.testing.assert_close(y, torch.full_like(y, 4.5), rtol=0, atol=0) + y.backward(torch.full_like(y, scale)) + if input_requires_grad: + torch.testing.assert_close( + x.grad, torch.full_like(x.grad, 16 * scale), rtol=0, atol=0 + ) + else: + assert x.grad is None + + def check_bias_grads(): + for param in biases: + grad = param.main_grad if grad_storage == "main_grad" else param.grad + assert grad is not None, "bias gradient was not returned by main backward" + torch.testing.assert_close( + grad, torch.full_like(grad, 128 * total_scale), rtol=0, atol=0 + ) + if grad_storage == "main_grad": + assert param.grad is None + assert hook_calls[id(param)] == step + 1 + + check_bias_grads() + if delay_wgrad: + module.backward_dw() + check_bias_grads() # Deferred weight hooks must not add dbias a second time. + for param in weights: + if freeze_weight: + assert param.grad is None + continue + grad = param.main_grad if grad_storage == "main_grad" else param.grad + assert grad is not None + expected = 32 * (total_scale if grad_storage == "main_grad" else scale) + torch.testing.assert_close(grad, torch.full_like(grad, expected), rtol=0, atol=0) + if grad_storage == "main_grad": + assert [param.main_grad.data_ptr() for param in weights] == main_grad_ptrs + finally: + FP8GlobalStateManager.reset() + + +@pytest.mark.parametrize("device", ["cuda", "meta"]) +@pytest.mark.parametrize("single_weight", [False, True]) +@pytest.mark.parametrize("single_bias", [False, True]) +def test_grouped_linear_delayed_wgrad_hook_ownership(device, single_weight, single_bias): + """Only weight hooks are deferred, including before meta parameters are materialized.""" + module = GroupedLinear( + 2, + 128, + 128, + bias=True, + device=device, + params_dtype=torch.bfloat16, + use_grouped_tensor=True, + single_grouped_weight=single_weight, + single_grouped_bias=single_bias, + delay_wgrad_compute=True, + ) + for name, param in module.named_parameters(): + assert getattr(param, "skip_backward_post_hook", False) == name.startswith("weight") + + +@pytest.mark.parametrize("backward_override", [None, "high_precision", "dequantized"]) +def test_grouped_linear_delay_wgrad_rejects_implicit_fallback(monkeypatch, backward_override): """Delayed wgrad reports when a grouped-tensor request used the legacy path.""" + if backward_override is not None and not mxfp8_available: + pytest.skip(reason_for_no_mxfp8) monkeypatch.setattr( "transformer_engine.pytorch.module.grouped_linear.is_module_grouped_tensor_path_supported", lambda *_args, **_kwargs: False, @@ -2837,10 +3251,14 @@ def test_grouped_linear_delay_wgrad_rejects_implicit_fallback(monkeypatch): delay_wgrad_compute=True, use_grouped_tensor=True, ) - x = torch.randn(16, 64, dtype=torch.bfloat16, device="cuda", requires_grad=True) - m_splits = torch.tensor([8, 8], dtype=torch.int64, device="cuda") + x = torch.randn(64, 64, dtype=torch.bfloat16, device="cuda", requires_grad=True) + m_splits = torch.tensor([32, 32], dtype=torch.int64, device="cuda") - grouped_linear(x, m_splits).sum().backward() + with autocast( + enabled=backward_override is not None, + recipe=recipe.MXFP8BlockScaling(backward_override=backward_override), + ): + grouped_linear(x, m_splits).sum().backward() with pytest.raises( RuntimeError, match="implicit fallback is unsupported with delay_wgrad_compute=True", diff --git a/tests/pytorch/test_grouped_mlp.py b/tests/pytorch/test_grouped_mlp.py index fd661ea1e00..16caa21d936 100644 --- a/tests/pytorch/test_grouped_mlp.py +++ b/tests/pytorch/test_grouped_mlp.py @@ -12,6 +12,7 @@ import sys import types from typing import Optional +from unittest.mock import Mock import pytest @@ -484,6 +485,87 @@ def test_meta_single_grouped_weight_with_delayed_wgrad(self) -> None: assert grouped_weight.skip_backward_post_hook + @pytest.mark.parametrize("num_groups", [1, 8]) + @pytest.mark.parametrize("accumulate_into_main_grad", [False, True]) + @pytest.mark.parametrize("delay_wgrad_compute", [False, True]) + def test_wgrad_storage_and_hooks( + self, + monkeypatch: pytest.MonkeyPatch, + num_groups: int, + accumulate_into_main_grad: bool, + delay_wgrad_compute: bool, + ) -> None: + """Check independent wgrad storage, external main_grad views, and delayed hooks.""" + from transformer_engine.pytorch.ops.basic import grouped_linear as gl + + if not is_op_fuser_grouped_tensor_path_supported(None, torch.bfloat16): + pytest.skip("Native grouped GEMM is unavailable.") + op = te.ops.GroupedLinear( + num_groups, + 128, + 128, + bias=False, + device="cuda", + dtype=torch.bfloat16, + accumulate_into_main_grad=accumulate_into_main_grad, + delay_wgrad_compute=delay_wgrad_compute, + ) + model = te.ops.Sequential(op) + flat = torch.full((num_groups, 128, 128), 0.25, device="cuda", dtype=torch.float32) + weights = [getattr(op, f"weight{i}") for i in range(num_groups)] + for i, weight in enumerate(weights): + with torch.no_grad(): + weight.fill_(1) + if accumulate_into_main_grad: + # Use views of an external flat buffer to check that GEMM preserves aliases. + weight.main_grad = flat[i] + weight.overwrite_main_grad = False + weight.grad_added_to_main_grad = False + x = torch.full( + (num_groups * 128, 128), 0.125, device="cuda", dtype=torch.bfloat16, requires_grad=True + ) + splits = torch.full((num_groups,), 128, device="cuda", dtype=torch.int64) + seen = [] + original = gl.general_grouped_gemm_for_grouped_tensor + + def gemm(a, b, out, **kwargs): + if kwargs.get("layout") == "NT": + assert len(out) == num_groups + num_storages = len({g.untyped_storage().data_ptr() for g in out}) + assert num_storages == (1 if accumulate_into_main_grad else num_groups) + assert len({g.data_ptr() for g in out}) == num_groups + if accumulate_into_main_grad: + assert out[0].data_ptr() == flat.data_ptr() + else: + assert all(g._base is None for g in out) + seen.append(out) + return original(a, b, out, **kwargs) + + monkeypatch.setattr(gl, "general_grouped_gemm_for_grouped_tensor", gemm) + hook = Mock() + op.register_wgrad_accumulation_and_reduce_hooks(hook) + y = model(x, splits) + torch.testing.assert_close(y, torch.full_like(y, 16), rtol=0, atol=0) + y.backward(torch.full_like(y, 0.5)) + if delay_wgrad_compute: + assert not seen + hook.assert_not_called() + op.backward_dw() + hook.assert_called_once() + assert len(seen) == 1 + if accumulate_into_main_grad: + MegatronTrainingHelper.verify_main_grad_accumulation( + weights, + expected_main_grads=[torch.full_like(flat[i], 8.25) for i in range(num_groups)], + ) + for i, weight in enumerate(weights): + assert weight.main_grad.data_ptr() == flat[i].data_ptr() + else: + for weight in weights: + torch.testing.assert_close( + weight.grad, torch.full_like(weight.grad, 8), rtol=0, atol=0 + ) + def test_single_grouped_bias_uses_registered_packed_storage(self) -> None: """The grouped bias compute view must alias the registered trainable parent.""" op = te.ops.GroupedLinear( diff --git a/tests/pytorch/test_module_grouped_linear_distributed_weight.py b/tests/pytorch/test_module_grouped_linear_distributed_weight.py index 05fba219a99..2a664084c29 100644 --- a/tests/pytorch/test_module_grouped_linear_distributed_weight.py +++ b/tests/pytorch/test_module_grouped_linear_distributed_weight.py @@ -20,6 +20,9 @@ a specific assertion rather than drifting numerically. """ +import gc +import weakref + import pytest import torch @@ -283,3 +286,93 @@ def _finalize(wgrads, **kwargs): assert calls == ["fwd", "bwd", "finalize"], f"unexpected hook dispatch: {calls}" assert torch.count_nonzero(weight.wgrad_scratch) > 0, "wgrad never reached grad_buffer" + + +@pytest.mark.parametrize("single_weight", [False, True]) +@pytest.mark.parametrize("quantized", [False, True]) +@pytest.mark.parametrize("fused", [False, True]) +@pytest.mark.parametrize("requires_dgrad", [False, True]) +def test_gathered_weights_are_transient( + monkeypatch, single_weight, quantized, fused, requires_dgrad +): + """No gathered copy survives forward; backward must gather even when dgrad is disabled.""" + from transformer_engine.pytorch.module.grouped_linear import ( + _GroupedLinear, + is_module_grouped_tensor_path_supported, + ) + from transformer_engine.pytorch.tensor.grouped_tensor import GroupedTensor + + fp8_recipe = MXFP8BlockScaling() if quantized else None + if not is_module_grouped_tensor_path_supported(fp8_recipe, DTYPE): + pytest.skip("Requested native grouped GEMM is unavailable.") + te.quantization.FP8GlobalStateManager.reset() + module = te.GroupedLinear( + 2, + IN_F, + OUT_F, + bias=False, + params_dtype=DTYPE, + device=DEVICE, + single_grouped_weight=single_weight, + use_grouped_tensor=True, + fuse_wgrad_accumulation=fused, + ) + weights = [module.weight] if single_weight else [module.weight0, module.weight1] + for weight in weights: + weight.main_grad = torch.zeros(weight.shape, device=DEVICE, dtype=torch.float32) + weight.wgrad_scratch = torch.zeros_like(weight.main_grad) + weight.grad_buffer = lambda w=weight: w.wgrad_scratch + gathered_refs, calls = [], [] + + def gather(phase): + calls.append(phase) + if single_weight: + copy = GroupedTensor.make_grouped_tensor_from_rowwise_data( + num_tensors=2, + tensor_shape=(OUT_F, IN_F), + rowwise_data=weights[0].rowwise_data.detach().clone(), + dtype=DTYPE, + ) + copies = [copy] + else: + copies = [w.detach().clone() for w in weights] + assert all(not w.requires_grad for w in copies) + gathered_refs.extend(weakref.ref(w) for w in copies) + return copies + + def finalize(grads, **kwargs): + calls.append("finalize") + grads = [grads] if single_weight else grads + if fused: + for weight, grad in zip(weights, grads): + assert grad.data_ptr() == weight.wgrad_scratch.data_ptr() + weight.main_grad.add_(grad) + return [None] * len(weights) + return list(grads) + + leader = weights[0] + leader.is_distributed_weight = True + leader.materialize_group_for_forward = lambda: gather("fwd") + leader.materialize_group_for_backward = lambda **kw: gather("bwd") + leader.finalize_group_grads = finalize + prep = [] + original = _GroupedLinear._prepare_weights_for_grouped_tensor_gemm + + def prepare(*args, **kwargs): + prep.append(kwargs["cache_weight"]) + return original(*args, **kwargs) + + monkeypatch.setattr(_GroupedLinear, "_prepare_weights_for_grouped_tensor_gemm", prepare) + splits, x = _inputs(2) + x.requires_grad_(requires_dgrad) + with te.autocast(enabled=quantized, recipe=fp8_recipe): + y = module(x, splits, is_first_microbatch=True) + gc.collect() + assert not module._fp8_workspaces + assert all(ref() is None for ref in gathered_refs) + y.sum().backward() + assert calls == ["fwd", "bwd", "finalize"] + assert prep == [False] * (1 + int(requires_dgrad)) + for weight in weights: + grad = weight.main_grad if fused else weight.grad + assert grad is not None and torch.count_nonzero(grad) > 0 diff --git a/tests/pytorch/test_quantized_tensor.py b/tests/pytorch/test_quantized_tensor.py index 4e64215cdf5..6cd1bd403b1 100644 --- a/tests/pytorch/test_quantized_tensor.py +++ b/tests/pytorch/test_quantized_tensor.py @@ -25,6 +25,7 @@ QuantizedTensor, ) +from transformer_engine.pytorch.quantized_tensor import prepare_for_saving, restore_from_saved from transformer_engine.pytorch.utils import is_non_tn_fp8_gemm_supported from references.ref_per_tensor_cs import ref_per_tensor_cs_cast @@ -1187,3 +1188,48 @@ def clone_quantized_output(): assert torch.equal( buf_without_noop, buf_with_noop ), f"{quantization}/{usage}: noop flag fails to take effect because {attr} changed." + + +def test_saved_storage_protocol(monkeypatch): + """Preserve custom metadata, pass the full suffix, and use the returned remainder.""" + value, tail, replacement = (torch.tensor(i) for i in range(3)) + remainder = [replacement, tail] + + class Storage: + __slots__ = () + + def prepare_for_saving(self): + return [value], self + + def restore_from_saved(self, tensors): + assert isinstance(tensors, list) + assert len(tensors) == 2 and tensors[0] is value and tensors[1] is tail + return remainder + + monkeypatch.setattr(torch.compiler, "is_compiling", lambda: False) + storage = Storage() + saved, objects = prepare_for_saving(value, storage) + restored, remaining = restore_from_saved( + objects + [None], saved + [tail], return_saved_tensors=True + ) + assert restored[0] is value and restored[1] is storage and restored[2] is replacement + assert isinstance(remaining, list) and len(remaining) == 1 and remaining[0] is tail + + +@pytest.mark.parametrize("compiled_metadata", [False, True]) +def test_restore_real_quantized_storage(monkeypatch, compiled_metadata): + """Restore actual MXFP8 buffers and None/plain tensor slots with either metadata form.""" + if not te.is_mxfp8_available(): + pytest.skip("MXFP8 is unavailable.") + q = MXFP8Quantizer(te.DType.kFloat8E4M3, rowwise=True, columnwise=True) + q.internal = True + x = torch.randn(128, 128, device="cuda", dtype=torch.bfloat16) + quantized = q(x) + expected = quantized.dequantize().clone() + with monkeypatch.context() as context: + context.setattr(torch.compiler, "is_compiling", lambda: compiled_metadata) + saved, objects = prepare_for_saving(x, quantized, None) + restored, tail = restore_from_saved(objects, tuple(saved) + (x,), return_saved_tensors=True) + assert restored[0] is x and restored[2] is None + assert isinstance(tail, tuple) and tail[0] is x + torch.testing.assert_close(restored[1].dequantize(), expected, rtol=0, atol=0) diff --git a/transformer_engine/pytorch/cpp_extensions/gemm.py b/transformer_engine/pytorch/cpp_extensions/gemm.py index 6939847c8dc..5eeca35e77a 100644 --- a/transformer_engine/pytorch/cpp_extensions/gemm.py +++ b/transformer_engine/pytorch/cpp_extensions/gemm.py @@ -12,7 +12,13 @@ import torch import transformer_engine_torch as tex from ..constants import MXFP8_BLOCK_SCALING_SIZE, NVFP4_BLOCK_SCALING_SIZE, TE_DType, DType -from ..utils import ceil_div, get_cached_ones_tensor, get_sm_count, _empty_tensor +from ..utils import ( + ceil_div, + get_cached_ones_tensor, + get_device_compute_capability, + get_sm_count, + _empty_tensor, +) from ..quantized_tensor import QuantizedTensorStorage, Quantizer from ..tensor.float8_blockwise_tensor import Float8BlockQuantizer @@ -1185,7 +1191,7 @@ def _is_fp8_blockwise(operand) -> bool: # Hopper (SM90) uses a single shared alpha/beta scalar; # Blackwell+ (SM100) supports per-group alpha/beta arrays. - per_group = torch.cuda.get_device_capability() >= (10, 0) + per_group = get_device_compute_capability() >= (10, 0) num_alphabeta = num_tensors if per_group else 1 if alpha is None: diff --git a/transformer_engine/pytorch/module/grouped_linear.py b/transformer_engine/pytorch/module/grouped_linear.py index 828ec74d9be..c4ebe227767 100644 --- a/transformer_engine/pytorch/module/grouped_linear.py +++ b/transformer_engine/pytorch/module/grouped_linear.py @@ -222,6 +222,7 @@ class GroupedLinearBwdArgs: m_splits: Optional[List[int]] = None num_gemms: int = 0 + num_bias_args: int = 0 weights_shape_1: int = 0 use_bias: bool = False @@ -253,11 +254,15 @@ def setup_saved_tensors(self, ctx: torch.autograd.function.FunctionCtx) -> None: self.inputmats = saved[:n] self.weights_fp8 = saved[n : 2 * n] self.saved_weights = saved[2 * n : 3 * n] - self.biases = saved[3 * n : 4 * n] + self.biases = saved[3 * n : 3 * n + self.num_bias_args] -def _grouped_linear_configure_quantizers(args: GroupedLinearFwdArgs, weights) -> None: - """Configure operand usage after materializing weights.""" +def _grouped_linear_configure_split_quantizers(args: GroupedLinearFwdArgs, weights) -> None: + """Configure per-expert quantizers for the split-quantize path after materialization. + + The grouped-tensor path configures its input quantizer before group_quantize + and its weight quantizers in _prepare_weights_for_grouped_tensor_gemm. + """ if args.input_quantizers[0] is not None: for input_quantizer in args.input_quantizers: input_quantizer.set_usage( @@ -314,7 +319,7 @@ def _grouped_linear_forward_impl( if is_dist_weight: weights = materialize_weight_for_forward(weights) - _grouped_linear_configure_quantizers(args, weights) + _grouped_linear_configure_split_quantizers(args, weights) weight_quantizers = args.weight_quantizers # Initialize input tensors @@ -448,6 +453,7 @@ def _grouped_linear_setup_ctx( bwd_args.m_splits = fwd_args.m_splits bwd_args.num_gemms = num_gemms + bwd_args.num_bias_args = len(fwd_args.biases) bwd_args.weights_shape_1 = fwd_args.inp.shape[-1] bwd_args.use_bias = fwd_args.use_bias @@ -550,6 +556,12 @@ def _grouped_linear_backward_impl( # Preprocess grad output out_features = grad_output.shape[-1] grad_output_view = grad_output.contiguous().view(-1, out_features) + delay_wgrad = args.wgrad_store is not None and args.wgrad_store.delay_wgrad_compute() + # Bias gradients belong to the main backward even when weights are frozen + # or their GEMM is deferred. Keep fused BF16 wgrad+dbias for immediate wgrad. + compute_dbias = args.use_bias and ( + args.fp8 or args.debug or not args.weights_requires_grad or delay_wgrad + ) grad_output_reference = args.grad_output_quantizers[0] if args.fp8 and isinstance(grad_output_reference, HybridQuantizer): # Usage is a runtime decision, not part of generation validation. @@ -566,7 +578,7 @@ def _grouped_linear_backward_impl( args.grad_output_quantizers, args.activation_dtype, with_quantized_output=args.fp8 or args.debug, - compute_dbias=(args.fp8 or args.debug) and args.use_bias, + compute_dbias=compute_dbias, disable_bulk_allocation=( args.cpu_offloading and isinstance(grad_output_reference, HybridQuantizer) @@ -637,11 +649,7 @@ def _grouped_linear_backward_impl( ) if args.weights_requires_grad: - if ( - is_dist_weight - and args.wgrad_store is not None - and args.wgrad_store.delay_wgrad_compute() - ): + if is_dist_weight and delay_wgrad: raise RuntimeError( "distributed-weight GroupedLinear requires delay_wgrad_compute=False." ) @@ -695,7 +703,9 @@ def _grouped_linear_backward_impl( layout="NT", grad=True, m_splits=m_splits, - use_bias=args.use_bias if grad_biases[0] is None else None, + # Preserve the high-precision GEMM epilogue and algorithm selection, + # including when delayed wgrad no longer owns bias accumulation. + use_bias=args.use_bias and not (args.fp8 or args.debug), bias=biases, use_split_accumulator=wgrad_use_split_accumulator, accumulate=( @@ -705,14 +715,12 @@ def _grouped_linear_backward_impl( ), ) # WGRAD - if args.wgrad_store is not None and args.wgrad_store.delay_wgrad_compute(): + if delay_wgrad: args.wgrad_store.put([inputmats, grad_output, wgrad_list], grouped_gemm_wgrad) else: _, grad_biases_, _ = grouped_gemm_wgrad(inputmats, grad_output, wgrad_list) - - for i in range(num_gemms): - if grad_biases[i] is None: - grad_biases[i] = grad_biases_[i] + if not compute_dbias: + grad_biases = grad_biases_ del grad_biases_ clear_tensor_data(*inputmats) @@ -727,10 +735,8 @@ def _grouped_linear_backward_impl( else: wgrad_list = [None] * num_gemms - if not args.use_bias or ( - args.wgrad_store is not None and args.wgrad_store.delay_wgrad_compute() and not args.fp8 - ): - grad_biases = [None] * num_gemms + if not args.use_bias: + grad_biases = [None] * args.num_bias_args dgrad_out = None if args.requires_dgrad: @@ -756,6 +762,7 @@ class GroupedLinearGraphSafeBwdArgs: grad_output_quantizers: List[Quantizer] = None is_dist_weight: bool = False num_gemms: int = 0 + num_bias_args: int = 0 in_features: int = 0 out_features: int = 0 activation_dtype: Optional[torch.dtype] = None @@ -808,7 +815,6 @@ def _grouped_linear_graph_safe_forward(args: GroupedLinearFwdArgs) -> Tuple[Any, is_dist_weight = is_distributed_weight(weights[0]) if is_dist_weight: weights = materialize_weight_for_forward(weights) - _grouped_linear_configure_quantizers(args, weights) weight_quantizers = args.weight_quantizers biases = args.biases out = args.out @@ -865,7 +871,9 @@ def _grouped_linear_graph_safe_forward(args: GroupedLinearFwdArgs) -> Tuple[Any, dtype=activation_dtype, ) - columnwise_usage = is_grad_enabled and args.input_requires_grad + columnwise_usage = (is_grad_enabled and args.input_requires_grad) or ( + is_fp8_activation_recompute_enabled() and not in_fp8_activation_recompute_phase() + ) weights_for_gemm, new_workspaces = _GroupedLinear._prepare_weights_for_grouped_tensor_gemm( weights, weight_quantizers, @@ -967,6 +975,7 @@ def _grouped_linear_graph_safe_setup( bwd_args.grad_output_quantizers = fwd_args.grad_output_quantizers bwd_args.m_splits_tensor = fwd_args.m_splits_tensor bwd_args.num_gemms = fwd_args.num_gemms + bwd_args.num_bias_args = len(fwd_args.biases) bwd_args.activation_dtype = fwd_args.activation_dtype bwd_args.fp8 = fwd_args.fp8 bwd_args.use_bias = fwd_args.use_bias @@ -1072,7 +1081,9 @@ def _grouped_linear_graph_safe_backward( grad_output_view = grad_output.contiguous().view(-1, grad_output.shape[-1]) dy_2d = cast_if_needed(grad_output_view, args.activation_dtype) dbias_packed = None - if args.fp8: + grouped_dy = None + needs_grouped_dy = args.requires_dgrad or args.weights_requires_grad + if args.fp8 and needs_grouped_dy: grad_output_quantizer = args.grad_output_quantizers[0] grad_output_quantizer.set_usage( rowwise=args.requires_dgrad, @@ -1100,7 +1111,7 @@ def _grouped_linear_graph_safe_backward( split_sizes, tensor_offsets=output_tensor_offsets, ) - else: + elif needs_grouped_dy: grouped_dy = _GroupedLinear._make_grouped_tensor( dy_2d, num_gemms=N, @@ -1118,8 +1129,7 @@ def _grouped_linear_graph_safe_backward( else: grad_bias_args = [dbias_packed[i].to(dtype=args.activation_dtype) for i in range(N)] else: - num_bias_args = 1 if args.single_grouped_bias else N - grad_bias_args = [None] * num_bias_args + grad_bias_args = [None] * args.num_bias_args dgrad = None if args.requires_dgrad: @@ -1226,7 +1236,7 @@ def grouped_gemm_wgrad(inputmats, grad_output_mats, grad_weights): use_split_accumulator=wgrad_use_split_accumulator, accumulate=accumulate, ) - return None, [None] * N, None + return None, (), None if args.wgrad_store is not None and args.wgrad_store.delay_wgrad_compute(): args.wgrad_store.put([grouped_x, grouped_dy, wgrad_output], grouped_gemm_wgrad) @@ -1354,6 +1364,11 @@ def _prepare_weights_for_grouped_tensor_gemm( raise RuntimeError( "Quantized single grouped weights require quantized grouped GEMM compute." ) + # Preserve usages already enabled on the parameter quantizer. + weight.quantizer.set_usage( + rowwise=True, + columnwise=True if columnwise_usage else None, + ) return weight, new_workspaces if not with_quantized_compute: @@ -1433,8 +1448,13 @@ def _prepare_weights_for_grouped_tensor_gemm( update_ws = is_first_microbatch is None or is_first_microbatch for idx, weight in enumerate(weights): - weight_quantizer = weight_quantizers[idx] - weight_quantizer.set_usage(rowwise=True, columnwise=columnwise_usage) + if isinstance(weight, QuantizedTensorStorage): + weight_quantizer = weight._quantizer + weight_columnwise_usage = True if columnwise_usage else None + else: + weight_quantizer = weight_quantizers[idx] + weight_columnwise_usage = columnwise_usage + weight_quantizer.set_usage(rowwise=True, columnwise=weight_columnwise_usage) weight_fp8, new_workspaces[idx] = quantize_weight( tensor=weight, quantizer=weight_quantizer, @@ -1530,7 +1550,7 @@ def _forward_grouped_tensor(ctx, fwd_args: GroupedLinearFwdArgs) -> Tuple[torch. ctx.inp_shape = fwd_args.inp.shape ctx.backward_objects = bwd_args if fwd_args.fp8 and requires_grad( - fwd_args.inp, fwd_args.weights[0], fwd_args.biases[0] + fwd_args.inp, fwd_args.weights[0], fwd_args.biases[0] if fwd_args.biases else None ): bwd_args.reduce_and_update_bwd_fp8_tensors = ( FP8GlobalStateManager.is_first_fp8_module() @@ -1554,6 +1574,7 @@ def forward( fwd_args.inp = inp num_weight_args = 1 if fwd_args.single_grouped_weight else num_gemms fwd_args.weights = list(weights_and_biases[:num_weight_args]) + # Only bias=False omits bias inputs; returned biases retain their autograd edges. fwd_args.biases = list(weights_and_biases[num_weight_args:]) if fwd_args.use_grouped_tensor_path: @@ -1568,7 +1589,9 @@ def forward( ctx.save_for_backward(*tensors_to_save) ctx.tensor_objects = tensor_objects ctx.backward_objects = bwd_args - if fwd_args.fp8 and requires_grad(inp, fwd_args.weights[0], fwd_args.biases[0]): + if fwd_args.fp8 and requires_grad( + inp, fwd_args.weights[0], fwd_args.biases[0] if fwd_args.biases else None + ): bwd_args.reduce_and_update_bwd_fp8_tensors = ( FP8GlobalStateManager.is_first_fp8_module() ) @@ -1650,7 +1673,8 @@ class GroupedLinear(TransformerEngineBaseModule): the model is trained with lower precision and the original FP32 parameters would not fit in GPU memory. delay_wgrad_compute : bool, default = False - Whether to delay weight gradient computation + Whether to delay weight gradient computation until ``backward_dw()``. + Bias gradients are computed and accumulated during the main backward. save_original_input : bool, default = False If set to ``True``, always saves the original input tensor rather than the cast tensor. In some scenarios, the input tensor is used by multiple modules, @@ -1844,13 +1868,13 @@ def __init__( self.reset_parameters(defer_init=is_meta) if self.wgrad_store.delay_wgrad_compute(): - for name, param in self.named_parameters(): - if name in ("weight", "bias"): - param.skip_backward_post_hook = True - continue + # Only weights are deferred. Biases use normal autograd accumulation hooks. + grouped_weight = getattr(self, "weight", None) + if grouped_weight is not None: + grouped_weight.skip_backward_post_hook = True + else: for i in range(self.num_gemms): - if name in (f"weight{i}", f"bias{i}"): - param.skip_backward_post_hook = True + getattr(self, f"weight{i}").skip_backward_post_hook = True def set_meta_tensor(self, fwd: bool, recipe: Recipe) -> None: """Init scales and amaxes for fwd | bwd.""" @@ -2282,11 +2306,10 @@ def forward( f"does not match number of GEMMs ({num_gemms})." ) - if FP8GlobalStateManager.fp8_graph_capturing(): - skip_fp8_weight_update = ( - FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor - ) - else: + skip_fp8_weight_update = ( + FP8GlobalStateManager.quantization_state.skip_fp8_weight_update_tensor + ) + if skip_fp8_weight_update is not None and not FP8GlobalStateManager.fp8_graph_capturing(): skip_fp8_weight_update = None if skip_fp8_weight_update is not None: is_first_microbatch = False @@ -2298,15 +2321,17 @@ def forward( try: weight_tensors = self._get_weight_tensors() - bias_tensors = self._get_bias_tensors() + # Keep real bias inputs even when returned for the caller to apply: they + # preserve the output's autograd dependency when input/weight are frozen. + bias_tensors = self._get_bias_tensors() if self.use_bias else [] use_grouped_bias = self.use_bias and self.single_grouped_bias - quantizers = self._get_quantizers() if not debug else self._get_debug_quantizers() - + quantizers = None if debug: + quantizers = self._get_debug_quantizers() if self.no_debug_features_active(list(chain(*quantizers))): debug = False - quantizers = self._get_quantizers() + quantizers = None if debug and (self.single_grouped_weight or self.single_grouped_bias): raise RuntimeError( "TE debug features do not support single grouped parameters. DebugQuantizer " @@ -2315,6 +2340,20 @@ def forward( "features for this GroupedLinear." ) + cpu_offloading = is_cpu_offload_enabled() + # Normal module execution does not quantize outputs. Debug supplies per-expert + # output quantizers and must retain the full split-path preparation. + output_quantizers = quantizers[2] if quantizers is not None else () + grouped_tensor_supported = self._grouped_tensor_path_supported( + debug, cpu_offloading, output_quantizers + ) + if quantizers is None: + quantizers = ( + self._get_grouped_runtime_quantizers() + if grouped_tensor_supported + else self._get_quantizers() + ) + ( input_quantizers, weight_quantizers, @@ -2383,42 +2422,7 @@ def forward( ) save_original_input = False - cpu_offloading = is_cpu_offload_enabled() - use_grouped_tensor_path = False - if self.use_grouped_tensor and not ( - self.fp8_calibration - or debug - or cpu_offloading - or any(q is not None for q in output_quantizers) - ): - if ( - self.fp8 - and _recipe.float8_block_scaling() - and (10, 0) <= get_device_compute_capability() <= (11, 0) - ): - raise RuntimeError( - "use_grouped_tensor=True does not support the FP8 block-scaling recipe on" - " Blackwell GPUs: the native grouped FP8 block-scaling path is Hopper-only." - " Set use_grouped_tensor=False, or unset" - " NVTE_GROUPED_LINEAR_USE_FUSED_GROUPED_GEMM if it enabled this path, to" - " use the MXFP8-emulated path on Blackwell." - ) - use_grouped_tensor_path = is_module_grouped_tensor_path_supported( - _recipe, - self.activation_dtype, - ) - if ( - self.use_grouped_tensor - and not use_grouped_tensor_path - and (self.single_grouped_weight or use_grouped_bias) - ): - raise RuntimeError( - "Single grouped parameters require the native grouped-tensor path, but the" - " active device, cuBLASLt version, quantization recipe, or GroupedLinear" - " feature configuration does not support it. Disable single_grouped_weight and" - " single_grouped_bias to allow the split-quantize fallback." - ) - if use_grouped_tensor_path: + if grouped_tensor_supported: if m_splits.device.type != "cuda": raise ValueError( "The native grouped_tensor path requires CUDA m_splits. Pass a CUDA int64 " @@ -2459,7 +2463,7 @@ def forward( is_first_microbatch=is_first_microbatch, cache_weight=cache_weight, # fused GroupedTensor path - use_grouped_tensor_path=use_grouped_tensor_path, + use_grouped_tensor_path=grouped_tensor_supported, single_grouped_weight=self.single_grouped_weight, single_grouped_bias=use_grouped_bias, # misc @@ -2490,6 +2494,9 @@ def forward( self.end_forward() if self.return_bias: + # bias=False still returns the legacy list of empty tensors when requested. + if not self.use_bias: + bias_tensors = self._get_bias_tensors() if use_grouped_bias: return out, bias_tensors[0] return out, [cast_if_needed(b, self.activation_dtype) for b in bias_tensors] @@ -2505,7 +2512,7 @@ def backward_dw(self): if self.wgrad_store.context is None or self.wgrad_store.context.empty(): return with get_nvtx_range_context("_GroupedLinear_wgrad"): - (_, grad_biases_, _), tensor_list = self.wgrad_store.pop() + (_, grad_biases, _), tensor_list = self.wgrad_store.pop() wgrad_output = tensor_list[2] weight_params = self._get_weight_tensors() if not self.fuse_wgrad_accumulation: @@ -2516,23 +2523,21 @@ def backward_dw(self): else: for i in range(self.num_gemms): weight_params[i].grad = wgrad_output[i].to(weight_params[i].dtype) - has_grad_biases = [ - grad_bias is not None and grad_bias.numel() != 0 for grad_bias in grad_biases_ - ] - if self.use_bias and any(has_grad_biases): - if self.use_grouped_tensor: - raise RuntimeError( - "GroupedLinear(use_grouped_tensor=True) fell back to the split-quantize " - "path, which produced per-expert bias gradients during delayed wgrad. " - "This implicit fallback is unsupported with delay_wgrad_compute=True. " - "Use a configuration supported by the grouped-tensor path, or set " - "use_grouped_tensor=False to select the legacy path explicitly." - ) - bias_params = [getattr(self, f"bias{i}") for i in range(self.num_gemms)] - for i in range(self.num_gemms): - if has_grad_biases[i] and bias_params[i].grad is None: - bias_params[i].grad = grad_biases_[i].to(bias_params[i].dtype) - del grad_biases_ + if ( + self.use_grouped_tensor + and self.apply_bias + and any( + grad_bias is not None and grad_bias.numel() != 0 for grad_bias in grad_biases + ) + ): + raise RuntimeError( + "GroupedLinear(use_grouped_tensor=True) fell back to the split-quantize " + "path, which produced per-expert bias gradients during delayed wgrad. " + "This implicit fallback is unsupported with delay_wgrad_compute=True. " + "Use a configuration supported by the grouped-tensor path, or set " + "use_grouped_tensor=False to select the legacy path explicitly." + ) + del grad_biases del wgrad_output del tensor_list self._trigger_wgrad_accumulation_and_reduce_hooks() @@ -2610,6 +2615,75 @@ def _get_bias_tensors(self) -> List[torch.Tensor]: return [grouped_bias] return [getattr(self, f"bias{i}") for i in range(self.num_gemms)] + def _grouped_tensor_path_supported(self, debug, cpu_offloading, output_quantizers): + """Resolve the actual path after prepare_forward installs the active recipe.""" + recipe = FP8GlobalStateManager.get_fp8_recipe() if self.fp8 else None + grouped_tensor_supported = False + if self.use_grouped_tensor and not ( + self.fp8_calibration + or debug + or cpu_offloading + or (output_quantizers and any(q is not None for q in output_quantizers)) + ): + if ( + self.fp8 + and recipe.float8_block_scaling() + and (10, 0) <= get_device_compute_capability() <= (11, 0) + ): + raise RuntimeError( + "use_grouped_tensor=True does not support the FP8 block-scaling recipe on " + "Blackwell GPUs: the native grouped FP8 block-scaling path is Hopper-only. " + "Set use_grouped_tensor=False, or unset " + "NVTE_GROUPED_LINEAR_USE_FUSED_GROUPED_GEMM if it enabled this path, to use " + "the MXFP8-emulated path on Blackwell." + ) + grouped_tensor_supported = is_module_grouped_tensor_path_supported( + recipe, + self.activation_dtype, + ) + if ( + self.use_grouped_tensor + and not grouped_tensor_supported + and (self.single_grouped_weight or (self.use_bias and self.single_grouped_bias)) + ): + raise RuntimeError( + "Single grouped parameters require the native grouped-tensor path, but the active " + "device, cuBLASLt version, quantization recipe, or GroupedLinear feature " + "configuration does not support it. Disable single_grouped_weight and " + "single_grouped_bias to allow the split-quantize fallback." + ) + return grouped_tensor_supported + + def _get_grouped_runtime_quantizers(self): + """Get only the quantizers consumed by native grouped execution. + + Recipe metadata and initialization still own num_gemms slots per role. Runtime + input/dY use the first slot; discrete weights still need one slot per expert. + """ + input_quantizers, grad_output_quantizers = [None], [None] + num_weights = 1 if self.single_grouped_weight else self.num_gemms + weight_quantizers = [None] * num_weights + if self.fp8 or self.primary_weights_in_fp8: + weight_quantizers = [ + self.quantizers["scaling_fwd"][ + self._offsets["weight"] + i * self._num_fp8_tensors_per_gemm["fwd"] + ] + for i in range(num_weights) + ] + for quantizer in weight_quantizers: + quantizer.internal = not self.primary_weights_in_fp8 + if self.fp8: + input_quantizer = self.quantizers["scaling_fwd"][self._offsets["input"]] + input_quantizer.internal = True + input_quantizer.optimize_for_gemm = True + input_quantizers = [input_quantizer] + if torch.is_grad_enabled(): + grad_output_quantizer = self.quantizers["scaling_bwd"][self._offsets["input"]] + grad_output_quantizer.internal = True + grad_output_quantizer.optimize_for_gemm = True + grad_output_quantizers = [grad_output_quantizer] + return input_quantizers, weight_quantizers, (), (), (), grad_output_quantizers + def _get_weight_quantizers(self) -> List[Quantizer]: """Get the weight quantizers of the module.""" if not self.fp8 and not self.fp8_calibration and not self.primary_weights_in_fp8: diff --git a/transformer_engine/pytorch/ops/basic/grouped_linear.py b/transformer_engine/pytorch/ops/basic/grouped_linear.py index b4810f6cc84..a3dae11310f 100644 --- a/transformer_engine/pytorch/ops/basic/grouped_linear.py +++ b/transformer_engine/pytorch/ops/basic/grouped_linear.py @@ -785,14 +785,23 @@ def pre_fuser_forward(self, *, requires_grad: bool) -> None: ) weight_requires_grad = requires_grad and weight_requires_grad - # Configure quantizer usages - for group_idx in range(self.num_groups): - input_quantizer = self.get_quantizer("forward", 2 * group_idx) - weight_quantizer = self.get_quantizer("forward", 2 * group_idx + 1) - grad_output_quantizer = self.get_quantizer("backward", group_idx) - input_quantizer.set_usage(rowwise=True, columnwise=weight_requires_grad) - weight_quantizer.set_usage(rowwise=True, columnwise=requires_grad) - grad_output_quantizer.set_usage(rowwise=True, columnwise=weight_requires_grad) + # The adjacent activation may consume these before our own forward/backward. + # Weight usage is only needed when preparing the GEMM operand. + weight = self.weight if self.single_grouped_weight else self.weight0 + dtype = ( + torch.get_autocast_dtype("cuda") if torch.is_autocast_enabled() else weight.dtype + ) + grouped = is_op_fuser_grouped_tensor_path_supported( + FP8GlobalStateManager.get_fp8_recipe(), dtype, self.split_alignment + ) + input_quantizers = (self._input_quantizers[0],) if grouped else self._input_quantizers + grad_output_quantizers = ( + (self._grad_output_quantizers[0],) if grouped else self._grad_output_quantizers + ) + for quantizer in input_quantizers: + quantizer.set_usage(rowwise=True, columnwise=weight_requires_grad) + for quantizer in grad_output_quantizers: + quantizer.set_usage(rowwise=True, columnwise=weight_requires_grad) def reset_recipe_state(self, *, recipe: Optional[Recipe]) -> None: super().reset_recipe_state(recipe=recipe) @@ -865,6 +874,18 @@ def reset_recipe_state(self, *, recipe: Optional[Recipe]) -> None: else: weight.update_quantizer(weight_quantizer.copy()) + # Cache object references for this recipe generation, not tensor workspaces. + # Keep the per-expert slots for legacy split quantization and checkpoint metadata. + self._input_quantizers = tuple( + self.get_quantizer("forward", 2 * i) for i in range(self.num_groups) + ) + self._weight_quantizers = tuple( + self.get_quantizer("forward", 2 * i + 1) for i in range(self.num_groups) + ) + self._grad_output_quantizers = tuple( + self.get_quantizer("backward", i) for i in range(self.num_groups) + ) + def op_forward(self, *args, **kwargs): raise RuntimeError( f"{self.__class__.__name__} operation has " @@ -884,7 +905,7 @@ def op_backward(self, *args, **kwargs): def _get_grouped_weight_for_gemm( self, weight_param: GroupedTensor, - weight_quantizers: list[Optional[Quantizer]], + weight_quantizers: Sequence[Optional[Quantizer]], columnwise_usage: bool, with_quantized_compute: bool, dtype: torch.dtype, @@ -940,7 +961,7 @@ def _get_grouped_weight_for_gemm( def _get_discrete_weights_for_gemm( self, weight_params: Optional[GroupedTensor] | list[torch.Tensor], - weight_quantizers: list[Optional[Quantizer]], + weight_quantizers: Sequence[Optional[Quantizer]], columnwise_usage: bool, with_quantized_compute: bool, dtype: torch.dtype, @@ -952,9 +973,16 @@ def _get_discrete_weights_for_gemm( for w, quantizer in zip(weight_params, weight_quantizers): if not with_quantized_compute: w = maybe_dequantize(w, dtype) - elif with_quantized_compute and not is_quantized_tensor(w): + else: quantizer.set_usage(rowwise=True, columnwise=columnwise_usage) - w = quantizer(w) + if is_quantized_tensor(w): + # Honor the op's requirements without disabling existing parameter usages. + required_columnwise = True if columnwise_usage else None + w.update_usage(rowwise_usage=True, columnwise_usage=required_columnwise) + if w._quantizer is not None: + w._quantizer.set_usage(rowwise=True, columnwise=required_columnwise) + else: + w = quantizer(w) out.append(w) return out @@ -1038,14 +1066,7 @@ def fuser_forward( input_requires_grad = ctx.requires_grad weight_requires_grad = ctx.requires_grad and weight_param.requires_grad - # Quantizers - input_quantizers = [None] * num_groups - weight_quantizers = [None] * num_groups with_quantized_compute = FP8GlobalStateManager.is_fp8_enabled() - if with_quantized_compute: - for group_idx in range(num_groups): - input_quantizers[group_idx] = self.get_quantizer("forward", 2 * group_idx) - weight_quantizers[group_idx] = self.get_quantizer("forward", 2 * group_idx + 1) # Get autocast dtype if needed if torch.is_autocast_enabled(): @@ -1089,13 +1110,17 @@ def fuser_forward( "tensor configuration." ) + weight_quantizers = self._weight_quantizers + if use_grouped_tensor_path and self.single_grouped_weight: + weight_quantizers = (weight_quantizers[0],) + if use_grouped_tensor_path: out, tensors_to_save = self._fuser_forward_grouped_tensor( input_=input_, split_sizes=split_sizes, scales=scales, with_quantized_compute=with_quantized_compute, - input_quantizers=input_quantizers, + input_quantizer=self._input_quantizers[0], weight_quantizers=weight_quantizers, dtype=dtype, input_requires_grad=input_requires_grad, @@ -1110,7 +1135,7 @@ def fuser_forward( split_sizes=split_sizes, scales=scales, with_quantized_compute=with_quantized_compute, - input_quantizers=input_quantizers, + input_quantizers=self._input_quantizers, weight_quantizers=weight_quantizers, dtype=dtype, input_requires_grad=input_requires_grad, @@ -1180,24 +1205,15 @@ def fuser_forward_save_ctx( ctx.save_for_backward(*tensors_to_save[0]) - num_groups = self.num_groups weight_param = self.weight if self.single_grouped_weight else self.weight0 with_quantized_compute = FP8GlobalStateManager.is_fp8_enabled() - input_quantizers = [None] * num_groups - weight_quantizers = [None] * num_groups - grad_output_quantizers = [None] * num_groups - if with_quantized_compute: - for group_idx in range(num_groups): - input_quantizers[group_idx] = self.get_quantizer("forward", 2 * group_idx) - weight_quantizers[group_idx] = self.get_quantizer("forward", 2 * group_idx + 1) - grad_output_quantizers[group_idx] = self.get_quantizer("backward", group_idx) - ctx.use_grouped_tensor_path = use_grouped_tensor_path ctx.with_quantized_compute = with_quantized_compute - ctx.input_quantizers = input_quantizers - ctx.weight_quantizers = weight_quantizers - ctx.grad_output_quantizers = grad_output_quantizers + if use_grouped_tensor_path: + ctx.grad_output_quantizer = self._grad_output_quantizers[0] + else: + ctx.grad_output_quantizers = self._grad_output_quantizers ctx.grad_input_quantizers = None # ``split_sizes``, offset metadata, and related tensors are routed # through ``save_for_backward`` (see ``_fuser_forward_split_quantize`` @@ -1223,8 +1239,8 @@ def _fuser_forward_split_quantize( split_sizes: torch.Tensor, scales: Optional[torch.Tensor], with_quantized_compute: bool, - input_quantizers: list[Optional[Quantizer]], - weight_quantizers: list[Optional[Quantizer]], + input_quantizers: Sequence[Optional[Quantizer]], + weight_quantizers: Sequence[Optional[Quantizer]], dtype: torch.dtype, input_requires_grad: bool, weight_requires_grad: bool, @@ -1339,8 +1355,8 @@ def _fuser_forward_grouped_tensor( split_sizes: torch.Tensor, scales: Optional[torch.Tensor], with_quantized_compute: bool, - input_quantizers: list[Optional[Quantizer]], - weight_quantizers: list[Optional[Quantizer]], + input_quantizer: Optional[Quantizer], + weight_quantizers: Sequence[Optional[Quantizer]], dtype: torch.dtype, input_requires_grad: bool, weight_requires_grad: bool, @@ -1364,8 +1380,6 @@ def _fuser_forward_grouped_tensor( original_shape = list(input_.size()) prequantized_input = with_quantized_compute and isinstance(input_, GroupedTensor) if with_quantized_compute: - input_quantizer = input_quantizers[0] - input_quantizer.set_usage(rowwise=True, columnwise=weight_requires_grad) input_quantizer.optimize_for_gemm = True if prequantized_input: # GroupedTensor forbids reshape and is already in the canonical @@ -1782,7 +1796,7 @@ def _fuser_backward_grouped_tensor( # Optionally get dbias is fusion available with bgrad_group_quantize dbias_packed = None if with_quantized_compute: - grad_output_quantizer = ctx.grad_output_quantizers[0] + grad_output_quantizer = ctx.grad_output_quantizer grad_output_quantizer.set_usage( rowwise=ctx.input_requires_grad, columnwise=ctx.weight_requires_grad, diff --git a/transformer_engine/pytorch/quantized_tensor.py b/transformer_engine/pytorch/quantized_tensor.py index f6f2794bdf9..b21baea9db7 100644 --- a/transformer_engine/pytorch/quantized_tensor.py +++ b/transformer_engine/pytorch/quantized_tensor.py @@ -330,23 +330,29 @@ def restore_from_saved( Note: please use `restore_from_func_ctx` instead if you are restoring tensors from a function context to make sure tensor_objects is detached and its memory can be freed """ tensor_objects = [] + cursor = 0 for tensor in tensors: if tensor is None or isinstance(tensor, torch.Tensor): - tensor_objects.append(saved_tensors[0]) - saved_tensors = saved_tensors[1:] + tensor_objects.append(saved_tensors[cursor]) + cursor += 1 elif isinstance(tensor, _SavedQuantizedTensor): count = len(tensor.inner_names) - inner = dict(zip(tensor.inner_names, saved_tensors[:count])) + inner = dict(zip(tensor.inner_names, saved_tensors[cursor : cursor + count])) tensor_objects.append( QuantizedTensorStorage.__tensor_unflatten__(inner, tensor.metadata, None, None) ) - saved_tensors = saved_tensors[count:] + cursor += count else: + # Storage restorers receive the full remaining list/tuple and own the + # returned remainder. Preserve that protocol for custom implementations. + if cursor: + saved_tensors = saved_tensors[cursor:] saved_tensors = tensor.restore_from_saved(saved_tensors) + cursor = 0 tensor_objects.append(tensor) if return_saved_tensors: - return tensor_objects, saved_tensors + return tensor_objects, saved_tensors[cursor:] if cursor else saved_tensors return tensor_objects