code: def test_send_backward_recv_backward(self):
      self._init_model_parallel()
      tensor = self.create_tensor(self.rank)
  
      next_tensor = None
      if parallel_state.is_pipeline_first_stage():
          next_tensor = p2p_communication.recv_backward(
              tensor_shape=self.shape, dtype=self.dtype
          )
      elif parallel_state.is_pipeline_last_stage():
          p2p_communication.send_backward(
              input_tensor_grad=tensor,
              tensor_shape=self.shape,
              dtype=self.dtype,
          )
      else:
          next_tensor = p2p_communication.send_backward_recv_backward(
              input_tensor_grad=tensor,
              recv_next=True,
              tensor_shape=self.shape,
              dtype=self.dtype,
          )
  
      if parallel_state.is_pipeline_last_stage():
          self.assertIsNone(next_tensor)
      else:
          expected_next_tensor = self.create_tensor(self.rank + 1)
          self.assertEqual(next_tensor, expected_next_tensor)
  
file_path: apex/tests/L0/run_transformer/test_p2p_comm.py
imports: import logging
  import unittest
  import torch
  from torch.testing._internal import common_utils
  from apex.transformer import parallel_state
  from apex.transformer.pipeline_parallel import p2p_communication
  from apex.transformer.testing.distributed_test_base import NcclDistributedTestBase
  from apex.transformer.testing.distributed_test_base import UccDistributedTestBase
label_desc: Use image node_id 6 for calling the P2PCommTestBase obj's underlying member method code with example usage: obj.test_send_backward_recv_backward() without return types