/ludwig/utils/torch_utils.py

import math
import os
import warnings
from abc import abstractmethod
from functools import lru_cache

import torch
from torch import nn
from torch.nn import Module, ModuleDict

from ludwig.api_annotations import DeveloperAPI
from ludwig.constants import ENCODER_OUTPUT
from ludwig.utils.entmax.activations import Entmax15, Sparsemax
from ludwig.utils.strings_utils import SpecialSymbol

_TORCH_INIT_PARAMS: tuple | None = None


@DeveloperAPI
def get_torch_device():
    if torch.cuda.is_available() and torch.cuda.device_count() > 0:
        # Use cublasLt for batched GEMM operations. The default cublas library has known
        # bugs with cublasSgemmStridedBatched on certain GPU/driver combinations.
        torch.backends.cuda.preferred_blas_library("cublaslt")
        return "cuda"

    if bool(os.environ.get("LUDWIG_ENABLE_MPS")):
        if torch.backends.mps.is_available() and torch.backends.mps.is_built():
            if not bool(os.environ.get("PYTORCH_ENABLE_MPS_FALLBACK")):
                warnings.warn(
                    "LUDWIG_ENABLE_MPS is set and MPS is available, but PYTORCH_ENABLE_MPS_FALLBACK has not been set. "
                    "Depending on your model config, some operations may not be compatible. If errors occur, try "
                    "setting `PYTORCH_ENABLE_MPS_FALLBACK=1` and resubmitting."
                )
            return "mps"
        else:
            warnings.warn("LUDWIG_ENABLE_MPS is set but MPS is not available, falling back to CPU.")

    return "cpu"


DEVICE = get_torch_device()


@DeveloperAPI
def place_on_device(x, device):
    """Recursively places the input on the specified device."""
    if isinstance(x, list):
        return [place_on_device(xi, device) for xi in x]
    elif isinstance(x, dict):
        return {k: place_on_device(v, device) for k, v in x.items()}
    elif isinstance(x, set):
        return {place_on_device(xi, device) for xi in x}
    elif isinstance(x, tuple):
        return tuple(place_on_device(xi, device) for xi in x)
    elif isinstance(x, torch.Tensor):
        return x.to(device)
    else:
        return x


@DeveloperAPI
def sequence_length_2D(sequence: torch.Tensor) -> torch.Tensor:
    """Returns the number of non-padding elements per sequence in batch.

    Args:
        sequence: A 2D tensor of shape [batch size x max sequence length].

    Returns:
        The count of non-zero elements per sequence.
    """
    used = (sequence != SpecialSymbol.PADDING.value).type(torch.int32)
    length = torch.sum(used, 1)
    return length


@DeveloperAPI
def sequence_length_3D(sequence: torch.Tensor) -> torch.Tensor:
    """Returns the number of non-zero elements per sequence in batch.

    Args:
        sequence: A 3D tensor of shape [batch size x max sequence length x hidden size].

    Returns:
        The count of non-zero elements per sequence.
    """
    used = torch.sign(torch.amax(torch.abs(sequence), dim=2))
    length = torch.sum(used, 1)
    length = length.int()
    return length


@DeveloperAPI
def sequence_mask(lengths: torch.Tensor, maxlen: int | None = None, dtype: torch.dtype = torch.bool):
    """Returns a mask of shape (batch_size x maxlen), where mask[i] is True for each element up to lengths[i],
    otherwise False i.e. if maxlen=5 and lengths[i] = 3, mask[i] = [True, True True, False False].

    Args:
        lengths: A 1d integer tensor of shape [batch size].
        maxlen: The maximum sequence length. If not specified, the max(lengths) is used.
        dtype: The type to output.

    Returns:
        A sequence mask tensor of shape (batch_size x maxlen).
    """
    if maxlen is None:
        maxlen = lengths.max()
    matrix = torch.unsqueeze(lengths, dim=-1)
    row_vector = torch.arange(0, maxlen, 1, device=lengths.device)
    mask = row_vector < matrix
    mask = mask.type(dtype)
    return mask


@DeveloperAPI
def periodic(inputs: torch.Tensor, period: int) -> torch.Tensor:
    """Returns periodic representation assuming 0 is start of period."""
    return torch.cos(inputs * 2 * math.pi / period)


initializer_registry = {
    "uniform": nn.init.uniform_,
    "normal": nn.init.normal_,
    "constant": nn.init.constant_,
    "ones": nn.init.ones_,
    "zeros": nn.init.zeros_,
    "eye": nn.init.eye_,
    "dirac": nn.init.dirac_,
    "xavier_uniform": nn.init.xavier_uniform_,
    "xavier_normal": nn.init.xavier_normal_,
    "kaiming_uniform": nn.init.kaiming_uniform_,
    "kaiming_normal": nn.init.kaiming_normal_,
    "orthogonal": nn.init.orthogonal_,
    "sparse": nn.init.sparse_,
    "identity": nn.init.eye_,
}


