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: Receive metadata about upcoming p2p transfers and return allocated buffers.
  
  Metadata is communicated in this order:
      * type (0: tensor, 1: list)
      * num_tensors if type=list
      foreach tensor in buffer:
          * ndims
          * shape
  
  Returns:
      Allocated buffer for receiving from send_stage.
docstr_tok: ["Receive","metadata","about","upcoming","p2p","transfers","and","return","allocated","buffers",".","Metadata","is","communicated","in","this","order",":","*","type","(","0",":","tensor",",","1",":","list",")","*","num_tensors","if","type=list","foreach","tensor","in","buffer",":","*","ndims","*","shape","Returns",":","Allocated","buffer","for","receiving","from","send_stage","."]
code: def _recv_tensor_meta(self, send_stage):
      """Receive metadata about upcoming p2p transfers and return allocated buffers.
  
      Metadata is communicated in this order:
          * type (0: tensor, 1: list)
          * num_tensors if type=list
          foreach tensor in buffer:
              * ndims
              * shape
  
      Returns:
          Allocated buffer for receiving from send_stage.
      """
  
      type_tensor = torch.LongTensor(data=[0]).to(self.device)
      p2p.recv(type_tensor, send_stage)
      recv_type = type_tensor.item()
  
      # A single tensor will be sent.
      if recv_type == 0:
          recv_ndims = torch.LongTensor(data=[0]).to(self.device)
          p2p.recv(recv_ndims, send_stage)
          recv_ndims = recv_ndims.item()
          recv_shape = torch.LongTensor([1] * recv_ndims).to(
              self.device
          )
          p2p.recv(recv_shape, send_stage)
          recv_shape = recv_shape.tolist()
          return self._allocate_buffer(recv_shape, num_buffers=1)[0]
  
      # List or tuple of tensors
      elif recv_type == 1 or recv_type == 2:
          count_tensor = torch.LongTensor(data=[0]).to(self.device)
          p2p.recv(count_tensor, send_stage)
          num_tensors = count_tensor.item()
          recv_shapes_and_dtypes = []
          for idx in range(num_tensors):
              recv_dtype = torch.LongTensor(data=[0]).to(self.device)
              p2p.recv(recv_dtype, send_stage)
              recv_dtype = self.ID_TO_DTYPE[recv_dtype.item()]
              recv_ndims = torch.LongTensor(data=[0]).to(self.device)
              p2p.recv(recv_ndims, send_stage)
              recv_ndims = recv_ndims.item()
              recv_shape = torch.LongTensor([1] * recv_ndims).to(
                  self.device
              )
              p2p.recv(recv_shape, send_stage)
              recv_shapes_and_dtypes.append(
                  (recv_shape.tolist(), recv_dtype)
              )
  
          buffers = self._allocate_buffers(
              recv_shapes_and_dtypes, num_buffers=1
          )[0]
          # Convert to tuples if requested.
          if recv_type == 2:
              buffers = tuple(buffers)
          return buffers
  
      else:
          raise NotImplementedError(
              f"Could not receive type {type(recv_type)}"
          )
  
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 28 for calling the PipelineEngine obj's underlying member method code with example usage: obj._recv_tensor_meta(send_stage) and returns: self, buffers