class_docstr: A training engine hybrid pipeline, data, and model parallel training.
  
  This engine is created by ``deepspeed.initialize()`` when a :class:`PipelineModule`
  is provided.
docstr: Communicate metadata about upcoming p2p transfers.
  
  Metadata is communicated in this order:
      * type (0: tensor, 1: list)
      * num_tensors if type=list
      foreach tensor in buffer:
          * ndims
          * shape
docstr_tok: ["Communicate","metadata","about","upcoming","p2p","transfers",".","Metadata","is","communicated","in","this","order",":","*","type","(","0",":","tensor",",","1",":","list",")","*","num_tensors","if","type=list","foreach","tensor","in","buffer",":","*","ndims","*","shape"]
code: def _send_tensor_meta(self, buffer, recv_stage):
      """Communicate metadata about upcoming p2p transfers.
  
      Metadata is communicated in this order:
          * type (0: tensor, 1: list)
          * num_tensors if type=list
          foreach tensor in buffer:
              * ndims
              * shape
      """
      send_bytes = 0
      if isinstance(buffer, torch.Tensor):
          type_tensor = torch.LongTensor(data=[0]).to(self.device)
          p2p.send(type_tensor, recv_stage)
          send_shape = torch.LongTensor(data=buffer.size()).to(
              self.device
          )
          send_ndims = torch.LongTensor(data=[len(buffer.size())]).to(
              self.device
          )
          p2p.send(send_ndims, recv_stage)
          p2p.send(send_shape, recv_stage)
          send_bytes += _tensor_bytes(buffer)
      elif isinstance(buffer, list):
          assert False
          type_tensor = torch.LongTensor(data=[1]).to(self.device)
          p2p.send(type_tensor, recv_stage)
          count_tensor = torch.LongTensor(data=[len(buffer)]).to(
              self.device
          )
          p2p.send(count_tensor, recv_stage)
          for tensor in buffer:
              assert isinstance(tensor, torch.Tensor)
              send_shape = torch.LongTensor(data=tensor.size()).to(
                  self.device
              )
              send_ndims = torch.LongTensor(
                  data=[len(tensor.size())]
              ).to(self.device)
              p2p.send(send_ndims, recv_stage)
              p2p.send(send_shape, recv_stage)
              send_bytes += _tensor_bytes(tensor)
      elif isinstance(buffer, tuple):
          type_tensor = torch.LongTensor(data=[2]).to(self.device)
          p2p.send(type_tensor, recv_stage)
          count_tensor = torch.LongTensor(data=[len(buffer)]).to(
              self.device
          )
          p2p.send(count_tensor, recv_stage)
          for idx, tensor in enumerate(buffer):
              assert isinstance(tensor, torch.Tensor)
              send_shape = torch.LongTensor(data=tensor.size()).to(
                  self.device
              )
              send_ndims = torch.LongTensor(
                  data=[len(tensor.size())]
              ).to(self.device)
              send_dtype = torch.LongTensor(
                  data=[self.DTYPE_TO_ID[tensor.dtype]]
              ).to(self.device)
              p2p.send(send_dtype, recv_stage)
              p2p.send(send_ndims, recv_stage)
              p2p.send(send_shape, recv_stage)
              # Useful for performance debugging.
              """
                  new_bytes = _tensor_bytes(tensor)
                  send_bytes += _tensor_bytes(tensor)
                  # Useful for performance debugging.
                  if self.grid.data_parallel_id == 0:
                      print(
                          f'STAGE={self.stage_id} pipe-send-volume[{idx}]: shape={send_shape} {new_bytes/1024**2:0.2f}MB'
                      )
                  """
      else:
          raise NotImplementedError(
              f"Could not send meta type {type(buffer)}"
          )
  
      # Useful for performance debugging.
      """
          if self.grid.data_parallel_id == 0:
              print(f'STAGE={self.stage_id} pipe-send-volume: {send_bytes/1024**2:0.2f}MB')
          """
  
imports: from types import MethodType
  import torch
  from deepspeed import comm
  from deepspeed.utils import logger
  from deepspeed.utils.timer import ThroughputTimer
  from ..engine import DeepSpeedEngine, MEMORY_OPT_ALLREDUCE_SIZE
  from ..utils import PartitionedTensor
  from ..dataloader import RepeatingLoader
  from .module import PipelineModule, PipelineError
  from .None import p2p
  from .None import schedule
label_desc: Use image node_id 27 for calling the PipelineEngine obj's underlying member method code with example usage: obj._send_tensor_meta(buffer, recv_stage) without return types