class SwiGLU(nn.Module):
    """SiLU-Gated Linear Unit (Shazeer, 2020).

    Used in LLaMA/PaLM.
    Splits input in half along last dim, applies SiLU to gate half, multiplies.
    Note: input_size must be 2x the desired output_size.
    """

    def forward(self, x):
        x1, x2 = x.chunk(2, dim=-1)
        return nn.functional.silu(x1) * x2


class GeGLU(nn.Module):
    """GELU-Gated Linear Unit (Shazeer, 2020).

    Used in T5 v1.1. Splits input in half along last dim, applies GELU to gate half, multiplies.
    """

    def forward(self, x):
        x1, x2 = x.chunk(2, dim=-1)
        return nn.functional.gelu(x1) * x2


class ReGLU(nn.Module):
    """ReLU-Gated Linear Unit (Shazeer, 2020).

    Splits input in half along last dim, applies ReLU to gate half, multiplies.
    """

    def forward(self, x):
        x1, x2 = x.chunk(2, dim=-1)
        return nn.functional.relu(x1) * x2


activations = {
    "elu": nn.ELU,
    "leakyRelu": nn.LeakyReLU,
    "logSigmoid": nn.LogSigmoid,
    "relu": nn.ReLU,
    "sigmoid": nn.Sigmoid,
    "tanh": nn.Tanh,
    "softmax": nn.Softmax,
    "gelu": nn.GELU,
    "silu": nn.SiLU,
    "swish": nn.SiLU,
    "mish": nn.Mish,
    "selu": nn.SELU,
    "prelu": nn.PReLU,
    "relu6": nn.ReLU6,
    "hardswish": nn.Hardswish,
    "hardsigmoid": nn.Hardsigmoid,
    "softplus": nn.Softplus,
    "celu": nn.CELU,
    "swiglu": SwiGLU,
    "geglu": GeGLU,
    "reglu": ReGLU,
    "sparsemax": Sparsemax,
    "entmax15": Entmax15,
    None: nn.Identity,
}


@DeveloperAPI
def get_activation(activation):
    return activations[activation]()


@DeveloperAPI
def reg_loss(model: nn.Module, regularizer: str, l1: float = 0.01, l2: float = 0.01):
    """Computes the regularization loss for a given model.

    Parameters:
        model: torch.nn.Module object to compute regularization loss for.
        regularizer: regularizer to use (currently l1, l2 and l1_l2 supported).
        l1: L1 regularization coefficient.
        l2: L2 regularization coefficient.

    Returns:
        Regularization loss for the model (float).
    """

    if regularizer == "l1":
        l1_reg = l1 * sum(torch.abs(p).sum() for p in model.parameters())
        return l1_reg
    if regularizer == "l2":
        l2_reg = l2 * sum(torch.square(p).sum() for p in model.parameters())
        return l2_reg
    if regularizer == "l1_l2":
        l1_reg = l1 * sum(torch.abs(p).sum() for p in model.parameters())
        l2_reg = l2 * sum(torch.square(p).sum() for p in model.parameters())
        return l1_reg + l2_reg


@DeveloperAPI
class LudwigModule(Module):
    def __init__(self):
        super().__init__()
        self._losses = {}
        self.register_buffer("device_tensor", torch.zeros(0), persistent=False)

    @property
    def device(self):
        return self.device_tensor.device

    def prepare_for_training(self):
        """This is called from within the Trainer object to do any final instantiation before model training."""

    def losses(self):
        collected_losses = []
        for loss in self._losses.values():
            collected_losses.append(loss)

        for child in self.children():
            if isinstance(child, LudwigModule):
                collected_losses.extend(child.losses())
            elif isinstance(child, ModuleDict):
                for c in child.values():
                    if hasattr(c, "losses"):  # Some modules, i.e. SequenceReducers, don't have losses.
                        collected_losses.extend(c.losses())
            elif isinstance(child, Module):
                pass
            else:
                raise ValueError

        return collected_losses

    def update_loss(self, key: str, loss: torch.Tensor):
        """This should be called in the forward pass to add a custom loss term to the combined loss."""
        self._losses[key] = loss

    @property
    def input_dtype(self):
        return torch.float32

    @property
    @abstractmethod
    def input_shape(self) -> torch.Size:
        """Returns size of the input tensor without the batch dimension."""
        # raise NotImplementedError("Abstract class.")

    @property
    def output_shape(self) -> torch.Size:
        """Returns size of the output tensor without the batch dimension."""
        return self._computed_output_shape()

    @lru_cache(maxsize=1)
    def _computed_output_shape(self) -> torch.Size:
        dummy_input = torch.rand(2, *self.input_shape, device=self.device)
        output_tensor = self.forward(dummy_input.type(self.input_dtype))

        if isinstance(output_tensor, torch.Tensor):
            return output_tensor.size()[1:]
        elif isinstance(output_tensor, dict) and ENCODER_OUTPUT in output_tensor:
            return output_tensor[ENCODER_OUTPUT].size()[1:]
        else:
            raise ValueError(
                f"Cannot determine output shape from encoder output of type {type(output_tensor).__name__}.\n"
                f"Expected a torch.Tensor or a dict with key '{ENCODER_OUTPUT}'."
            )


