diff --git a/examples/pytorch/mnist/main.py b/examples/pytorch/mnist/main.py index 3754e3643b..24081fb259 100644 --- a/examples/pytorch/mnist/main.py +++ b/examples/pytorch/mnist/main.py @@ -76,7 +76,10 @@ def calibrate(model, device, test_loader, fp8): with torch.no_grad(): for data, target in test_loader: data, target = data.to(device), target.to(device) - with te.autocast(enabled=fp8, calibrating=True): + with te.autocast( + enabled=fp8, + calibration_config=te.QuantizationCalibrationConfig(), + ): output = model(data) diff --git a/qa/L0_pytorch_unittest/test.sh b/qa/L0_pytorch_unittest/test.sh index a78a99d7f9..a7162cd13d 100644 --- a/qa/L0_pytorch_unittest/test.sh +++ b/qa/L0_pytorch_unittest/test.sh @@ -41,6 +41,7 @@ python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_nvfp4.xml $TE_PA python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_mxfp8.xml $TE_PATH/tests/pytorch/mxfp8 || test_fail "test_mxfp8" python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_weight_swizzle_in_layers.xml $TE_PATH/tests/pytorch/test_weight_swizzle_in_layers.py || test_fail "test_weight_swizzle_in_layers.py" python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_quantized_tensor.xml $TE_PATH/tests/pytorch/test_quantized_tensor.py || test_fail "test_quantized_tensor.py" +python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_ptq_calibration_metadata_buffering.xml $TE_PATH/tests/pytorch/test_ptq_calibration_metadata_buffering.py || test_fail "test_ptq_calibration_metadata_buffering.py" python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_torch_compile.xml $TE_PATH/tests/pytorch/test_torch_compile.py || test_fail "test_torch_compile.py" python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_float8blockwisetensor.xml $TE_PATH/tests/pytorch/test_float8blockwisetensor.py || test_fail "test_float8blockwisetensor.py" python3 -m pytest --tb=auto --junitxml=$XML_LOG_DIR/pytest_test_float8_blockwise_scaling_exact.xml $TE_PATH/tests/pytorch/test_float8_blockwise_scaling_exact.py || test_fail "test_float8_blockwise_scaling_exact.py" diff --git a/tests/pytorch/test_ptq_calibration_metadata_buffering.py b/tests/pytorch/test_ptq_calibration_metadata_buffering.py new file mode 100644 index 0000000000..8c17fb02aa --- /dev/null +++ b/tests/pytorch/test_ptq_calibration_metadata_buffering.py @@ -0,0 +1,852 @@ +# Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# See LICENSE for license information. + +import dataclasses +import inspect +from types import SimpleNamespace + +import pytest +import torch +from torch.utils.checkpoint import checkpoint + +from transformer_engine.common.recipe import DelayedScaling, Float8CurrentScaling +from transformer_engine.pytorch import is_fp8_available, is_nvfp4_available +from transformer_engine.pytorch import distributed as te_distributed +from transformer_engine.pytorch.constants import DType +from transformer_engine.pytorch.graph import make_graphed_callables +from transformer_engine.pytorch.module import GroupedLinear, LayerNormLinear, LayerNormMLP, Linear +from transformer_engine.pytorch.module import _common +from transformer_engine.pytorch.module import grouped_linear +from transformer_engine.pytorch.module import layernorm_mlp +from transformer_engine.pytorch.quantization import ( + FP8GlobalStateManager, + FP8GlobalState, + TEAutocastState, + QuantizationCalibrationConfig, + autocast, + fp8_autocast, +) +from transformer_engine.pytorch.quantized_tensor import QuantizedTensorStorage, Quantizer +from transformer_engine.pytorch.tensor.float8_blockwise_tensor import Float8BlockQuantizer +from transformer_engine.pytorch.tensor.float8_tensor import ( + Float8CurrentScalingQuantizer, + Float8Quantizer, +) +from transformer_engine.pytorch.tensor.hybrid_tensor import HybridQuantizer +from transformer_engine.pytorch.tensor.identity_tensor import IdentityQuantizer +from transformer_engine.pytorch.tensor.mxfp8_tensor import MXFP8Quantizer +from transformer_engine.pytorch.tensor.nvfp4_tensor import NVFP4Quantizer +from transformer_engine.pytorch.tensor.utils import get_quantization_recipe_name + +nvfp4_available, reason_for_no_nvfp4 = is_nvfp4_available(return_reason=True) +fp8_available, reason_for_no_fp8 = is_fp8_available(return_reason=True) + + +@pytest.fixture(autouse=True) +def reset_quantization_state(): + yield + FP8GlobalStateManager.reset() + + +def _make_test_quantizer(quantizer_cls): + """Construct a quantizer without allocating recipe-specific CUDA state.""" + quantizer = object.__new__(quantizer_cls) + Quantizer.__init__(quantizer, rowwise=True, columnwise=False) + return quantizer + + +def _make_test_quantized_storage(**metadata): + """Construct bare quantized storage carrying only the requested metadata.""" + storage = QuantizedTensorStorage() + for name, value in metadata.items(): + setattr(storage, name, value) + return storage + + +def test_calibration_api_additions_preserve_existing_parameter_order(): + state_fields = [field.name for field in dataclasses.fields(FP8GlobalState)] + assert state_fields[-1] == "calibration_config" + assert state_fields[:2] == ["fp8_enabled", "fp8_calibration"] + + assert list(inspect.signature(FP8GlobalStateManager.autocast_enter).parameters) == [ + "enabled", + "calibrating", + "fp8_recipe", + "fp8_group", + "_graph", + "calibration_config", + ] + assert list(inspect.signature(autocast).parameters) == [ + "enabled", + "calibrating", + "recipe", + "amax_reduction_group", + "_graph", + "calibration_config", + ] + assert list(inspect.signature(fp8_autocast).parameters) == [ + "enabled", + "calibrating", + "fp8_recipe", + "fp8_group", + "_graph", + "calibration_config", + ] + assert list(inspect.signature(make_graphed_callables).parameters)[-2:] == [ + "capture_time_hooks", + "calibration_config", + ] + assert not inspect.signature(FP8GlobalStateManager.get_autocast_state).parameters + assert list(inspect.signature(FP8GlobalStateManager.set_autocast_state).parameters) == ["state"] + autocast_state = FP8GlobalStateManager.get_autocast_state() + assert isinstance(autocast_state, TEAutocastState) + assert [field.name for field in dataclasses.fields(autocast_state)] == [ + "fp8_enabled", + "fp8_calibration", + "calibration_config", + "fp8_recipe", + "fp8_distributed_group", + "is_first_fp8_module", + "fp8_graph_capturing", + ] + + +def test_activation_recompute_detection_uses_te_marker(monkeypatch): + monkeypatch.setattr( + te_distributed, + "in_fp8_activation_recompute_phase", + lambda: True, + ) + + assert _common._is_in_activation_recompute_phase() + + +def test_calibrating_argument_enables_default_calibration_config(): + assert FP8GlobalStateManager.get_calibration_config() is None + with autocast(enabled=False, calibrating=True): + assert FP8GlobalStateManager.quantization_state.fp8_calibration + assert FP8GlobalStateManager.get_calibration_config() == QuantizationCalibrationConfig() + assert FP8GlobalStateManager.get_calibration_config() is None + + +def test_explicit_calibration_config_is_active_in_autocast(): + config = QuantizationCalibrationConfig(transformer_engine_calibration_decay=0.5) + with autocast(enabled=False, calibration_config=config): + assert FP8GlobalStateManager.quantization_state.fp8_calibration + assert FP8GlobalStateManager.get_calibration_config() is config + assert FP8GlobalStateManager.get_autocast_state().calibration_config is config + + +def test_nested_autocast_restores_custom_calibration_config(): + config = QuantizationCalibrationConfig(transformer_engine_calibration_decay=0.5) + with autocast(enabled=False, calibration_config=config): + with autocast(enabled=False): + assert FP8GlobalStateManager.get_calibration_config() is None + assert FP8GlobalStateManager.get_calibration_config() is config + + +def test_global_calibration_boolean_enables_default_config(): + qstate = FP8GlobalStateManager.quantization_state + qstate.fp8_calibration = True + try: + assert FP8GlobalStateManager.get_calibration_config() == QuantizationCalibrationConfig() + finally: + qstate.fp8_calibration = False + + +def test_autocast_enter_preserves_calibrating_boolean_api(): + saved_state = FP8GlobalStateManager.get_autocast_state() + try: + FP8GlobalStateManager.autocast_enter(False, True) + assert FP8GlobalStateManager.is_fp8_calibration() + assert FP8GlobalStateManager.get_calibration_config() == QuantizationCalibrationConfig() + finally: + FP8GlobalStateManager.autocast_exit(False, False) + FP8GlobalStateManager.set_autocast_state(saved_state) + + +def test_calibrating_argument_accepts_explicit_calibration_config(): + config = QuantizationCalibrationConfig(transformer_engine_calibration_decay=0.5) + with autocast( + enabled=False, + calibrating=True, + calibration_config=config, + ): + assert FP8GlobalStateManager.get_calibration_config() is config + + +def test_calibration_config_rejects_negative_decay(): + with pytest.raises( + ValueError, + match="transformer_engine_calibration_decay must be non-negative", + ): + QuantizationCalibrationConfig(transformer_engine_calibration_decay=-0.1) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +@pytest.mark.parametrize( + ("module_name", "expected_buffer_count"), + ( + ("linear", 2), + ("layernorm_linear", 2), + ("layernorm_mlp", 4), + ("grouped_linear", 4), + ), +) +@pytest.mark.parametrize("enabled", (False, True)) +def test_calibration_config_registers_module_scaling_factor_buffers( + module_name, expected_buffer_count, enabled +): + module_kwargs = { + "params_dtype": torch.bfloat16, + "device": "cuda", + "bias": False, + } + if module_name == "linear": + module = Linear(32, 32, **module_kwargs) + elif module_name == "layernorm_linear": + module = LayerNormLinear(32, 32, **module_kwargs) + elif module_name == "layernorm_mlp": + module = LayerNormMLP(32, 32, **module_kwargs) + else: + module = GroupedLinear(2, 32, 32, use_grouped_tensor=False, **module_kwargs) + + inp = torch.randn((16, 32), dtype=torch.bfloat16, device="cuda") + with autocast( + enabled=enabled, + recipe=Float8CurrentScaling(), + calibration_config=QuantizationCalibrationConfig(), + ): + if module_name == "grouped_linear": + module(inp, [8, 8]) + else: + module(inp) + + buffers = { + name: value for name, value in module.named_buffers() if name.endswith("_te_ptq_calibrated") + } + assert len(buffers) == expected_buffer_count + assert all(torch.isfinite(value).all() for value in buffers.values()) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +def test_linear_calibration_config_applies_activation_decay_only(): + module = Linear(32, 32, params_dtype=torch.bfloat16, device="cuda", bias=False) + calibration_config = QuantizationCalibrationConfig(transformer_engine_calibration_decay=0.5) + buffer_suffix = "_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated" + + with autocast( + enabled=False, + recipe=Float8CurrentScaling(), + calibration_config=calibration_config, + ): + module(torch.full((16, 32), 448.0, dtype=torch.bfloat16, device="cuda")) + + torch.testing.assert_close( + module.get_buffer(f"input{buffer_suffix}"), torch.ones(1, device="cuda") + ) + weight_scale = module.get_buffer(f"weight{buffer_suffix}").clone() + + with autocast( + enabled=False, + recipe=Float8CurrentScaling(), + calibration_config=calibration_config, + ): + module(torch.full((16, 32), 112.0, dtype=torch.bfloat16, device="cuda")) + + torch.testing.assert_close( + module.get_buffer(f"input{buffer_suffix}"), torch.full((1,), 0.5, device="cuda") + ) + torch.testing.assert_close(module.get_buffer(f"weight{buffer_suffix}"), weight_scale) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +@pytest.mark.parametrize("use_reentrant", (False, True)) +def test_external_activation_recomputation_does_not_update_calibration(use_reentrant): + module = Linear(32, 32, params_dtype=torch.bfloat16, device="cuda", bias=False) + calibration_config = QuantizationCalibrationConfig(transformer_engine_calibration_decay=0.5) + recipe = Float8CurrentScaling() + buffer_name = "input_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated" + + with ( + torch.no_grad(), + autocast(enabled=False, recipe=recipe, calibration_config=calibration_config), + ): + module(torch.full((16, 32), 448.0, dtype=torch.bfloat16, device="cuda")) + + def checkpointed_forward(inp): + with autocast(enabled=False, recipe=recipe, calibration_config=calibration_config): + return module(inp) + + inp = torch.full( + (16, 32), + 56.0, + dtype=torch.bfloat16, + device="cuda", + requires_grad=True, + ) + out = checkpoint(checkpointed_forward, inp, use_reentrant=use_reentrant) + scale_after_forward = module.get_buffer(buffer_name).clone() + torch.testing.assert_close(scale_after_forward, torch.full((1,), 0.5, device="cuda")) + + out.sum().backward() + + torch.testing.assert_close(module.get_buffer(buffer_name), scale_after_forward) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +def test_layernorm_mlp_recomputation_does_not_update_calibration(monkeypatch): + recomputation_states = [] + is_in_recompute = _common._is_in_activation_recompute_phase + + def record_recomputation_state(): + state = is_in_recompute() + recomputation_states.append(state) + return state + + monkeypatch.setattr( + layernorm_mlp, + "_is_in_activation_recompute_phase", + record_recomputation_state, + ) + module = LayerNormMLP( + 32, + 32, + params_dtype=torch.bfloat16, + device="cuda", + bias=False, + activation="relu", + checkpoint=True, + ) + calibration_config = QuantizationCalibrationConfig(transformer_engine_calibration_decay=0.5) + torch.manual_seed(123) + calibration_input = torch.randn((16, 32), dtype=torch.bfloat16, device="cuda") + with torch.no_grad(): + module.layer_norm_weight.fill_(448.0) + with ( + torch.no_grad(), + autocast( + enabled=False, + recipe=Float8CurrentScaling(), + calibration_config=calibration_config, + ), + ): + module(calibration_input) + + buffer_name = "fc1_input_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated" + previous_scale = module.get_buffer(buffer_name).clone() + with torch.no_grad(): + module.layer_norm_weight.fill_(56.0) + inp = calibration_input.detach().clone().requires_grad_() + + with autocast( + enabled=False, + recipe=Float8CurrentScaling(), + calibration_config=calibration_config, + ): + out = module(inp) + + calibration_buffers = { + name: value.clone() + for name, value in module.named_buffers() + if name.endswith("_te_ptq_calibrated") + } + assert calibration_buffers + torch.testing.assert_close( + calibration_buffers[buffer_name], + previous_scale * 0.5, + ) + + out.sum().backward() + + assert any(recomputation_states) + for name, value in calibration_buffers.items(): + torch.testing.assert_close(module.get_buffer(name), value) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +def test_linear_calibration_config_buffers_delayed_scaling_amax(): + module = Linear(32, 32, params_dtype=torch.bfloat16, device="cuda", bias=False) + + with autocast( + enabled=False, + recipe=DelayedScaling(), + calibration_config=QuantizationCalibrationConfig(), + ): + module(torch.full((16, 32), 2.0, dtype=torch.bfloat16, device="cuda")) + + calibration_buffers = { + name: value for name, value in module.named_buffers() if name.endswith("_te_ptq_calibrated") + } + assert set(calibration_buffers) == { + "input_tensor_amax_fp8_delayed_scaling_te_ptq_calibrated", + "weight_tensor_amax_fp8_delayed_scaling_te_ptq_calibrated", + } + torch.testing.assert_close( + calibration_buffers["input_tensor_amax_fp8_delayed_scaling_te_ptq_calibrated"], + torch.full((1,), 2.0, device="cuda"), + ) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +def test_calibrating_boolean_registers_scaling_factor_buffers(): + module = Linear(32, 32, params_dtype=torch.bfloat16, device="cuda", bias=False) + inp = torch.randn((16, 32), dtype=torch.bfloat16, device="cuda") + + with autocast( + enabled=False, + calibrating=True, + recipe=Float8CurrentScaling(), + ): + module(inp) + + assert ( + len([name for name, _ in module.named_buffers() if name.endswith("_te_ptq_calibrated")]) + == 2 + ) + + +@pytest.mark.parametrize( + ("recipe", "metadata_name", "expected_value"), + ( + ("fp8_current_scaling", "scale_inv", 0.25), + ("fp8_delayed_scaling", "amax", 448.0), + ("nvfp4", "amax", 2688.0), + ("nvfp4_rowwise", "amax_rowwise", 1344.0), + ), +) +def test_scale_buffer_info_selects_recipe_metadata(recipe, metadata_name, expected_value): + tensor = _make_test_quantized_storage( + _scale_inv=torch.tensor([0.25], dtype=torch.float32), + _amax_rowwise=torch.tensor([2688.0 if recipe == "nvfp4" else 1344.0], dtype=torch.float32), + ) + if recipe == "fp8_delayed_scaling": + quantizer = _make_test_quantizer(Float8Quantizer) + quantizer.amax = torch.tensor([0.0], dtype=torch.float32) + tensor = torch.tensor([expected_value]) + elif recipe == "fp8_current_scaling": + quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + else: + quantizer = _make_test_quantizer(NVFP4Quantizer) + quantizer.row_scaled_nvfp4 = recipe == "nvfp4_rowwise" + + assert get_quantization_recipe_name(quantizer) == recipe + quantizer.calibrate(tensor) + buffers = _common._get_calibration_metadata_buffers("input", quantizer) + buffer_name = f"input_tensor_{metadata_name}_{recipe}_te_ptq_calibrated" + value = buffers[buffer_name] + + torch.testing.assert_close(value, torch.tensor([expected_value])) + assert value is quantizer._calibration_state[metadata_name] + + +@pytest.mark.parametrize("recipe", ("mxfp8", "fp8_block_scaling")) +def test_scale_buffer_info_skips_non_global_scaling_recipes(recipe): + tensor = SimpleNamespace(_rowwise_scale_inv=torch.ones(2, 2)) + quantizer_cls = MXFP8Quantizer if recipe == "mxfp8" else Float8BlockQuantizer + quantizer = _make_test_quantizer(quantizer_cls) + + assert get_quantization_recipe_name(quantizer) == recipe + quantizer.calibrate(tensor) + assert not _common._get_calibration_metadata_buffers("input", quantizer) + + +def test_custom_quantizer_defaults_to_no_calibration_metadata(): + quantizer = Quantizer(rowwise=True, columnwise=False) + + assert get_quantization_recipe_name(quantizer) == "" + assert not _common._get_calibration_metadata_buffers("input", quantizer) + + +def test_hybrid_quantizer_calibration_is_noop(): + quantizer = HybridQuantizer( + rowwise_quantizer=IdentityQuantizer(), + columnwise_quantizer=IdentityQuantizer(), + ) + + quantizer.calibrate(torch.ones(1)) + + assert not quantizer._calibration_state + + +def test_resolve_calibration_quantizer_prefers_tensor_owner_and_unwraps_parent(): + parent_quantizer = object() + tensor_quantizer = SimpleNamespace(parent_quantizer=parent_quantizer) + tensor = SimpleNamespace(_quantizer=tensor_quantizer) + + assert _common._resolve_calibration_quantizer(tensor, object()) is parent_quantizer + + +def test_quantizer_calibration_state_is_keyed_by_quantized_metadata(): + quantizer = Quantizer(rowwise=True, columnwise=False) + + quantizer._update_calibration_value("amax", torch.tensor([2.0]), calibration_decay=0.0) + quantizer._update_calibration_value("scale_inv", torch.tensor([0.5]), calibration_decay=0.0) + + assert set(quantizer._calibration_state) == {"amax", "scale_inv"} + torch.testing.assert_close(quantizer._calibration_state["amax"], torch.tensor([2.0])) + torch.testing.assert_close(quantizer._calibration_state["scale_inv"], torch.tensor([0.5])) + + +def test_grouped_calibration_metadata_buffers_are_per_gemm(): + inputs = [ + _make_test_quantized_storage(_scale_inv=torch.tensor([0.25])), + _make_test_quantized_storage(_scale_inv=torch.tensor([0.5])), + ] + weights = [ + _make_test_quantized_storage(_scale_inv=torch.tensor([0.75])), + _make_test_quantized_storage(_scale_inv=torch.tensor([1.0])), + ] + input_quantizers = [ + _make_test_quantizer(Float8CurrentScalingQuantizer), + _make_test_quantizer(Float8CurrentScalingQuantizer), + ] + weight_quantizers = [ + _make_test_quantizer(Float8CurrentScalingQuantizer), + _make_test_quantizer(Float8CurrentScalingQuantizer), + ] + calibration_buffers = {} + + grouped_linear._calibrate_grouped_tensors( + inputs, + weights, + input_quantizers, + weight_quantizers, + transformer_engine_calibration_decay=0.0, + ) + grouped_linear._update_grouped_calibration_metadata_buffers( + calibration_buffers, + inputs, + weights, + input_quantizers, + weight_quantizers, + ) + + assert set(calibration_buffers) == { + "input_gemm0_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated", + "input_gemm1_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated", + "weight_gemm0_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated", + "weight_gemm1_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated", + } + torch.testing.assert_close( + calibration_buffers["input_gemm1_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated"], + torch.tensor([0.5]), + ) + assert ( + calibration_buffers["input_gemm1_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated"] + is input_quantizers[1]._calibration_state["scale_inv"] + ) + + +def test_grouped_calibration_applies_decay_only_to_activations(): + input_quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + input_quantizer._calibration_state = {"scale_inv": torch.tensor([4.0])} + weight_quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + weight_quantizer._calibration_state = {"scale_inv": torch.tensor([4.0])} + + grouped_linear._calibrate_grouped_tensors( + [_make_test_quantized_storage(_scale_inv=torch.tensor([1.0]))], + [_make_test_quantized_storage(_scale_inv=torch.tensor([1.0]))], + [input_quantizer], + [weight_quantizer], + transformer_engine_calibration_decay=0.5, + ) + + torch.testing.assert_close(input_quantizer._calibration_state["scale_inv"], torch.tensor([2.0])) + torch.testing.assert_close( + weight_quantizer._calibration_state["scale_inv"], torch.tensor([1.0]) + ) + + +def test_calibration_supports_legacy_custom_quantizer_signature(): + class LegacyCustomQuantizer: + def calibrate(self, tensor): + self.observed_tensor = tensor + + tensor = torch.tensor([1.0]) + quantizer = LegacyCustomQuantizer() + grouped_linear._calibrate_grouped_tensors( + [tensor], + [], + [quantizer], + [], + transformer_engine_calibration_decay=0.5, + ) + + assert quantizer.observed_tensor is tensor + + +def test_grouped_calibration_metadata_uses_per_gemm_delayed_scaling_amax(): + quantizers = [] + for amax in (1.0, 2.0): + quantizer = _make_test_quantizer(Float8Quantizer) + quantizer.amax = torch.tensor([amax]) + quantizers.append(quantizer) + calibration_buffers = {} + + grouped_linear._calibrate_grouped_tensors( + [torch.tensor([1.0]), torch.tensor([2.0])], + [torch.tensor([1.0]), torch.tensor([2.0])], + quantizers, + quantizers, + transformer_engine_calibration_decay=0.0, + ) + grouped_linear._update_grouped_calibration_metadata_buffers( + calibration_buffers, + [torch.tensor([1.0]), torch.tensor([2.0])], + [torch.tensor([1.0]), torch.tensor([2.0])], + quantizers, + quantizers, + ) + + torch.testing.assert_close( + calibration_buffers["input_gemm0_tensor_amax_fp8_delayed_scaling_te_ptq_calibrated"], + torch.tensor([1.0]), + ) + torch.testing.assert_close( + calibration_buffers["input_gemm1_tensor_amax_fp8_delayed_scaling_te_ptq_calibrated"], + torch.tensor([2.0]), + ) + + +@pytest.mark.parametrize( + ("observed_scale", "expected_scale"), + ( + # Decayed max is greater than the observed. + (1.0, 2.0), + # Decayed max is less than the observed. + (3.0, 3.0), + ), +) +def test_activation_scale_buffer_uses_decaying_maximum(observed_scale, expected_scale): + name = "fc1_input_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated" + quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + initial_buffer = torch.tensor([4.0]) + quantizer._calibration_state = {"scale_inv": initial_buffer} + quantizer.calibrate( + _make_test_quantized_storage(_scale_inv=torch.tensor([observed_scale])), + calibration_decay=0.5, + ) + buffers = _common._get_calibration_metadata_buffers("fc1_input", quantizer) + value = buffers[name] + + torch.testing.assert_close(value, torch.tensor([expected_scale])) + assert value is initial_buffer + assert value is quantizer._calibration_state["scale_inv"] + + +def test_zero_decay_keeps_observed_metadata_reference(): + observed_scale = torch.tensor([2.0]) + quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + + quantizer.calibrate( + _make_test_quantized_storage(_scale_inv=observed_scale), + calibration_decay=0.0, + ) + + calibrated_scale = quantizer._calibration_state["scale_inv"] + assert calibrated_scale.data_ptr() == observed_scale.data_ptr() + + +def test_decaying_calibration_rejects_metadata_shape_change(): + quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + quantizer._calibration_state = {"scale_inv": torch.ones(1)} + + with pytest.raises(RuntimeError, match="calibration value shape changed"): + quantizer.calibrate( + _make_test_quantized_storage(_scale_inv=torch.ones(2)), + calibration_decay=0.5, + ) + + +@pytest.mark.parametrize("transformer_engine_calibration_decay", (0.0, 0.5)) +@pytest.mark.parametrize("initial_scale", (None, 4.0)) +def test_nan_activation_scale_does_not_update_buffer( + transformer_engine_calibration_decay, initial_scale +): + quantizer = _make_test_quantizer(Float8CurrentScalingQuantizer) + if initial_scale is not None: + quantizer._calibration_state = {"scale_inv": torch.tensor([initial_scale])} + + quantizer.calibrate( + _make_test_quantized_storage(_scale_inv=torch.tensor([float("nan")])), + calibration_decay=transformer_engine_calibration_decay, + ) + result = _common._get_calibration_metadata_buffers("fc1_input", quantizer) + + if initial_scale is None: + assert not result + assert not quantizer._calibration_state + else: + value = result["fc1_input_tensor_scale_inv_fp8_current_scaling_te_ptq_calibrated"] + torch.testing.assert_close(value, torch.tensor([initial_scale])) + + +def test_current_scaling_calibrates_from_high_precision_tensor(): + quantizer = Float8CurrentScalingQuantizer( + fp8_dtype=DType.kFloat8E4M3, + device=torch.device("cpu"), + ) + + quantizer.calibrate(torch.tensor([-112.0, 224.0])) + value = quantizer._calibration_state["scale_inv"] + + torch.testing.assert_close(value, torch.tensor([0.5])) + + +@pytest.mark.parametrize("input_value", (0.0, float("inf"))) +@pytest.mark.parametrize("force_pow_2_scales", (False, True)) +def test_current_scaling_calibration_handles_non_finite_scale(input_value, force_pow_2_scales): + quantizer = Float8CurrentScalingQuantizer( + fp8_dtype=DType.kFloat8E4M3, + device=torch.device("cpu"), + force_pow_2_scales=force_pow_2_scales, + ) + + quantizer.calibrate(torch.tensor([input_value])) + + torch.testing.assert_close(quantizer._calibration_state["scale_inv"], torch.ones(1)) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +@pytest.mark.parametrize("fp8_dtype", (DType.kFloat8E4M3, DType.kFloat8E5M2)) +def test_delayed_scaling_high_precision_calibration_matches_quantization(fp8_dtype): + torch.manual_seed(123) + tensor = torch.randn((32, 32), dtype=torch.bfloat16, device="cuda") + + active_quantizer = Float8Quantizer( + scale=torch.ones(1, dtype=torch.float32, device="cuda"), + amax=torch.zeros(1, dtype=torch.float32, device="cuda"), + fp8_dtype=fp8_dtype, + rowwise=True, + columnwise=False, + ) + quantized_tensor = active_quantizer(tensor) + active_quantizer.calibrate(quantized_tensor) + + calibration_quantizer = Float8Quantizer( + scale=torch.ones(1, dtype=torch.float32, device="cuda"), + amax=torch.zeros(1, dtype=torch.float32, device="cuda"), + fp8_dtype=fp8_dtype, + rowwise=True, + columnwise=False, + ) + calibration_quantizer.calibrate(tensor) + assert ( + calibration_quantizer.copy()._calibration_state is calibration_quantizer._calibration_state + ) + + torch.testing.assert_close( + active_quantizer._calibration_state["amax"], + active_quantizer.amax, + atol=0.0, + rtol=0.0, + ) + torch.testing.assert_close( + calibration_quantizer._calibration_state["amax"], + active_quantizer.amax, + atol=0.0, + rtol=0.0, + ) + + +@pytest.mark.skipif(not fp8_available, reason=reason_for_no_fp8) +@pytest.mark.parametrize("fp8_dtype", (DType.kFloat8E4M3, DType.kFloat8E5M2)) +@pytest.mark.parametrize("force_pow_2_scales", (False, True)) +@pytest.mark.parametrize("amax_epsilon", (0.0, 4.0)) +def test_current_scaling_high_precision_calibration_matches_quantization( + fp8_dtype, force_pow_2_scales, amax_epsilon +): + torch.manual_seed(123) + tensor = torch.randn((32, 32), dtype=torch.bfloat16, device="cuda") + quantizer_kwargs = { + "fp8_dtype": fp8_dtype, + "device": torch.device("cuda"), + "rowwise": True, + "columnwise": False, + "force_pow_2_scales": force_pow_2_scales, + "amax_epsilon": amax_epsilon, + } + + active_quantizer = Float8CurrentScalingQuantizer(**quantizer_kwargs) + quantized_tensor = active_quantizer(tensor) + active_quantizer.calibrate(quantized_tensor) + + calibration_quantizer = Float8CurrentScalingQuantizer(**quantizer_kwargs) + calibration_quantizer.calibrate(tensor) + + torch.testing.assert_close( + active_quantizer._calibration_state["scale_inv"], + quantized_tensor._scale_inv, + atol=0.0, + rtol=0.0, + ) + torch.testing.assert_close( + calibration_quantizer._calibration_state["scale_inv"], + quantized_tensor._scale_inv, + atol=0.0, + rtol=0.0, + ) + + +@pytest.mark.skipif(not nvfp4_available, reason=reason_for_no_nvfp4) +@pytest.mark.parametrize( + ("row_scaled_nvfp4", "with_rht", "with_post_rht_amax", "with_random_sign_mask"), + ( + (False, False, False, False), + (False, True, True, False), + (False, True, True, True), + (True, False, False, False), + ), +) +def test_nvfp4_high_precision_calibration_matches_quantization( + row_scaled_nvfp4, with_rht, with_post_rht_amax, with_random_sign_mask +): + torch.manual_seed(123) + tensor = torch.randn((32, 32), dtype=torch.bfloat16, device="cuda") + quantizer_kwargs = { + "rowwise": True, + "columnwise": not row_scaled_nvfp4, + "with_rht": with_rht, + "with_post_rht_amax": with_post_rht_amax, + "row_scaled_nvfp4": row_scaled_nvfp4, + "with_random_sign_mask": with_random_sign_mask, + } + + active_quantizer = NVFP4Quantizer(**quantizer_kwargs) + quantized_tensor = active_quantizer(tensor) + active_quantizer.calibrate(quantized_tensor) + calibration_quantizer = NVFP4Quantizer(**quantizer_kwargs) + calibration_quantizer.calibrate(tensor) + assert ( + calibration_quantizer.copy()._calibration_state is calibration_quantizer._calibration_state + ) + + expected_metadata_name = "amax_rowwise" if row_scaled_nvfp4 else "amax" + active_amax = active_quantizer._calibration_state[expected_metadata_name] + calibrated_amax = calibration_quantizer._calibration_state[expected_metadata_name] + torch.testing.assert_close( + active_amax, + quantized_tensor._amax_rowwise, + atol=0.0, + rtol=0.0, + ) + torch.testing.assert_close( + calibrated_amax, + quantized_tensor._amax_rowwise, + atol=0.0, + rtol=0.0, + ) + + +def test_shallow_quantizer_copy_shares_calibration_state(): + quantizer = Float8CurrentScalingQuantizer( + fp8_dtype=DType.kFloat8E4M3, + device=torch.device("cpu"), + ) + copied_quantizer = quantizer.copy() + copied_quantizer.calibrate(torch.tensor([-112.0, 224.0])) + value = copied_quantizer._calibration_state["scale_inv"] + + assert quantizer._calibration_state["scale_inv"] is value diff --git a/tests/pytorch/test_torch_compile.py b/tests/pytorch/test_torch_compile.py index e5d7169da1..071b47b399 100644 --- a/tests/pytorch/test_torch_compile.py +++ b/tests/pytorch/test_torch_compile.py @@ -1442,6 +1442,7 @@ def test_quantizer_value_object(factory): rebuilt = eval(repr_str, dict(globals_)) # pylint: disable=eval-used assert rebuilt == a and rebuilt is not a assert hash(rebuilt) == hash(a) + assert rebuilt._calibration_state == {} # The deprecated amax-reduction group is never part of the value. assert getattr(rebuilt, "amax_reduction_group", None) is None @@ -2022,6 +2023,7 @@ def fn(inp): "fuse_wgrad_accumulation", "delayed_wgrad", "quantized_input", + "scale_buffering", ] @@ -2055,6 +2057,17 @@ def fn(inp): return model(inp) return model, fn, "no_grad", None, "a quantized input tensor" + if case == "scale_buffering": + fp8_recipe = recipe.Float8CurrentScaling() + + def fn(inp): + with te.autocast( + recipe=fp8_recipe, + calibration_config=te.QuantizationCalibrationConfig(), + ): + return model(inp) + + return model, fn, "bwd", None, "Transformer Engine calibration metadata buffering" raise ValueError(case) @@ -2113,6 +2126,21 @@ def make_inp(): model.weight.grad, model_ref.weight.grad, atol=_EAGER_ATOL, rtol=_EAGER_RTOL ) torch.testing.assert_close(out.detach(), out_ref.detach(), atol=_EAGER_ATOL, rtol=_EAGER_RTOL) + if case == "scale_buffering": + calibration_buffers_ref = { + name: value + for name, value in model_ref.named_buffers() + if name.endswith("_te_ptq_calibrated") + } + calibration_buffers = { + name: value + for name, value in model.named_buffers() + if name.endswith("_te_ptq_calibrated") + } + assert calibration_buffers + assert calibration_buffers.keys() == calibration_buffers_ref.keys() + for name, value in calibration_buffers.items(): + torch.testing.assert_close(value, calibration_buffers_ref[name]) torch._dynamo.reset() compiled_fg = torch.compile(fn, fullgraph=True) diff --git a/transformer_engine/debug/features/log_fp8_tensor_stats.py b/transformer_engine/debug/features/log_fp8_tensor_stats.py index 96f1b644cf..e05a90f0d8 100644 --- a/transformer_engine/debug/features/log_fp8_tensor_stats.py +++ b/transformer_engine/debug/features/log_fp8_tensor_stats.py @@ -23,35 +23,12 @@ ) from transformer_engine.pytorch.tensor.mxfp8_tensor import MXFP8Quantizer from transformer_engine.pytorch.tensor.float8_blockwise_tensor import Float8BlockQuantizer - -try: - from transformer_engine.pytorch.tensor.nvfp4_tensor import NVFP4Quantizer - - _nvfp4_available = True -except ImportError: - _nvfp4_available = False - NVFP4Quantizer = None +from transformer_engine.pytorch.tensor.utils import get_quantization_recipe_name ALL_RECIPE_NAMES = ["fp8_delayed_scaling", "fp8_current_scaling", "mxfp8", "fp8_block_scaling"] -def _get_recipe_name(quantizer: Optional[Quantizer]): - if quantizer is None: - return "" - if isinstance(quantizer, Float8Quantizer): - return "fp8_delayed_scaling" - if isinstance(quantizer, Float8CurrentScalingQuantizer): - return "fp8_current_scaling" - if isinstance(quantizer, MXFP8Quantizer): - return "mxfp8" - if isinstance(quantizer, Float8BlockQuantizer): - return "fp8_block_scaling" - if _nvfp4_available and isinstance(quantizer, NVFP4Quantizer): - return "nvfp4" - raise ValueError(f"Unsupported quantizer type: {type(quantizer)}") - - def _get_new_quantizer(recipe_name, fp8_dtype): if recipe_name == "fp8_block_scaling": return Float8BlockQuantizer(fp8_dtype=fp8_dtype, rowwise=True, columnwise=True) @@ -336,7 +313,9 @@ def inspect_tensor( ) return - recipe_name = _get_recipe_name(quantizer) + recipe_name = get_quantization_recipe_name(quantizer) + if recipe_name == "nvfp4_rowwise": + recipe_name = "nvfp4" for stat in config["stats"]: self.check_if_stat_is_supported( diff --git a/transformer_engine/debug/pytorch/debug_quantization.py b/transformer_engine/debug/pytorch/debug_quantization.py index 107dd7a373..552209ad9a 100644 --- a/transformer_engine/debug/pytorch/debug_quantization.py +++ b/transformer_engine/debug/pytorch/debug_quantization.py @@ -429,7 +429,7 @@ def any_feature_enabled(self) -> bool: return True return False - def calibrate(self, tensor: torch.Tensor): + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0): """Calibration override, should not be invoked.""" raise RuntimeError("[NVTORCH-INSPECT ERROR] Calibration with debug is not supported") diff --git a/transformer_engine/pytorch/__init__.py b/transformer_engine/pytorch/__init__.py index 576fb57c5e..11ef1bf5a9 100644 --- a/transformer_engine/pytorch/__init__.py +++ b/transformer_engine/pytorch/__init__.py @@ -55,6 +55,8 @@ from transformer_engine.pytorch.quantization import QuantizerRole from transformer_engine.pytorch.quantization import QuantizerRequest from transformer_engine.pytorch.quantization import DelayedScalingRequest +from transformer_engine.pytorch.quantization import TEAutocastState +from transformer_engine.pytorch.quantization import QuantizationCalibrationConfig from transformer_engine.pytorch.utils import get_cudnn_version from transformer_engine.pytorch.utils import get_device_compute_capability from transformer_engine.pytorch.utils import is_bf16_available diff --git a/transformer_engine/pytorch/attention/dot_product_attention/dot_product_attention.py b/transformer_engine/pytorch/attention/dot_product_attention/dot_product_attention.py index 658dab5d88..ba865543a7 100644 --- a/transformer_engine/pytorch/attention/dot_product_attention/dot_product_attention.py +++ b/transformer_engine/pytorch/attention/dot_product_attention/dot_product_attention.py @@ -1121,7 +1121,10 @@ def init_fp8_metadata(self, num_gemms: int = 1) -> None: # With quantization off the base class does all that is needed. qstate = FP8GlobalStateManager.quantization_state if torch.compiler.is_compiling() and not ( - qstate.fp8_enabled or qstate.fp8_calibration or qstate.fp8_parameters + qstate.fp8_enabled + or qstate.fp8_calibration + or qstate.calibration_config is not None + or qstate.fp8_parameters ): super().init_fp8_metadata(num_gemms=num_gemms) return diff --git a/transformer_engine/pytorch/attention/fused_mla_q_uproj.py b/transformer_engine/pytorch/attention/fused_mla_q_uproj.py index b15e53e6ee..f45a40ecb9 100644 --- a/transformer_engine/pytorch/attention/fused_mla_q_uproj.py +++ b/transformer_engine/pytorch/attention/fused_mla_q_uproj.py @@ -203,6 +203,7 @@ def run( from cuda.bindings import driver as cuda + # pylint: disable-next=c-extension-no-member stream = cuda.CUstream(torch.cuda.current_stream(x.device).cuda_stream) wrapper = cls._kernel() diff --git a/transformer_engine/pytorch/dynamo/quantizer_opaque.py b/transformer_engine/pytorch/dynamo/quantizer_opaque.py index da595b1070..10a1cc0cc6 100644 --- a/transformer_engine/pytorch/dynamo/quantizer_opaque.py +++ b/transformer_engine/pytorch/dynamo/quantizer_opaque.py @@ -45,6 +45,9 @@ def _rebuild_quantizer(cls: type, items: Tuple[Tuple[str, Any], ...]) -> Any: if name == "dtype": value = DType.cast(value) object.__setattr__(obj, name, value) + # Runtime calibration metadata is intentionally excluded from value semantics, + # but rebuilt quantizers still need an initialized state for copy()/quantize(). + object.__setattr__(obj, "_calibration_state", {}) # Restore non-value derived state that ``__init__`` would normally build but # that cannot live in the value key (e.g. NVFP4's ``rht_matrix`` tensor). finalize = getattr(obj, "_rebuild_derived_state", None) diff --git a/transformer_engine/pytorch/graph.py b/transformer_engine/pytorch/graph.py index 04fa56721d..bad73a02e6 100644 --- a/transformer_engine/pytorch/graph.py +++ b/transformer_engine/pytorch/graph.py @@ -20,6 +20,7 @@ from .quantization import ( autocast, FP8GlobalStateManager, + QuantizationCalibrationConfig, get_default_fp8_recipe, ) from .distributed import get_all_rng_states, graph_safe_rng_available @@ -1445,6 +1446,7 @@ def make_graphed_callables( pre_warmup_hook: Optional[Callable] = None, post_warmup_hook: Optional[Callable] = None, capture_time_hooks: Optional[List[Optional[Dict[str, Dict]]]] = None, + calibration_config: Optional[QuantizationCalibrationConfig] = None, ) -> Union[Callable, Tuple[Callable, ...]]: """ Make CUDA graph version of Transformer Engine modules @@ -1518,10 +1520,10 @@ def make_graphed_callables( whether or not to enable low precision quantization (FP8/FP4). If tuple, the length must match the number of modules. calibrating: bool, default = False - calibration mode allows collecting statistics such as amax and scale - data of quantized tensors even when executing without quantization enabled. - This is useful for saving an inference ready checkpoint while training - using a higher precision. + Enables calibration with the default configuration. + calibration_config: QuantizationCalibrationConfig, default = None + Custom configuration for collecting checkpointable quantization scaling + factors. Providing a config also enables calibration. recipe: recipe.Recipe, default = None recipe used for low precision quantization. amax_reduction_group: torch._C._distributed_c10d.ProcessGroup, default = None @@ -1668,6 +1670,7 @@ def call_func(self, *args, **kwargs): recipe=recipe, amax_reduction_group=amax_reduction_group, _graph=True, + calibration_config=calibration_config, ): outputs = old_call_funcs[block_cls](self, *args, **kwargs) return outputs diff --git a/transformer_engine/pytorch/module/_common.py b/transformer_engine/pytorch/module/_common.py index 908e88f30d..c870546941 100644 --- a/transformer_engine/pytorch/module/_common.py +++ b/transformer_engine/pytorch/module/_common.py @@ -5,8 +5,10 @@ """Internal function used by multiple modules.""" import dataclasses +import functools +import inspect import queue -from typing import Any, Callable, List, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Optional, Tuple, Union import torch @@ -17,6 +19,49 @@ from ..utils import get_default_init_method +@functools.lru_cache(maxsize=None) +def _supports_calibration_decay(quantizer_type: type) -> bool: + """Whether a quantizer's calibrate override accepts calibration_decay.""" + try: + parameters = inspect.signature(quantizer_type.calibrate).parameters + except (TypeError, ValueError): + return False + return "calibration_decay" in parameters or any( + parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in parameters.values() + ) + + +def _is_in_activation_recompute_phase() -> bool: + """Whether a forward is running during activation recomputation.""" + from ..distributed import in_fp8_activation_recompute_phase + + if in_fp8_activation_recompute_phase(): + return True + # Special hidden PyTorch AutoGrad identifier for activation recompute. + current_graph_task_id = getattr(torch._C, "_current_graph_task_id", None) + return current_graph_task_id is not None and current_graph_task_id() != -1 + + +def _resolve_calibration_quantizer(tensor: Any, quantizer: Any) -> Any: + """Get the quantizer that owns calibration state for a tensor.""" + quantizer = getattr(tensor, "_quantizer", None) or quantizer + return getattr(quantizer, "parent_quantizer", quantizer) + + +def _get_calibration_metadata_buffers(tensor_name: str, quantizer: Any) -> Dict[str, torch.Tensor]: + """Get checkpoint-buffer aliases from quantizer calibration state.""" + if quantizer is None: + return {} + recipe = quantizer.get_quantization_recipe_name() + if not recipe: + return {} + return { + # Standard naming for PTQ calibration data. + f"{tensor_name}_tensor_{metadata_name}_{recipe}_te_ptq_calibrated": value + for metadata_name, value in quantizer._calibration_state.items() + } + + def set_quantizer_amax_reduction_group(quantizer, amax_reduction_group) -> None: """Set the amax reduction group on a quantizer; no-op if it doesn't support it. diff --git a/transformer_engine/pytorch/module/grouped_linear.py b/transformer_engine/pytorch/module/grouped_linear.py index 8b078944f1..16760af961 100644 --- a/transformer_engine/pytorch/module/grouped_linear.py +++ b/transformer_engine/pytorch/module/grouped_linear.py @@ -4,7 +4,7 @@ """GroupedLinear API""" -from typing import Union, Optional, Callable, Tuple, List +from typing import Union, Optional, Callable, Tuple, List, Dict from itertools import chain import os import warnings @@ -31,7 +31,14 @@ _clear_high_precision_init_val, _get_high_precision_init_val, ) -from ._common import can_reconstruct_wgrad_input_from_original, WeightGradStore +from ._common import ( + _get_calibration_metadata_buffers, + _is_in_activation_recompute_phase, + _resolve_calibration_quantizer, + _supports_calibration_decay, + can_reconstruct_wgrad_input_from_original, + WeightGradStore, +) from . import _split_quantization from ..quantization import FP8GlobalStateManager, QuantizerRole from ..utils import ( @@ -152,6 +159,50 @@ def is_module_grouped_tensor_path_supported( return False +def _update_grouped_calibration_metadata_buffers( + calibration_buffers: Dict[str, Optional[torch.Tensor]], + input_tensors: List[Union[torch.Tensor, QuantizedTensorStorage]], + weight_tensors: List[Union[torch.Tensor, QuantizedTensorStorage]], + input_quantizers: List[Optional[Quantizer]], + weight_quantizers: List[Optional[Quantizer]], +) -> None: + """Update GroupedLinear PTQ buffers from per-GEMM calibration state.""" + for index, tensor in enumerate(input_tensors): + quantizer = _resolve_calibration_quantizer(tensor, input_quantizers[index]) + calibration_buffers.update( + _get_calibration_metadata_buffers(f"input_gemm{index}", quantizer) + ) + for index, tensor in enumerate(weight_tensors): + quantizer = _resolve_calibration_quantizer(tensor, weight_quantizers[index]) + calibration_buffers.update( + _get_calibration_metadata_buffers(f"weight_gemm{index}", quantizer) + ) + + +def _calibrate_grouped_tensors( + input_tensors: List[Union[torch.Tensor, QuantizedTensorStorage]], + weight_tensors: List[Union[torch.Tensor, QuantizedTensorStorage]], + input_quantizers: List[Optional[Quantizer]], + weight_quantizers: List[Optional[Quantizer]], + transformer_engine_calibration_decay: float, +) -> None: + """Calibrate GroupedLinear input and weight quantizers once per GEMM.""" + for tensor, quantizer in zip(input_tensors, input_quantizers): + quantizer = _resolve_calibration_quantizer(tensor, quantizer) + if quantizer is not None: + if _supports_calibration_decay(type(quantizer)): + quantizer.calibrate( + tensor, + calibration_decay=transformer_engine_calibration_decay, + ) + else: + quantizer.calibrate(tensor) + for tensor, quantizer in zip(weight_tensors, weight_quantizers): + quantizer = _resolve_calibration_quantizer(tensor, quantizer) + if quantizer is not None: + quantizer.calibrate(tensor) + + class _GroupedLinear(torch.autograd.Function): """GroupedLinear semi-top level module Calls custom cuda extensions. @@ -428,6 +479,8 @@ def _forward_grouped_tensor( save_original_input: bool, single_grouped_weight: bool, single_grouped_bias: bool, + calibration_buffers: Optional[Dict[str, Optional[torch.Tensor]]], + transformer_engine_calibration_decay: float, weights: Tuple[torch.Tensor, ...], biases: Tuple[torch.Tensor, ...], out: Optional[torch.Tensor] = None, @@ -536,6 +589,31 @@ def _forward_grouped_tensor( use_split_accumulator=use_split_accumulator, ) + if calibration_buffers is not None and not _is_in_activation_recompute_phase(): + grouped_inputs = grouped_x.quantized_tensors + if grouped_inputs is None: + grouped_inputs = grouped_x.split_into_quantized_tensors() + if isinstance(weights_for_gemm, GroupedTensorStorage): + grouped_weights = weights_for_gemm.quantized_tensors + if grouped_weights is None: + grouped_weights = weights_for_gemm.split_into_quantized_tensors() + else: + grouped_weights = weights_for_gemm + _calibrate_grouped_tensors( + grouped_inputs, + grouped_weights, + input_quantizers, + weight_quantizers, + transformer_engine_calibration_decay, + ) + _update_grouped_calibration_metadata_buffers( + calibration_buffers, + grouped_inputs, + grouped_weights, + input_quantizers, + weight_quantizers, + ) + if is_grad_enabled: input_to_save = grouped_x if weight_requires_grad: @@ -657,6 +735,8 @@ def forward( single_grouped_weight, single_grouped_bias, use_grouped_tensor, + calibration_buffers, + transformer_engine_calibration_decay, ) = non_tensor_args recipe = FP8GlobalStateManager.get_fp8_recipe() if fp8 else None backward_override = recipe.backward_override if recipe is not None else None @@ -809,6 +889,8 @@ def forward( save_original_input=save_original_input, single_grouped_weight=single_grouped_weight, single_grouped_bias=single_grouped_bias, + calibration_buffers=calibration_buffers, + transformer_engine_calibration_decay=transformer_engine_calibration_decay, weights=weights, biases=biases, out=out, @@ -855,6 +937,23 @@ def forward( else: weights_fp8 = [cast_if_needed(weight, activation_dtype) for weight in weights] + # Calibrate quantizers and buffer their metadata when requested. + if calibration_buffers is not None and not _is_in_activation_recompute_phase(): + _calibrate_grouped_tensors( + inputmats, + weights_fp8, + input_quantizers, + weight_quantizers, + transformer_engine_calibration_decay, + ) + _update_grouped_calibration_metadata_buffers( + calibration_buffers, + inputmats, + weights_fp8, + input_quantizers, + weight_quantizers, + ) + # Initialize biases bias_dtype = activation_dtype if fp8 and activation_dtype == torch.float32: @@ -890,11 +989,6 @@ def forward( use_split_accumulator=use_split_accumulator, ) - if fp8_calibration: - for i in range(num_gemms): - input_quantizers[i].calibrate(inputmats[i]) - weight_quantizers[i].calibrate(weights[i]) - if cpu_offloading: mark_not_offload(*weights_fp8, *weights) @@ -2272,6 +2366,14 @@ def forward( if cache_weight else [None] * num_gemms ) + calibration_config = FP8GlobalStateManager.get_calibration_config() + calibration_buffers = None + if calibration_config is not None: + calibration_buffers = { + name: value + for name, value in self._buffers.items() + if name.endswith("_te_ptq_calibrated") + } non_tensor_args = ( self.apply_bias, @@ -2298,6 +2400,12 @@ def forward( self.single_grouped_weight, use_grouped_bias, self.use_grouped_tensor, + calibration_buffers, + ( + calibration_config.transformer_engine_calibration_decay + if calibration_config is not None + else 0.0 + ), ) out, new_workspaces = linear_fn( *autograd_ctx, @@ -2310,6 +2418,16 @@ def forward( *bias_tensors, ) + if calibration_buffers is not None: + # Assign scaling-factor calibration buffers to the model. + # Materializing a new buffer requires a CUDA graph warmup step. + for name, value in calibration_buffers.items(): + if value is not None: + if name in self._buffers: + setattr(self, name, value) + else: + self.register_buffer(name, value, persistent=False) + if cache_weight: for i, ws in enumerate(new_workspaces): if ws is not None: @@ -2474,7 +2592,8 @@ def _get_quantizers(self): [None] * self.num_gemms, [None] * self.num_gemms, ) - if self.fp8: + # Calibration-only mode needs quantizers to observe non-quantized tensors. + if self.fp8 or self.fp8_calibration: input_quantizers = [ self.quantizers["scaling_fwd"][ self._offsets["input"] + i * self._num_fp8_tensors_per_gemm["fwd"] diff --git a/transformer_engine/pytorch/module/layernorm_linear.py b/transformer_engine/pytorch/module/layernorm_linear.py index 561e813348..f0d85f7d32 100644 --- a/transformer_engine/pytorch/module/layernorm_linear.py +++ b/transformer_engine/pytorch/module/layernorm_linear.py @@ -17,7 +17,10 @@ from transformer_engine.common.recipe import Recipe from transformer_engine.pytorch.torch_version import torch_version -from transformer_engine.pytorch.tensor.utils import clear_columnwise_cache, is_custom +from transformer_engine.pytorch.tensor.utils import ( + clear_columnwise_cache, + is_custom, +) from .base import ( fill_userbuffers_buffer_for_all_gather, get_ub, @@ -66,6 +69,10 @@ from ..jit import no_torch_dynamo from ..graph import is_graph_capturing from ._common import ( + _get_calibration_metadata_buffers, + _is_in_activation_recompute_phase, + _resolve_calibration_quantizer, + _supports_calibration_decay, apply_normalization, noop_cat, set_quantizer_amax_reduction_group, @@ -124,7 +131,7 @@ def forward( eps, is_first_microbatch, fp8, - fp8_calibration, + _fp8_calibration, wgrad_store, fuse_wgrad_accumulation, input_quantizer, @@ -160,6 +167,8 @@ def forward( symmetric_ar_type, debug, is_fsdp2, + transformer_engine_calibration_decay, + calibration_buffers, ) = non_tensor_args if fp8: backward_override = FP8GlobalStateManager.get_fp8_recipe().backward_override @@ -376,12 +385,27 @@ def forward( bias_dtype = torch.bfloat16 bias = cast_if_needed(bias, bias_dtype) if bias is not None else bias - # Calibrate quantizers if needed - if not fp8 and fp8_calibration: - if input_quantizer is not None: - input_quantizer.calibrate(ln_out_total) - if weight_quantizer is not None: - weight_quantizer.calibrate(weight) + input_calibration_quantizer = _resolve_calibration_quantizer(ln_out_total, input_quantizer) + weight_calibration_quantizer = _resolve_calibration_quantizer(weightmat, weight_quantizer) + + # Calibrate quantizers and buffer their metadata when requested. + if calibration_buffers is not None and not _is_in_activation_recompute_phase(): + if input_calibration_quantizer is not None: + if _supports_calibration_decay(type(input_calibration_quantizer)): + input_calibration_quantizer.calibrate( + ln_out_total, + calibration_decay=transformer_engine_calibration_decay, + ) + else: + input_calibration_quantizer.calibrate(ln_out_total) + if weight_calibration_quantizer is not None: + weight_calibration_quantizer.calibrate(weightmat) + calibration_buffers.update( + _get_calibration_metadata_buffers("input", input_calibration_quantizer) + ) + calibration_buffers.update( + _get_calibration_metadata_buffers("weight", weight_calibration_quantizer) + ) # Choose whether to use GEMM kernel with split accumulator use_split_accumulator = _2X_ACC_FPROP @@ -1765,6 +1789,15 @@ def forward( self._fp8_workspaces.get(cache_name) if cache_name is not None else None ) + calibration_config = FP8GlobalStateManager.get_calibration_config() + calibration_buffers = None + if calibration_config is not None: + calibration_buffers = { + name: value + for name, value in self._buffers.items() + if name.endswith("_te_ptq_calibrated") + } + non_tensor_args = ( self.eps, is_first_microbatch, @@ -1805,6 +1838,12 @@ def forward( self.symmetric_ar_type, debug, self.is_fsdp2, + ( + calibration_config.transformer_engine_calibration_decay + if calibration_config is not None + else 0.0 + ), + calibration_buffers, ) out, ln_out, new_weight_workspace = fwd_fn( *autograd_ctx, @@ -1817,6 +1856,14 @@ def forward( non_tensor_args, ) + if calibration_buffers is not None: + for name, value in calibration_buffers.items(): + if value is not None: + if name in self._buffers: + setattr(self, name, value) + else: + self.register_buffer(name, value, persistent=False) + if new_weight_workspace is not None and cache_name is not None: if isinstance(new_weight_workspace, torch.Tensor): new_weight_workspace = new_weight_workspace.detach() @@ -1837,10 +1884,13 @@ def forward( return out def _get_quantizers(self, fp8_output, fp8_grad, is_grad_enabled): - if not self.fp8: + if not self.fp8 and not self.fp8_calibration: + # Calibration-only mode needs quantizers to observe non-quantized tensors, + # so return None only when neither FP8 nor calibration is active. return [None] * 6 - self._warn_missing_output_quantizer_role(fp8_output, fp8_grad) + if self.fp8: + self._warn_missing_output_quantizer_role(fp8_output, fp8_grad) grad_input_quantizer = None grad_weight_quantizer = None diff --git a/transformer_engine/pytorch/module/layernorm_mlp.py b/transformer_engine/pytorch/module/layernorm_mlp.py index 3ee0cda50c..a57398930d 100644 --- a/transformer_engine/pytorch/module/layernorm_mlp.py +++ b/transformer_engine/pytorch/module/layernorm_mlp.py @@ -18,7 +18,10 @@ from transformer_engine.common.recipe import Recipe from transformer_engine.pytorch.torch_version import torch_version -from transformer_engine.pytorch.tensor.utils import clear_columnwise_cache, is_custom +from transformer_engine.pytorch.tensor.utils import ( + clear_columnwise_cache, + is_custom, +) from .base import ( fill_userbuffers_buffer_for_all_gather, _ub_communicators, @@ -73,6 +76,10 @@ from ..tensor.hybrid_tensor import HybridQuantizer from ..tensor.identity_tensor import IdentityQuantizer from ._common import ( + _get_calibration_metadata_buffers, + _is_in_activation_recompute_phase, + _resolve_calibration_quantizer, + _supports_calibration_decay, apply_normalization, set_quantizer_amax_reduction_group, set_quantizer_usage_for_wgrad_all_gather, @@ -244,6 +251,8 @@ def _forward( checkpoint, debug, is_fsdp2, + transformer_engine_calibration_decay, + calibration_buffers, recompute_for_bwd, ) = non_tensor_args if fp8: @@ -346,6 +355,8 @@ def _forward( "checkpoint": checkpoint, "debug": debug, "is_fsdp2": is_fsdp2, + "transformer_engine_calibration_decay": transformer_engine_calibration_decay, + "calibration_buffers": calibration_buffers, "recompute_for_bwd": True, # set this to true for recomputation phase } # Make sure input dimensions are compatible @@ -557,12 +568,39 @@ def _forward( if fc2_bias is not None: fc2_bias = cast_if_needed(fc2_bias, bias_dtype) - # Calibrate quantizers if needed - if not fp8 and fp8_calibration: - if fc1_input_quantizer is not None: - fc1_input_quantizer.calibrate(ln_out_total) - if fc1_weight_quantizer is not None: - fc1_weight_quantizer.calibrate(fc1_weight) + fc1_input_calibration_quantizer = _resolve_calibration_quantizer( + ln_out_total, fc1_input_quantizer + ) + fc1_weight_calibration_quantizer = _resolve_calibration_quantizer( + fc1_weight_final, fc1_weight_quantizer + ) + + fc1_input_calibration_buffers = {} + fc1_weight_calibration_buffers = {} + + # Calibrate FC1 quantizers and collect their metadata when requested. + should_calibrate = ( + calibration_buffers is not None + and not _is_in_activation_recompute_phase() + and not is_recomputation + ) + if should_calibrate: + if fc1_input_calibration_quantizer is not None: + if _supports_calibration_decay(type(fc1_input_calibration_quantizer)): + fc1_input_calibration_quantizer.calibrate( + ln_out_total, + calibration_decay=transformer_engine_calibration_decay, + ) + else: + fc1_input_calibration_quantizer.calibrate(ln_out_total) + if fc1_weight_calibration_quantizer is not None: + fc1_weight_calibration_quantizer.calibrate(fc1_weight_final) + fc1_input_calibration_buffers = _get_calibration_metadata_buffers( + "fc1_input", fc1_input_calibration_quantizer + ) + fc1_weight_calibration_buffers = _get_calibration_metadata_buffers( + "fc1_weight", fc1_weight_calibration_quantizer + ) # ------------------------------------------------------ # FC1 GEMM @@ -654,9 +692,17 @@ def _forward( else: act_out = activation_func(fc1_out, fc2_input_quantizer, **act_params) - if not fp8 and fp8_calibration: - if fc2_input_quantizer is not None: - fc2_input_quantizer.calibrate(act_out) + fc2_input_calibration_quantizer = _resolve_calibration_quantizer( + act_out, fc2_input_quantizer + ) + if should_calibrate and fc2_input_calibration_quantizer is not None: + if _supports_calibration_decay(type(fc2_input_calibration_quantizer)): + fc2_input_calibration_quantizer.calibrate( + act_out, + calibration_decay=transformer_engine_calibration_decay, + ) + else: + fc2_input_calibration_quantizer.calibrate(act_out) # we want to skip fc2 computation if we are checkpointing and recomputing, # otherwise we compute fc2 @@ -673,10 +719,22 @@ def _forward( ): # we can safely get rid of these if this is the case clear_tensor_data(fc1_out) - if not fp8 and fp8_calibration: - - if fc2_weight_quantizer is not None: - fc2_weight_quantizer.calibrate(fc2_weight) + fc2_weight_calibration_quantizer = _resolve_calibration_quantizer( + fc2_weight_final, fc2_weight_quantizer + ) + if should_calibrate: + if fc2_weight_calibration_quantizer is not None: + fc2_weight_calibration_quantizer.calibrate(fc2_weight_final) + calibration_buffers.update(fc1_input_calibration_buffers) + calibration_buffers.update(fc1_weight_calibration_buffers) + calibration_buffers.update( + _get_calibration_metadata_buffers("fc2_input", fc2_input_calibration_quantizer) + ) + calibration_buffers.update( + _get_calibration_metadata_buffers( + "fc2_weight", fc2_weight_calibration_quantizer + ) + ) # Configure Userbuffers reduce-scatter if needed ub_obj_fc2out = None @@ -2001,7 +2059,6 @@ def __init__( self.zero_centered_gamma = zero_centered_gamma self.symmetric_ar_type = symmetric_ar_type self.checkpoint = checkpoint - # GEMM-GELU fusion is currently only supported with split GEMM-AG overlap self.gemm_gelu_fusion = ( bool(int(os.getenv("NVTE_GEMM_GELU_FUSION", "0"))) @@ -2381,6 +2438,15 @@ def forward( self._fp8_workspaces.get(cache_name_fc2) if cache_name_fc2 is not None else None ) + calibration_config = FP8GlobalStateManager.get_calibration_config() + calibration_buffers = None + if calibration_config is not None: + calibration_buffers = { + name: value + for name, value in self._buffers.items() + if name.endswith("_te_ptq_calibrated") + } + non_tensor_args = ( self.eps, is_first_microbatch, @@ -2431,6 +2497,12 @@ def forward( self.checkpoint, debug, self.is_fsdp2, + ( + calibration_config.transformer_engine_calibration_decay + if calibration_config is not None + else 0.0 + ), + calibration_buffers, ) out, ln_out, new_fc1_ws, new_fc2_ws = fwd_fn( *autograd_ctx, @@ -2446,6 +2518,14 @@ def forward( non_tensor_args, ) + if calibration_buffers is not None: + for name, value in calibration_buffers.items(): + if value is not None: + if name in self._buffers: + setattr(self, name, value) + else: + self.register_buffer(name, value, persistent=False) + if new_fc1_ws is not None and cache_name_fc1 is not None: if isinstance(new_fc1_ws, torch.Tensor): new_fc1_ws = new_fc1_ws.detach() @@ -2486,6 +2566,7 @@ def _get_quantizers(self, fp8_output, is_grad_enabled): fc2_grad_output_quantizer, ) = [None] * 10 fc1_weight_quantizer, fc2_weight_quantizer = self._get_weight_quantizers() + # Calibration-only mode needs quantizers to observe non-quantized tensors. if self.fp8 or self.fp8_calibration: fc1_input_quantizer = self.quantizers["scaling_fwd"][FP8FwdTensorIdx.GEMM1_INPUT] fc1_input_quantizer.internal = True diff --git a/transformer_engine/pytorch/module/linear.py b/transformer_engine/pytorch/module/linear.py index 94de69e975..86d990a7ea 100644 --- a/transformer_engine/pytorch/module/linear.py +++ b/transformer_engine/pytorch/module/linear.py @@ -32,6 +32,10 @@ _2X_ACC_WGRAD, ) from ._common import ( + _get_calibration_metadata_buffers, + _is_in_activation_recompute_phase, + _resolve_calibration_quantizer, + _supports_calibration_decay, can_reconstruct_wgrad_input_from_original, noop_cat, set_quantizer_amax_reduction_group, @@ -93,7 +97,10 @@ ) from ..tensor.float8_tensor import Float8CurrentScalingQuantizer, Float8Quantizer from ..tensor.mxfp8_tensor import MXFP8Quantizer -from ..tensor.utils import clear_columnwise_cache, is_custom +from ..tensor.utils import ( + clear_columnwise_cache, + is_custom, +) from ..export import is_in_onnx_export_mode, assert_warmed_up from ..cpu_offload import ( is_cpu_offload_enabled, @@ -176,6 +183,10 @@ class LinearFwdArgs: fuse_wgrad_accumulation: bool wgrad_store: Optional[Any] + # Transformer Engine calibration metadata buffering + calibration_buffers: Optional[Dict[str, Optional[torch.Tensor]]] + transformer_engine_calibration_decay: float + # --- Misc --- cpu_offloading: bool is_grad_enabled: bool @@ -370,6 +381,7 @@ def _linear_forward_impl( ctx_attrs)``. ``new_weight_workspace`` is the freshly produced FP8 weight workspace (returned alongside ``out`` so the caller can refresh its cache). The last two are ``None`` when gradients are disabled. + Calibration metadata buffers are updated through ``args.calibration_buffers``. """ weight = args.weight @@ -598,12 +610,27 @@ def _linear_forward_impl( bias_dtype = torch.bfloat16 bias = cast_if_needed(bias, bias_dtype) if bias is not None else bias - # Calibrate quantizers if needed - if not fp8 and args.fp8_calibration: - if input_quantizer is not None: - input_quantizer.calibrate(inputmat_total) - if weight_quantizer is not None: - weight_quantizer.calibrate(weight) + input_calibration_quantizer = _resolve_calibration_quantizer(inputmat_total, input_quantizer) + weight_calibration_quantizer = _resolve_calibration_quantizer(weightmat, weight_quantizer) + + # Calibrate quantizers and buffer their metadata when requested. + if args.calibration_buffers is not None and not _is_in_activation_recompute_phase(): + if input_calibration_quantizer is not None: + if _supports_calibration_decay(type(input_calibration_quantizer)): + input_calibration_quantizer.calibrate( + inputmat_total, + calibration_decay=args.transformer_engine_calibration_decay, + ) + else: + input_calibration_quantizer.calibrate(inputmat_total) + if weight_calibration_quantizer is not None: + weight_calibration_quantizer.calibrate(weightmat) + args.calibration_buffers.update( + _get_calibration_metadata_buffers("input", input_calibration_quantizer) + ) + args.calibration_buffers.update( + _get_calibration_metadata_buffers("weight", weight_calibration_quantizer) + ) # Choose whether to use GEMM kernel with split accumulator use_split_accumulator = _2X_ACC_FPROP @@ -2475,7 +2502,14 @@ def forward( bias_tensor if (self.apply_bias and not self.gemm_bias_unfused_add) else None ) wgrad_store = self.wgrad_store if self.wgrad_store.delay_wgrad_compute() else None - + calibration_config = FP8GlobalStateManager.get_calibration_config() + calibration_buffers = None + if calibration_config is not None: + calibration_buffers = { + name: value + for name, value in self._buffers.items() + if name.endswith("_te_ptq_calibrated") + } fwd_args = LinearFwdArgs( # tensors weight=weight_tensor, @@ -2532,6 +2566,13 @@ def forward( # weight-grad scheduling fuse_wgrad_accumulation=self.fuse_wgrad_accumulation, wgrad_store=wgrad_store, + # Buffering TE calibration metadata, e.g. scaling factors. + calibration_buffers=calibration_buffers, + transformer_engine_calibration_decay=( + calibration_config.transformer_engine_calibration_decay + if calibration_config is not None + else 0.0 + ), # misc cpu_offloading=is_cpu_offload_enabled(), is_grad_enabled=is_grad_enabled, @@ -2555,6 +2596,16 @@ def forward( weight_tensor, inp, linear_bias_tensor, fwd_args, is_grad_enabled ) + if calibration_buffers is not None: + # Assign Transformer Engine calibration metadata buffers to the model. + # Requires CUDA graph warmup step. + for name, value in calibration_buffers.items(): + if value is not None: + if name in self._buffers: + setattr(self, name, value) + else: + self.register_buffer(name, value, persistent=False) + if new_weight_workspace is not None and cache_name is not None: if isinstance(new_weight_workspace, torch.Tensor): new_weight_workspace = new_weight_workspace.detach() @@ -2570,10 +2621,13 @@ def forward( return out def _get_quantizers(self, fp8_output, fp8_grad, is_grad_enabled): - if not self.fp8: + if not self.fp8 and not self.fp8_calibration: + # Calibration-only mode needs quantizers to observe non-quantized tensors, + # so return None only when neither FP8 nor calibration is active. return [None] * 6 - self._warn_missing_output_quantizer_role(fp8_output, fp8_grad) + if self.fp8: + self._warn_missing_output_quantizer_role(fp8_output, fp8_grad) grad_input_quantizer = None grad_weight_quantizer = None @@ -2643,6 +2697,8 @@ def _compile_eager_fallback_reason( prepare_forward. Quantizer checks stay in compile_unsupported_reason.""" if debug: return "debug instrumentation (nvidia-dlfw-inspect)" + if FP8GlobalStateManager.get_calibration_config() is not None: + return "Transformer Engine calibration metadata buffering" weight_tensor, bias_tensor = self._get_weight_and_bias_tensors() if is_distributed_weight(weight_tensor): return "a DistributedWeight (custom weight parallelism, e.g. GTP)" diff --git a/transformer_engine/pytorch/quantization.py b/transformer_engine/pytorch/quantization.py index 34ae20b498..580b6ada84 100644 --- a/transformer_engine/pytorch/quantization.py +++ b/transformer_engine/pytorch/quantization.py @@ -385,6 +385,37 @@ def is_nvfp4_available(return_reason: bool = False) -> Union[bool, Tuple[bool, s return check_nvfp4_support()[0] +@dataclass(frozen=True, slots=True) +class QuantizationCalibrationConfig: + """Configuration for collecting checkpointable quantization scaling factors. + + Parameters + ---------- + transformer_engine_calibration_decay : float, default = 0.0 + Decay applied to buffered activation scaling factors before incorporating + each new observation. With zero decay, only the latest value is retained. + """ + + transformer_engine_calibration_decay: float = 0.0 + + def __post_init__(self) -> None: + if self.transformer_engine_calibration_decay < 0.0: + raise ValueError("transformer_engine_calibration_decay must be non-negative") + + +@dataclass(frozen=True, slots=True) +class TEAutocastState: + """Snapshot of process-global quantization autocast state.""" + + fp8_enabled: bool + fp8_calibration: bool + calibration_config: Optional[QuantizationCalibrationConfig] + fp8_recipe: Optional[Recipe] + fp8_distributed_group: Optional[dist_group_type] + is_first_fp8_module: bool + fp8_graph_capturing: bool + + @dataclass(slots=True) class FP8GlobalState: """Mutable process-global FP8 state stored on an instance. @@ -410,6 +441,7 @@ class FP8GlobalState: default_factory=dict ) skip_fp8_weight_update_tensor: Optional[torch.Tensor] = None + calibration_config: Optional[QuantizationCalibrationConfig] = None class FP8GlobalStateManager: @@ -571,9 +603,22 @@ def is_fp8_enabled(cls) -> bool: @classmethod def is_fp8_calibration(cls) -> bool: - """Is FP8 calibration""" + """Whether quantization calibration is enabled.""" return cls.quantization_state.fp8_calibration + @classmethod + def get_calibration_config(cls) -> Optional[QuantizationCalibrationConfig]: + """Get the active quantization calibration configuration.""" + qstate = cls.quantization_state + if not qstate.fp8_calibration: + # User-declined calibration using the legacy fp8_calibration config. + return None + if qstate.calibration_config is not None: + # User-provided calibration config. + return qstate.calibration_config + # Default (fp8_calibration=True) config. + return QuantizationCalibrationConfig() + @classmethod def with_fp8_parameters(cls) -> bool: """Should the parameters be stored as FP8""" @@ -616,30 +661,30 @@ def get_fp8_group(cls) -> Union[dist_group_type, None]: return cls.quantization_state.fp8_distributed_group @classmethod - def get_autocast_state(cls) -> tuple: + def get_autocast_state(cls) -> TEAutocastState: """Snapshot the autocast-related fields of the quantization state.""" qstate = cls.quantization_state - return ( - qstate.fp8_enabled, - qstate.fp8_calibration, - qstate.fp8_recipe, - qstate.fp8_distributed_group, - qstate.is_first_fp8_module, - qstate.fp8_graph_capturing, + return TEAutocastState( + fp8_enabled=qstate.fp8_enabled, + fp8_calibration=qstate.fp8_calibration, + calibration_config=qstate.calibration_config, + fp8_recipe=qstate.fp8_recipe, + fp8_distributed_group=qstate.fp8_distributed_group, + is_first_fp8_module=qstate.is_first_fp8_module, + fp8_graph_capturing=qstate.fp8_graph_capturing, ) @classmethod - def set_autocast_state(cls, state: tuple) -> None: + def set_autocast_state(cls, state: TEAutocastState) -> None: """Restore a previously saved autocast state snapshot.""" qstate = cls.quantization_state - ( - qstate.fp8_enabled, - qstate.fp8_calibration, - qstate.fp8_recipe, - qstate.fp8_distributed_group, - qstate.is_first_fp8_module, - qstate.fp8_graph_capturing, - ) = state + qstate.fp8_enabled = state.fp8_enabled + qstate.fp8_calibration = state.fp8_calibration + qstate.calibration_config = state.calibration_config + qstate.fp8_recipe = state.fp8_recipe + qstate.fp8_distributed_group = state.fp8_distributed_group + qstate.is_first_fp8_module = state.is_first_fp8_module + qstate.fp8_graph_capturing = state.fp8_graph_capturing @staticmethod def reduce_tensor_across_group_op_max(tensor: torch.Tensor, group: dist_group_type) -> None: @@ -734,9 +779,13 @@ def autocast_enter( fp8_recipe: Optional[Recipe] = None, fp8_group: Optional[dist_group_type] = None, _graph: bool = False, + calibration_config: Optional[QuantizationCalibrationConfig] = None, ) -> None: """Set state and tracking variables for entry into FP8 region.""" + if calibrating and calibration_config is None: + calibration_config = QuantizationCalibrationConfig() + fp8_recipe = get_default_fp8_recipe() if fp8_recipe is None else fp8_recipe autocast_key = cls.get_unique_autocast_key(fp8_recipe, fp8_group) qstate = cls.quantization_state @@ -746,7 +795,8 @@ def autocast_enter( ) qstate.fp8_enabled = enabled - qstate.fp8_calibration = calibrating + qstate.fp8_calibration = calibration_config is not None + qstate.calibration_config = calibration_config qstate.fp8_recipe = fp8_recipe qstate.fp8_distributed_group = fp8_group qstate.fp8_graph_capturing = _graph @@ -944,6 +994,7 @@ def fp8_autocast( fp8_recipe: Optional[Recipe] = None, fp8_group: Optional[dist_group_type] = None, _graph: bool = False, + calibration_config: Optional[QuantizationCalibrationConfig] = None, ) -> "autocast": """ .. warning:: @@ -966,6 +1017,7 @@ def fp8_autocast( recipe=fp8_recipe, amax_reduction_group=fp8_group, _graph=_graph, + calibration_config=calibration_config, ) @@ -997,10 +1049,12 @@ class autocast: enabled : bool, default = True whether or not to enable low precision quantization (FP8/FP4). calibrating : bool, default = False - calibration mode allows collecting statistics such as amax and scale - data of quantized tensors even when executing without quantization enabled. - This is useful for saving an inference ready checkpoint while training - using a higher precision. + Enables calibration with the default configuration. Calibration + collects and buffers Transformer Engine calibration metadata even when executing + without quantization enabled. + calibration_config : QuantizationCalibrationConfig, default = None + Custom configuration for collecting checkpointable quantization scaling + factors. Providing a config also enables calibration. recipe : recipe.Recipe, default = None recipe used for low precision quantization. amax_reduction_group : torch._C._distributed_c10d.ProcessGroup, default = None @@ -1012,7 +1066,7 @@ class autocast: # to avoid overheads. __slots__ = ( "_enabled", - "_calibrating", + "_calibration_config", "_recipe", "_amax_reduction_group", "_graph", @@ -1026,9 +1080,14 @@ def __init__( recipe: Optional["Recipe"] = None, amax_reduction_group: Optional["dist_group_type"] = None, _graph: bool = False, + calibration_config: Optional[QuantizationCalibrationConfig] = None, ) -> None: self._enabled = enabled - self._calibrating = calibrating + self._calibration_config = ( + QuantizationCalibrationConfig() + if calibrating and calibration_config is None + else calibration_config + ) self._recipe = recipe self._amax_reduction_group = amax_reduction_group self._graph = _graph @@ -1046,7 +1105,8 @@ def __enter__(self) -> "autocast": self._fp8_state = FP8GlobalStateManager.get_autocast_state() FP8GlobalStateManager.autocast_enter( enabled=self._enabled, - calibrating=self._calibrating, + calibrating=False, + calibration_config=self._calibration_config, fp8_recipe=self._recipe, fp8_group=self._amax_reduction_group, _graph=self._graph, diff --git a/transformer_engine/pytorch/quantized_tensor.py b/transformer_engine/pytorch/quantized_tensor.py index 7149a5a163..038f0b53ee 100644 --- a/transformer_engine/pytorch/quantized_tensor.py +++ b/transformer_engine/pytorch/quantized_tensor.py @@ -404,6 +404,7 @@ def __init__(self, *, rowwise: bool, columnwise: bool) -> None: self.columnwise_usage = columnwise self.internal = False self.optimize_for_gemm = False + self._calibration_state: Dict[str, torch.Tensor] = {} def __repr__(self): return ( @@ -564,14 +565,57 @@ def create_metadata( "nontensor_kwargs": meta["nontensor_kwargs"], } - def calibrate(self, tensor: torch.Tensor) -> None: - """Calibrate quantizer state - - Updates quantization state as if quantizing a tensor, but - without actually performing the quantization. + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + """Observe a tensor and update persistent calibration state. + ``calibration_decay`` decays the historical maximum before incorporating + the current observation. A value of zero retains only the current metadata. """ + def get_quantization_recipe_name(self) -> str: + """Get the stable name of the quantization recipe.""" + return "" + + def _update_calibration_value( + self, + metadata_name: str, + observed_value: Optional[torch.Tensor], + *, + calibration_decay: float, + ) -> None: + """Merge an observation into quantizer-owned calibration state.""" + if observed_value is None or torch.isnan(observed_value).any(): + # Un-initialized scale. Ignore it. + return + observed_value = observed_value.detach() + calibration_state = self._calibration_state + calibration_value = calibration_state.get(metadata_name) + if calibration_decay > 0.0: + if calibration_value is not None and calibration_value.shape != observed_value.shape: + raise RuntimeError( + "Quantizer calibration value shape changed from " + f"{tuple(calibration_value.shape)} to {tuple(observed_value.shape)}" + ) + if calibration_value is None: + # Initialize the rolling activation scaling factor. + # Requires CUDA graph warmup step. + calibration_value = torch.zeros_like(observed_value) + calibration_state[metadata_name] = calibration_value + # Track a decaying maximum so early-training activation + # outliers do not permanently determine the inference scale. + calibration_value.mul_(calibration_decay) + torch.maximum(calibration_value, observed_value, out=calibration_value) + else: + # Without scale history, keep a reference to the current metadata + # without allocating or copying a separate buffer. + # Requires CUDA graph warmup step, and only access this value + # at an appropriate time (e.g. checkpointing) if captured by CG. + calibration_state[metadata_name] = observed_value + + def _share_calibration_state_with(self, quantizer: "Quantizer") -> None: + """Make a shallow quantizer copy share persistent calibration state.""" + quantizer._calibration_state = self._calibration_state + def set_usage( self, *, rowwise: Optional[bool] = None, columnwise: Optional[bool] = None ) -> None: diff --git a/transformer_engine/pytorch/tensor/float8_blockwise_tensor.py b/transformer_engine/pytorch/tensor/float8_blockwise_tensor.py index 6105b18a73..2aac4ce6c0 100644 --- a/transformer_engine/pytorch/tensor/float8_blockwise_tensor.py +++ b/transformer_engine/pytorch/tensor/float8_blockwise_tensor.py @@ -243,10 +243,16 @@ def is_quantizable(self, inp: torch.Tensor) -> bool: return False return True - def calibrate(self, tensor: torch.Tensor) -> None: - # NOTE: This interface is specific to requirements like delayed scaling - # where state from an estimator influences distribution parameters. - pass + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + """Float8BlockQuantizer does not yet support calibration.""" + # NOTE(@cspades): Currently, PTQ calibration requirements don't need + # non-global / blockwise scaling factors, which are usually computed + # on-the-fly during inference. Implement this interface for future + # applications of blockwise FP8 calibration. + + def get_quantization_recipe_name(self) -> str: + """Get the stable name of the quantization recipe.""" + return "fp8_block_scaling" def _get_compatible_recipe(self) -> Union[type[Recipe], None]: return Float8BlockScaling diff --git a/transformer_engine/pytorch/tensor/float8_tensor.py b/transformer_engine/pytorch/tensor/float8_tensor.py index cf37c36c59..5ac2846e09 100644 --- a/transformer_engine/pytorch/tensor/float8_tensor.py +++ b/transformer_engine/pytorch/tensor/float8_tensor.py @@ -17,14 +17,14 @@ ) from ..utils import canonicalize_process_group, devices_match, is_non_tn_fp8_gemm_supported from .storage.float8_tensor_storage import Float8TensorStorage, _FromFloat8Func -from ..quantized_tensor import QuantizedTensor, Quantizer +from ..quantized_tensor import QuantizedTensor, QuantizedTensorStorage, Quantizer from ..dynamo import register_value_opaque_quantizer from ._quantization_helpers import ( _IdentityFunc, _resolve_view_shape, safe_quantized_repr, ) -from ..constants import dist_group_type, DType +from ..constants import dist_group_type, DType, TE_DType_To_Torch aten = torch.ops.aten @@ -93,6 +93,7 @@ def copy(self) -> Float8Quantizer: columnwise=self.columnwise_usage, ) quantizer.internal = self.internal + self._share_calibration_state_with(quantizer) return quantizer @@ -124,9 +125,22 @@ def quantize_impl(self, tensor: torch.Tensor) -> QuantizedTensor: """Quantize tensor implementation""" return tex.quantize(tensor, self) - def calibrate(self, tensor: torch.Tensor) -> None: - amin, amax = tensor.aminmax() - self.amax.copy_(torch.max(-amin, amax)) + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + if isinstance(tensor, (QuantizedTensor, QuantizedTensorStorage)): + # Retrieve the quantization metadata from quantized storage. + observed_amax = self.amax + else: + # If quantized amax metadata does not yet exist or calibrate() is called directly, + # then recompute the absmax without quantization. + # This path is NOT performant and should only be used for non-quantized Tensor calibration. + amin, amax = tensor.aminmax() + observed_amax = torch.max(-amin, amax).float().reshape(1) + self.amax.copy_(observed_amax) + self._update_calibration_value("amax", observed_amax, calibration_decay=calibration_decay) + + def get_quantization_recipe_name(self) -> str: + """Get the stable name of the quantization recipe.""" + return "fp8_delayed_scaling" def get_columnwise_shape(self, rowwise_data_shape: Iterable[int]) -> Tuple[int, ...]: """Calculate the shape of the columnwise data for Float8 1D blockwise quantization.""" @@ -276,6 +290,7 @@ def copy(self) -> Float8CurrentScalingQuantizer: ) quantizer.internal = self.internal quantizer.optimize_for_gemm = self.optimize_for_gemm + self._share_calibration_state_with(quantizer) return quantizer @@ -315,9 +330,38 @@ def quantize_impl(self, tensor: torch.Tensor) -> QuantizedTensor: """Quantize tensor implementation""" return tex.quantize(tensor, self) - def calibrate(self, tensor: torch.Tensor) -> None: - # current scaling don't need to calibrate - return + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + """Compute and calibrate quantization metadata.""" + if isinstance(tensor, (QuantizedTensor, QuantizedTensorStorage)): + # Retrieve the quantization metadata from quantized storage. + scale_inv = tensor._scale_inv + else: + # Direct calibration of a non-quantized tensor must reconstruct the metadata. + # This path is NOT performant and should only be used for non-quantized Tensor calibration. + amin, amax = tensor.aminmax() + amax = torch.maximum(-amin, amax).float().reshape(1) + if self.with_amax_reduction and torch.distributed.is_initialized(): + torch.distributed.all_reduce( + amax, + op=torch.distributed.ReduceOp.MAX, + group=self._canonicalized_amax_reduction_group(), + ) + if self.amax_epsilon > 0.0: + amax.clamp_min_(self.amax_epsilon) + fp8_max = torch.finfo(TE_DType_To_Torch[self.dtype]).max + scale = fp8_max / amax + finite_scale = torch.finfo(tensor.dtype).max + scale.nan_to_num_(nan=float("nan"), posinf=finite_scale, neginf=finite_scale) + if self.force_pow_2_scales: + _, exponent = torch.frexp(scale) + scale = torch.ldexp(torch.ones_like(scale), exponent - 1) + scale.masked_fill_(torch.isinf(amax) | (amax == 0), 1.0) + scale_inv = torch.reciprocal(scale) + self._update_calibration_value("scale_inv", scale_inv, calibration_decay=calibration_decay) + + def get_quantization_recipe_name(self) -> str: + """Get the stable name of the quantization recipe.""" + return "fp8_current_scaling" def create_tensor_from_data( self, diff --git a/transformer_engine/pytorch/tensor/hybrid_tensor.py b/transformer_engine/pytorch/tensor/hybrid_tensor.py index 26d0798b92..874d101c96 100644 --- a/transformer_engine/pytorch/tensor/hybrid_tensor.py +++ b/transformer_engine/pytorch/tensor/hybrid_tensor.py @@ -157,6 +157,9 @@ def copy(self) -> "HybridQuantizer": quantizer.optimize_for_gemm = self.optimize_for_gemm return quantizer + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + """HybridQuantizer calibrate() has not yet been implemented.""" + @property def with_amax_reduction(self) -> bool: """Whether either sub-quantizer has cross-rank amax reduction enabled.""" diff --git a/transformer_engine/pytorch/tensor/identity_tensor.py b/transformer_engine/pytorch/tensor/identity_tensor.py index ec171564fe..cb6838d116 100644 --- a/transformer_engine/pytorch/tensor/identity_tensor.py +++ b/transformer_engine/pytorch/tensor/identity_tensor.py @@ -162,9 +162,8 @@ def update_quantized( dst._dtype = data.dtype return dst - def calibrate(self, tensor: torch.Tensor) -> None: - # No state to calibrate. - return + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + """No-op since identity quantization has no calibration state.""" def _get_compatible_recipe(self): # Only reachable via CustomRecipe (qfactory returns IdentityQuantizer). diff --git a/transformer_engine/pytorch/tensor/mxfp8_tensor.py b/transformer_engine/pytorch/tensor/mxfp8_tensor.py index 267806e43e..d6920f3aea 100644 --- a/transformer_engine/pytorch/tensor/mxfp8_tensor.py +++ b/transformer_engine/pytorch/tensor/mxfp8_tensor.py @@ -132,9 +132,16 @@ def is_quantizable(self, inp: torch.Tensor) -> bool: return False return True - def calibrate(self, tensor: torch.Tensor) -> None: - # TODO(ksivamani): No calibration needed for mxfp8? - pass + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + """Calibrate an MXFP8 tensor.""" + # NOTE(@cspades): Currently, PTQ calibration requirements don't need + # non-global / blockwise scaling factors, which are usually computed + # on-the-fly during inference. Implement this interface for future + # applications of MXFP8 calibration. + + def get_quantization_recipe_name(self) -> str: + """Get the stable name of the quantization recipe.""" + return "mxfp8" def get_scale_shape( self, diff --git a/transformer_engine/pytorch/tensor/nvfp4_tensor.py b/transformer_engine/pytorch/tensor/nvfp4_tensor.py index 5e537c3ff0..08b5be2bfb 100644 --- a/transformer_engine/pytorch/tensor/nvfp4_tensor.py +++ b/transformer_engine/pytorch/tensor/nvfp4_tensor.py @@ -22,7 +22,7 @@ ) from .storage.nvfp4_tensor_storage import NVFP4TensorStorage, _FromNVFP4Func -from ..quantized_tensor import QuantizedTensor, Quantizer +from ..quantized_tensor import QuantizedTensor, QuantizedTensorStorage, Quantizer from ..dynamo import register_value_opaque_quantizer from ._quantization_helpers import _IdentityFunc, safe_quantized_repr @@ -247,6 +247,7 @@ def copy(self) -> NVFP4Quantizer: ) quantizer.internal = self.internal quantizer.optimize_for_gemm = self.optimize_for_gemm + self._share_calibration_state_with(quantizer) return quantizer @@ -339,8 +340,34 @@ def convert_shape_for_fp4(shape: Iterable[int]) -> Tuple[int, ...]: shape[-1] = shape[-1] // 2 return tuple(shape) - def calibrate(self, tensor: torch.Tensor) -> None: - pass # Calibration is no-op + def calibrate(self, tensor: torch.Tensor, *, calibration_decay: float = 0.0) -> None: + metadata_name = "amax_rowwise" if self.row_scaled_nvfp4 else "amax" + if isinstance(tensor, (QuantizedTensor, QuantizedTensorStorage)): + # Retrieve the quantization metadata from quantized storage. + observed_amax = tensor._amax_rowwise + else: + # Direct calibration of a non-quantized tensor must reconstruct the metadata. + # This path is NOT performant and should only be used for non-quantized Tensor calibration. + if self.row_scaled_nvfp4: + amin, amax = tensor.aminmax(dim=-1) + else: + amin, amax = tensor.aminmax() + observed_amax = torch.maximum(-amin, amax).reshape(-1).float() + if self.with_amax_reduction and torch.distributed.is_initialized(): + torch.distributed.all_reduce( + observed_amax, + op=torch.distributed.ReduceOp.MAX, + group=self._canonicalized_amax_reduction_group(), + ) + self._update_calibration_value( + metadata_name, + observed_amax, + calibration_decay=calibration_decay, + ) + + def get_quantization_recipe_name(self) -> str: + """Get the stable name of the quantization recipe.""" + return "nvfp4_rowwise" if self.row_scaled_nvfp4 else "nvfp4" def _canonicalized_amax_reduction_group(self) -> dist_group_type: """Get process group for amax reduction""" diff --git a/transformer_engine/pytorch/tensor/utils.py b/transformer_engine/pytorch/tensor/utils.py index cef45c0223..fef222dcff 100644 --- a/transformer_engine/pytorch/tensor/utils.py +++ b/transformer_engine/pytorch/tensor/utils.py @@ -27,6 +27,14 @@ from ..constants import NVFP4_BLOCK_SCALING_SIZE, DType +def get_quantization_recipe_name(quantizer: Optional[Quantizer]) -> str: + """Get a stable recipe name from a quantizer.""" + quantizer = getattr(quantizer, "parent_quantizer", quantizer) + if quantizer is None: + return "" + return quantizer.get_quantization_recipe_name() + + def replace_raw_data(tensor: QuantizedTensor, new_raw_data: torch.Tensor): r"""Change a quantized tensor's data buffer while preserving values