lightseekorg--tokenspeed/python/tokenspeed/runtime/layers/dense/fp8.py

# SPDX-License-Identifier: MIT AND Apache-2.0
# SPDX-FileCopyrightText: Copyright (c) 2026 LightSeek Foundation
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
#
# Copyright (c) 2026 LightSeek Foundation
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.


import logging

import tokenspeed_kernel
import torch
from tokenspeed_kernel.ops.gemm.fp8_utils import (
    per_token_group_quant_fp8,
    per_token_quant_fp8,
    static_quant_fp8,
)
from tokenspeed_kernel.platform import Platform
from torch.nn.parameter import Parameter

logger = logging.getLogger(__name__)

try:
    from tokenspeed_kernel.thirdparty.deep_gemm import ceil_to_ue8m0 as _ceil_to_ue8m0
    from tokenspeed_kernel.thirdparty.deep_gemm import (
        transform_sf_into_required_layout as _transform_sf,
    )
except ImportError:
    _ceil_to_ue8m0 = None
    _transform_sf = None

from tokenspeed.runtime.layers.dense.utils import normalize_e4m3fn_to_e4m3fnuz
from tokenspeed.runtime.layers.parameter import (
    BlockQuantScaleParameter,
    ModelWeightParameter,
    PerTensorScaleParameter,
)
from tokenspeed.runtime.layers.quantization.base_config import LinearMethodBase
from tokenspeed.runtime.layers.quantization.fp8 import Fp8Config
from tokenspeed.runtime.layers.quantization.utils import convert_to_channelwise

platform = Platform.get()