def freeze_parameters(module: nn.Module):
    """Freezes the parameters of a torch module."""
    for p in module.parameters():
        p.requires_grad = False


@DeveloperAPI
class FreezeModule(nn.Module):
    def __init__(self, module: nn.Module, frozen: bool):
        super().__init__()
        if frozen:
            freeze_parameters(module)
            module.eval()
        else:
            module.train()
        self.module = module
        self.frozen = frozen

    def train(self, mode: bool = True):
        if self.frozen:
            # Ignores any attempt to set params trainable
            return self

        return super().train(mode)


@DeveloperAPI
class Dense(LudwigModule):
    def __init__(
        self,
        input_size,
        output_size,
        use_bias=True,
        weights_initializer="xavier_uniform",
        bias_initializer="zeros",
    ):
        super().__init__()
        self.dense = nn.Linear(in_features=input_size, out_features=output_size, bias=use_bias)
        weights_initializer = initializer_registry[weights_initializer]
        weights_initializer(self.dense.weight)

        if use_bias:
            bias_initializer = initializer_registry[bias_initializer]
            bias_initializer(self.dense.bias)

    @property
    def input_shape(self) -> torch.Size:
        return self.dense.input_shape

    def forward(self, input: torch.Tensor) -> torch.Tensor:
        output = torch.squeeze(self.dense(input), dim=-1)
        return output


@DeveloperAPI
def initialize_pytorch(
    gpus: int | str | list[int] | None = None,
    gpu_memory_limit: float | None = None,
    allow_parallel_threads: bool = True,
):
    param_tuple = (gpus, gpu_memory_limit, allow_parallel_threads)
    if _TORCH_INIT_PARAMS is not None:
        if param_tuple != _TORCH_INIT_PARAMS:
            warnings.warn(
                "PyTorch has already been initialized. Changes to `gpus`, "
                "`gpu_memory_limit`, and `allow_parallel_threads` will be ignored. "
                "Start a new Python process to modify these values."
            )
        return

    # For reproducivility / determinism, set parallel threads to 1.
    # For performance, leave unset to allow PyTorch to select the best value automatically.
    if not allow_parallel_threads:
        torch.set_num_threads(1)
        torch.set_num_interop_threads(1)
        if torch.cuda.is_available() and torch.cuda.device_count() > 0:
            torch.backends.cudnn.deterministic = True
            torch.backends.cudnn.benchmark = False

    if isinstance(gpus, int):
        gpus = [gpus]
    elif isinstance(gpus, str):
        gpus = gpus.strip()
        gpus = [int(g) for g in gpus.split(",")]

    if gpus and len(gpus) == 1 and gpus[0] == -1:
        # CUDA_VISIBLE_DEVICES syntax for disabling all GPUs
        os.environ["CUDA_VISIBLE_DEVICES"] = ""
    elif torch.cuda.is_available() and torch.cuda.device_count() > 0:
        # Set visible devices so GPU utilization is isolated
        # (no GPU contention between workers).
        if gpus is not None:
            if len(gpus) == 1:
                torch.cuda.set_device(gpus[0])
            elif len(gpus) > 1:
                os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(str(i) for i in gpus)

        # Limit the amount of memory that can be consumed per GPU
        if gpu_memory_limit is not None:
            for gpu in gpus or range(torch.cuda.device_count()):
                torch.cuda.memory.set_per_process_memory_fraction(gpu_memory_limit, gpu)

    _set_torch_init_params(param_tuple)


def _set_torch_init_params(params: tuple | None):
    global _TORCH_INIT_PARAMS
    _TORCH_INIT_PARAMS = params


def _get_torch_init_params() -> tuple | None:
    return _TORCH_INIT_PARAMS


@DeveloperAPI
def model_size(model: nn.Module):
    """Computes PyTorch model size in bytes."""
    size = 0
    size += sum(param.nelement() * param.element_size() for param in model.parameters())
    size += sum(buffer.nelement() * buffer.element_size() for buffer in model.buffers())
    return size