code: def test_sendrecv(
      ray_start_distributed_2_nodes,
      group_name,
      array_size,
      src_rank,
      dst_rank,
      backend,
  ):
      if src_rank == dst_rank:
          return
      world_size = 8
      actors, _ = create_collective_workers(
          num_workers=world_size, group_name=group_name, backend=backend
      )
      ray.get(
          [
              a.set_buffer.remote(
                  np.ones(array_size, dtype=np.float32) * (i + 1)
              )
              for i, a in enumerate(actors)
          ]
      )
      refs = []
      for i in range(world_size):
          refs.append(actors[i].get_buffer.remote())
      refs[src_rank] = actors[src_rank].do_send.remote(
          group_name, dst_rank
      )
      refs[dst_rank] = actors[dst_rank].do_recv.remote(
          group_name, src_rank
      )
      results = ray.get(refs)
      assert (
          results[src_rank]
          == np.ones(array_size, dtype=np.float32) * (src_rank + 1)
      ).all()
      assert (
          results[dst_rank]
          == np.ones(array_size, dtype=np.float32) * (src_rank + 1)
      ).all()
      ray.get([a.destroy_group.remote(group_name) for a in actors])
  
imports: import numpy
  import pytest
  import ray
  from ray.util.collective.types import Backend
  from ray.util.collective.tests.cpu_util import create_collective_workers
label_desc: Use image node_id 1 for calling a global function with example usage: test_sendrecv(ray_start_distributed_2_nodes, group_name, array_size, src_rank, dst_rank, backend) without return types