code: def test_send_forward_recv_forward(self):
      self._init_model_parallel()
      prev_tensor = None
      tensor = self.create_tensor(self.rank)
      if parallel_state.is_pipeline_first_stage():
          p2p_communication.send_forward(
              output_tensor=tensor,
              tensor_shape=self.shape,
              dtype=self.dtype,
          )
      elif parallel_state.is_pipeline_last_stage():
          prev_tensor = p2p_communication.recv_forward(
              tensor_shape=self.shape, dtype=self.dtype
          )
      else:
          prev_tensor = p2p_communication.send_forward_recv_forward(
              output_tensor=tensor,
              recv_prev=True,
              tensor_shape=self.shape,
              dtype=self.dtype,
          )
  
      if parallel_state.is_pipeline_first_stage():
          self.assertIsNone(prev_tensor)
      else:
          expected_prev_tensor = self.create_tensor(self.rank - 1)
          self.assertEqual(prev_tensor, expected_prev_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 5 for calling the P2PCommTestBase obj's underlying member method code with example usage: obj.test_send_forward_recv_forward() without return types