ColossalAI/colossalai/communication/utils.py

import torch
import torch.distributed as dist

from colossalai.context.parallel_mode import ParallelMode
from colossalai.core import global_context as gpc
from colossalai.utils import get_current_device


def send_tensor_meta(tensor, need_meta=True, next_rank=None):
    """Sends tensor meta information before sending a specific tensor. 
    Since the recipient must know the shape of the tensor in p2p communications,
    meta information of the tensor should be sent before communications. This function
    synchronizes with :func:`recv_tensor_meta`.

    :param tensor: Tensor to be sent
    :param need_meta: If False, meta information won't be sent
    :param next_rank: The rank of the next member in pipeline parallel group
    :type tensor: Tensor
    :type need_meta: bool, optional
    :type next_rank: int
    :return: False
    :rtype: bool
    """
    if need_meta:
        if next_rank is None:
            next_rank = gpc.get_next_global_rank(ParallelMode.PIPELINE)

        tensor_kwargs = {'dtype': torch.long, 'device': get_current_device()}

        send_shape = torch.tensor(tensor.size(), **tensor_kwargs)
        send_ndims = torch.tensor(len(tensor.size()), **tensor_kwargs)
        ops = [
            dist.P2POp(dist.isend, send_ndims, next_rank),
            dist.P2POp(dist.isend, send_shape, next_rank)
        ]
        reqs = dist.batch_isend_irecv(ops)
        for req in reqs:
            req.wait()
        torch.cuda.synchronize()

    return False


def recv_tensor_meta(tensor_shape, prev_rank=None):
    """Recieves tensor meta information before recieving a specific tensor. 
    Since the recipient must know the shape of the tensor in p2p communications,
    meta information of the tensor should be recieved before communications. This function
    synchronizes with :func:`send_tensor_meta`.

    :param tensor_shape: The shape of the tensor to be recieved
    :param prev_rank: The rank of the source of the tensor
    :type tensor_shape: torch.Size
    :type prev_rank: int, optional
    :return: The shape of the tensor to be recieved
    :rtype: torch.Size
    """
    if tensor_shape is None:
        if prev_rank is None:
            prev_rank = gpc.get_prev_global_rank(ParallelMode.PIPELINE)

        tensor_kwargs = {'dtype': torch.long, 'device': get_current_device()}

        recv_ndims = torch.empty((), **tensor_kwargs)
        dist.recv(recv_ndims, prev_rank)
        recv_shape = torch.empty(recv_ndims, **tensor_kwargs)
        dist.recv(recv_shape, prev_rank)

        tensor_shape = torch.Size(recv_shape)

    return tensor_shape