class Fp8LinearMethod(LinearMethodBase):
    """Linear method for FP8.
    Supports loading FP8 checkpoints with static weight scale and
    dynamic/static activation scale.

    Also supports loading quantized FP16/BF16 model checkpoints with dynamic
    activation scaling. The weight scaling factor will be initialized after
    the model weights are loaded.

    Limitations:
    1. Only support per-tensor quantization due to torch._scaled_mm support.
    2. Only support float8_e4m3fn data type due to the limitation of
       torch._scaled_mm (https://github.com/pytorch/pytorch/blob/2e48b39603411a41c5025efbe52f89560b827825/aten/src/ATen/native/cuda/Blas.cpp#L854-L856)

    Args:
        quant_config: The quantization config.
    """

    def __init__(self, quant_config: Fp8Config):
        self.quant_config = quant_config
        self.block_quant = self.quant_config.weight_block_size is not None

    def create_weights(
        self,
        layer: torch.nn.Module,
        input_size_per_partition: int,
        output_partition_sizes: list[int],
        input_size: int,
        output_size: int,
        params_dtype: torch.dtype,
        **extra_weight_attrs,
    ):
        output_size_per_partition = sum(output_partition_sizes)
        weight_loader = extra_weight_attrs.get("weight_loader")

        if self.block_quant:
            block_n, block_k = (
                self.quant_config.weight_block_size[0],
                self.quant_config.weight_block_size[1],
            )
            # Required by row parallel
            if input_size > input_size_per_partition:
                if input_size_per_partition % block_k != 0:
                    raise ValueError(
                        f"Weight input_size_per_partition = "
                        f"{input_size_per_partition} is not divisible by "
                        f"weight quantization block_k = {block_k}."
                    )
            # Required by column parallel or enabling merged weights
            if (
                output_size > output_size_per_partition
                or len(output_partition_sizes) > 1
            ):
                for output_partition_size in output_partition_sizes:
                    if output_partition_size % block_n != 0:
                        raise ValueError(
                            f"Weight output_partition_size = "
                            f"{output_partition_size} is not divisible by "
                            f"weight quantization block_n = {block_n}."
                        )

        layer.logical_widths = output_partition_sizes
        layer.input_size_per_partition = input_size_per_partition
        layer.output_size_per_partition = output_size_per_partition
        layer.orig_dtype = params_dtype

        # WEIGHT
        weight_dtype = (
            torch.float8_e4m3fn
            if self.quant_config.is_checkpoint_fp8_serialized
            else params_dtype
        )

        weight = ModelWeightParameter(
            data=torch.empty(
                output_size_per_partition, input_size_per_partition, dtype=weight_dtype
            ),
            input_dim=1,
            output_dim=0,
            weight_loader=weight_loader,
        )
        layer.register_parameter("weight", weight)

        # If checkpoint is serialized fp8, load them.
        # Otherwise, wait until process_weights_after_loading.
        if self.quant_config.is_checkpoint_fp8_serialized:
            # WEIGHT SCALE
            if self.block_quant:
                if hasattr(self.quant_config, "activation_scheme"):
                    if self.quant_config.activation_scheme != "dynamic":
                        raise ValueError(
                            "Block FP8 requires dynamic activation quantization."
                        )
                elif hasattr(self.quant_config, "linear_activation_scheme"):
                    if self.quant_config.linear_activation_scheme != "dynamic":
                        raise ValueError(
                            "Block FP8 requires dynamic linear activation quantization."
                        )
                scale = BlockQuantScaleParameter(
                    data=torch.empty(
                        (output_size_per_partition + block_n - 1) // block_n,
                        (input_size_per_partition + block_k - 1) // block_k,
                        dtype=torch.float32,
                    ),
                    input_dim=1,
                    output_dim=0,
                    weight_loader=weight_loader,
                )
                scale[:] = torch.finfo(torch.float32).min
                layer.register_parameter("weight_scale_inv", scale)
            else:
                scale = PerTensorScaleParameter(
                    data=torch.empty(len(output_partition_sizes), dtype=torch.float32),
                    weight_loader=weight_loader,
                )
                scale[:] = torch.finfo(torch.float32).min
                layer.register_parameter("weight_scale", scale)

            # INPUT ACTIVATION SCALE
            if (
                hasattr(self.quant_config, "activation_scheme")
                and self.quant_config.activation_scheme == "static"
            ) or (
                hasattr(self.quant_config, "linear_activation_scheme")
                and self.quant_config.linear_activation_scheme == "static"
            ):
                scale = PerTensorScaleParameter(
                    data=torch.empty(len(output_partition_sizes), dtype=torch.float32),
                    weight_loader=weight_loader,
                )

                scale[:] = torch.finfo(torch.float32).min
                layer.register_parameter("input_scale", scale)
            else:
                layer.register_parameter("input_scale", None)

    def process_weights_after_loading(self, layer: torch.nn.Module) -> None:
        if self.block_quant:
            # If ROCm, normalize the weights and scales to e4m3fnuz
            if platform.is_fp8e4m3fnuz:
                # activation_scheme: dynamic
                weight, weight_scale, _ = normalize_e4m3fn_to_e4m3fnuz(
                    weight=layer.weight,
                    weight_scale=layer.weight_scale_inv,
                    input_scale=None,
                )
                layer.input_scale = None
            else:
                weight, weight_scale = layer.weight.data, layer.weight_scale_inv.data
            layer.weight.data = weight.data
            layer.weight_scale_inv.data = weight_scale.data
            layer._use_deep_gemm_fp8 = False
            is_bmm = getattr(layer, "is_bmm", False)
            is_ue8m0 = getattr(self.quant_config, "scale_fmt", None) == "ue8m0"
            if _transform_sf is not None and _ceil_to_ue8m0 is not None and is_ue8m0:
                N, K = layer.weight.shape
                block_n, block_k = self.quant_config.weight_block_size
                if is_bmm:
                    # Grouped (batched) projection (V4 attention wo_a, weight
                    # [groups * n, K], consumed per group as [n, K]). Transform
                    # the block scale into the deep_gemm MN-major layout with the
                    # group axis so deep_gemm.fp8_einsum("bhr,hdr->bhd") runs the
                    # output projection as one native FP8 GEMM (no FP32 dequant).
                    # recipe is (1, block_n, block_k) at load; the runtime einsum
                    # uses (1, 1, block_n) on SM100.
                    g = layer.bmm_batch_size
                    n = N // g
                    if n % block_n == 0 and K % block_k == 0:
                        sf = _ceil_to_ue8m0(layer.weight_scale_inv.data).view(
                            g, n // block_n, K // block_k
                        )
                        layer.weight_scale_inv.data = _transform_sf(
                            sf=sf,
                            mn=n,
                            k=K,
                            recipe=(1, block_n, block_k),
                            num_groups=g,
                            is_sfa=False,
                        )
                        layer._deep_gemm_block_size = [block_n, block_k]
                        layer._use_deep_gemm_fp8 = True
                elif N % 64 == 0 and K % 128 == 0:
                    sf = _ceil_to_ue8m0(layer.weight_scale_inv.data)
                    layer.weight_scale_inv.data = _transform_sf(
                        sf=sf,
                        mn=N,
                        k=K,
                        recipe=(1, block_n, block_k),
                        is_sfa=False,
                    )
                    layer._use_deep_gemm_fp8 = True
            if is_bmm and not layer._use_deep_gemm_fp8:
                # The is_bmm runtime path (DeepSeek-V4 o_proj) has no FP32
                # fallback, so fail fast at load with a clear message instead of
                # a cryptic AttributeError on the first forward.
                raise RuntimeError(
                    "is_bmm weight requires the deep_gemm FP8 block-scale path "
                    "but it could not be prepared (deep_gemm_available="
                    f"{_transform_sf is not None}, ue8m0={is_ue8m0}, "
                    f"weight={tuple(layer.weight.shape)}); ensure FP8 block-quant "
                    "ue8m0 weights with block-aligned dims and deep_gemm installed."
                )
        else:
            layer.weight = Parameter(layer.weight.data, requires_grad=False)

            # If checkpoint not serialized fp8, quantize the weights.
            if not self.quant_config.is_checkpoint_fp8_serialized:
                # apply per-channel quantization default as
                qweight, weight_scale = per_token_group_quant_fp8(
                    layer.weight, layer.weight.shape[-1]
                )
                weight_scale = weight_scale.t().contiguous()

                # Update the layer with the new values.
                layer.weight = Parameter(qweight.t(), requires_grad=False)
                layer.weight_scale = Parameter(weight_scale, requires_grad=False)
                layer.input_scale = None

            # If checkpoint is fp8, handle that there are N scales for N
            # shards in a fused module
            else:
                layer.weight_scale = Parameter(
                    layer.weight_scale.data, requires_grad=False
                )
                if (
                    hasattr(self.quant_config, "activation_scheme")
                    and self.quant_config.activation_scheme == "static"
                ) or (
                    hasattr(self.quant_config, "linear_activation_scheme")
                    and self.quant_config.linear_activation_scheme == "static"
                ):
                    layer.input_scale = Parameter(
                        layer.input_scale.data, requires_grad=False
                    )

                weight = layer.weight
                weight_scale = convert_to_channelwise(
                    layer.weight_scale, layer.logical_widths
                )

                # Update layer with new values.
                layer.weight = Parameter(weight.t(), requires_grad=False)
                layer.weight_scale = Parameter(weight_scale, requires_grad=False)
                if (
                    hasattr(self.quant_config, "activation_scheme")
                    and self.quant_config.activation_scheme == "static"
                ) or (
                    hasattr(self.quant_config, "linear_activation_scheme")
                    and self.quant_config.linear_activation_scheme == "static"
                ):
                    layer.input_scale = Parameter(
                        layer.input_scale.max(), requires_grad=False
                    )

    def apply(
        self,
        layer: torch.nn.Module,
        x: torch.Tensor,
        bias: torch.Tensor | None = None,
        block_scale: torch.Tensor | None = None,
        output_dtype: torch.dtype | None = None,
    ) -> torch.Tensor:

        if self.block_quant:
            input_2d = x.view(-1, x.shape[-1])
            output_shape = [*x.shape[:-1], layer.weight.shape[0]]
            output_dtype = output_dtype or x.dtype

            override = (
                "deep_gemm_mm_fp8_blockscale"
                if getattr(layer, "_use_deep_gemm_fp8", False)
                else None
            )
            output = tokenspeed_kernel.mm(
                input_2d,
                layer.weight,
                A_scales=block_scale,
                B_scales=layer.weight_scale_inv,
                bias=bias,
                out_dtype=output_dtype,
                quant="mxfp8",
                block_size=self.quant_config.weight_block_size,
                override=override,
            )
            return output.to(dtype=output_dtype).view(*output_shape)
        else:
            input = x
            weight = layer.weight
            weight_scale = layer.weight_scale
            input_scale = layer.input_scale

            # View input as 2D matrix for fp8 methods
            input_2d = input.view(-1, input.shape[-1])
            output_shape = [*input.shape[:-1], weight.shape[1]]

            if input_scale is not None:
                if input_scale.numel() != 1:
                    raise ValueError(
                        f"input_scale must contain exactly one value, got {input_scale.numel()}."
                    )
                qinput, x_scale = static_quant_fp8(input_2d, input_scale)
            else:
                qinput, x_scale = per_token_quant_fp8(input_2d)

            qinput = qinput.view(-1, qinput.shape[-1])

            output = tokenspeed_kernel.mm(
                qinput,
                weight,
                A_scales=x_scale,
                B_scales=weight_scale,
                out_dtype=input.dtype,
            )
            if bias is not None:
                output = output + bias
            return output.view(*output_shape)