code: def _all_to_all(
      x: torch.Tensor,
      pg: ProcessGroup,
      scatter_dim: int,
      gather_dim: int,
  ) -> torch.Tensor:
      world_size = pg.tp_world_size()
      if world_size == 1:
          return x
  
      # TODO: enabling mpi backend to support CPU all_to_all
      assert (
          x.device.type == "cuda"
      ), f"Currently, the collective function dual_all_to_all only supports nccl backend"
  
      shapes = list(x.size())
      shapes[scatter_dim] = shapes[scatter_dim] // world_size
  
      scatter_list = [
          each.contiguous()
          for each in torch.tensor_split(x, world_size, scatter_dim)
      ]
      gather_list = [
          torch.empty(*shapes, dtype=x.dtype, device=x.device)
          for _ in range(world_size)
      ]
      torch.distributed.all_to_all(
          gather_list, scatter_list, group=pg.tp_process_group()
      )
  
      return torch.cat(gather_list, dim=gather_dim).contiguous()
  
imports: from typing import List, Optional, Union
  import torch
  import torch.distributed
  from colossalai.legacy.global_variables import tensor_parallel_env
  from colossalai.legacy.nn.layer.utils import divide
  from colossalai.legacy.tensor import ColoTensorSpec, ProcessGroup
  from colossalai.tensor import ColoTensor
label_desc: Use image node_id 13 for calling a global function with example usage: _all_to_all(x, pg, scatter_dim, gather_dim) and returns: torch, x