code: def reduce_tensor_dp_group(
      tensor: torch.Tensor,
      dtype: Optional[torch.dtype] = None,
      dst_local_rank: Optional[int] = None,
      dst_global_rank: Optional[int] = None,
      group: Optional[dist.ProcessGroup] = None,
  ):
      """
      Reduce the tensor in the data parallel process group
  
      :param tensor: A tensor object to reduce/all-reduce
      :param dtype: The data type used in communication
      :param dst_rank: The source rank for reduce. If dst_rank is None,
      :param parallel_mode: Communication parallel mode
      all-reduce will be used instead of reduce. Default is None.
  
      :type tensor: torch.Tensor
      :type dtype: torch.dtype, optional
      :type dst_rank: int, optional
      :type pg: ProcessGroup, optional
      """
      # use the original dtype
      if dtype is None:
          dtype = tensor.dtype
  
      # cast the data to specified dtype for reduce/all-reduce
      if tensor.dtype != dtype:
          tensor_to_reduce = tensor.to(dtype)
      else:
          tensor_to_reduce = tensor
  
      world_size = dist.get_world_size(group=group)
      tensor_to_reduce.div_(world_size)
  
      # if rank is None, all reduce will be used
      # else, reduce is used
      use_all_reduce = dst_local_rank is None
  
      if use_all_reduce:
          dist.all_reduce(tensor_to_reduce, group=group)
      else:
          dist.reduce(
              tensor=tensor_to_reduce, dst=dst_global_rank, group=group
          )
  
      # recover the original dtype
      if tensor.dtype != dtype and tensor is not tensor_to_reduce:
          local_rank = dist.get_rank(group=group)
          if use_all_reduce or dst_local_rank == local_rank:
              tensor.copy_(tensor_to_reduce)
  
      return tensor
  
file_path: ColossalAI/colossalai/zero/low_level/_utils.py
imports: import math
  from typing import Optional
  import torch
  import torch.distributed
  from torch._utils import _flatten_dense_tensors, _unflatten_dense_tensors
label_desc: Use image node_id 10 for calling a global function with example usage: reduce_tensor_dp_group(tensor, dtype, dst_local_rank, dst_global_rank, group) and returns: tensor