communication_op.py

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

from typing import Any, Optional, Union, Tuple

import torch
import torch.distributed

from .parallel_state import get_tp_group

def tensor_model_parallel_all_reduce(input_: torch.Tensor) -> torch.Tensor:
    """All-reduce the input tensor across model parallel group."""
    return get_tp_group().all_reduce(input_)

def tensor_model_parallel_all_reduce_crp_m32(input_: torch.Tensor,
                                         pa_rms_weight: torch.Tensor,
                                         pa_residual: torch.Tensor,
                                         pa_rms_eps: float,
                                         pa_quant_dtype: Optional[torch.dtype] = torch.int8,
                                         update_input: Optional[bool] = True) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
    """All-reduce the input tensor across model parallel group."""
    # allreduce fused rms and quant
    return get_tp_group().all_reduce_crq_m32(input_=input_, 
                                         pa_rms_weight=pa_rms_weight,
                                         pa_residual=pa_residual,
                                         pa_rms_eps=pa_rms_eps,
                                         pa_quant_dtype=pa_quant_dtype,
                                         update_input=update_input)

def tensor_model_parallel_all_gather(input_: torch.Tensor,
                                     dim: int = -1) -> torch.Tensor:
    """All-gather the input tensor across model parallel group."""
    return get_tp_group().all_gather(input_, dim)


def tensor_model_parallel_reduce_scatter(input_: torch.Tensor,
                                         dim: int = -1) -> torch.Tensor:
    """Reduce-Scatter the input tensor across model parallel group."""
    return get_tp_group().reduce_scatter(input_, dim)


def tensor_model_parallel_gather(input_: torch.Tensor,
                                 dst: int = 0,
                                 dim: int = -1) -> Optional[torch.Tensor]:
    """Gather the input tensor across model parallel group."""
    return get_tp_group().gather(input_, dst, dim)


def broadcast_tensor_dict(tensor_dict: Optional[dict[Any, Union[torch.Tensor,
                                                                Any]]] = None,
                          src: int = 0):
    if not torch.distributed.is_initialized():
        return tensor_dict
    return get_tp_group().broadcast_tensor_dict(tensor_dict, src)