code: def test_sendrecv(
      self,
      X,
      device_option,
      specify_send_blob,
      specify_recv_blob,
      gc,
      dc,
  ):
      # Use mpi4py's broadcast to make sure that all nodes inherit the
      # same hypothesis test.
      X = COMM.bcast(X)
      device_option = COMM.bcast(device_option)
      specify_send_blob = COMM.bcast(specify_send_blob)
      specify_recv_blob = COMM.bcast(specify_recv_blob)
      X[:] = RANK
  
      self.assertTrue(
          workspace.RunOperatorOnce(
              core.CreateOperator(
                  "CreateCommonWorld",
                  [],
                  "comm",
                  engine="MPI",
                  device_option=device_option,
              )
          )
      )
      self.assertTrue(workspace.FeedBlob("X", X, device_option))
      for src in range(SIZE):
          for dst in range(SIZE):
              tag = src * SIZE + dst
              if src == dst:
                  continue
              elif RANK == src:
                  X[:] = RANK
                  self.assertTrue(
                      workspace.FeedBlob("X", X, device_option)
                  )
                  if specify_send_blob:
                      self.assertTrue(
                          workspace.FeedBlob(
                              "dst", np.array(dst, dtype=np.int32)
                          )
                      )
                      self.assertTrue(
                          workspace.FeedBlob(
                              "tag", np.array(tag, dtype=np.int32)
                          )
                      )
                      mpi_op = core.CreateOperator(
                          "SendTensor",
                          ["comm", "X", "dst", "tag"],
                          [],
                          engine="MPI",
                          raw_buffer=True,
                          device_option=device_option,
                      )
                  else:
                      mpi_op = core.CreateOperator(
                          "SendTensor",
                          ["comm", "X"],
                          [],
                          engine="MPI",
                          dst=dst,
                          tag=tag,
                          raw_buffer=True,
                          device_option=device_option,
                      )
                  self.assertTrue(workspace.RunOperatorOnce(mpi_op))
              elif RANK == dst:
                  if specify_recv_blob:
                      self.assertTrue(
                          workspace.FeedBlob(
                              "src", np.array(src, dtype=np.int32)
                          )
                      )
                      self.assertTrue(
                          workspace.FeedBlob(
                              "tag", np.array(tag, dtype=np.int32)
                          )
                      )
                      mpi_op = core.CreateOperator(
                          "ReceiveTensor",
                          ["comm", "X", "src", "tag"],
                          ["X", "src", "tag"],
                          engine="MPI",
                          src=src,
                          tag=tag,
                          raw_buffer=True,
                          device_option=device_option,
                      )
                  else:
                      mpi_op = core.CreateOperator(
                          "ReceiveTensor",
                          ["comm", "X"],
                          ["X", "src", "tag"],
                          engine="MPI",
                          src=src,
                          tag=tag,
                          raw_buffer=True,
                          device_option=device_option,
                      )
                  self.assertTrue(workspace.RunOperatorOnce(mpi_op))
                  received = workspace.FetchBlob("X")
                  np.testing.assert_array_equal(received, src)
                  src_blob = workspace.FetchBlob("src")
                  np.testing.assert_array_equal(src_blob, src)
                  tag_blob = workspace.FetchBlob("tag")
                  np.testing.assert_array_equal(tag_blob, tag)
              # simply wait for the guys to finish
              COMM.barrier()
      workspace.ResetWorkspace()
  
file_path: pytorch/caffe2/python/operator_test/mpi_test.py
imports: from hypothesis import given
  import hypothesis.strategies
  import numpy
  import unittest
  from caffe2.python import core, workspace, dyndep
  import caffe2.python.hypothesis_test_util
label_desc: Use image node_id 4 for calling the TestMPI obj's underlying member method code with example usage: obj.test_sendrecv(X, device_option, specify_send_blob, specify_recv_blob, gc, dc) without return types