code: def test_broadcast_torch_cupy(
      ray_start_distributed_multigpu_2_nodes_4_gpus,
      src_rank,
      src_gpu_index,
  ):
      import torch
  
      world_size = 2
      num_gpu_per_worker = 2
      actors, _ = create_collective_multigpu_workers(world_size)
      ray.get(actors[0].set_buffer.remote([10], value0=2, value1=3))
      ray.get(
          actors[1].set_buffer.remote(
              [10],
              value0=4,
              value1=5,
              tensor_type0="torch",
              tensor_type1="torch",
          )
      )
      results = ray.get(
          [
              a.do_broadcast_multigpu.remote(
                  src_rank=src_rank, src_gpu_index=src_gpu_index
              )
              for a in actors
          ]
      )
      for i in range(world_size):
          for j in range(num_gpu_per_worker):
              val = (src_rank + 1) * 2 + src_gpu_index
              if i == 0:
                  assert (
                      results[i][j]
                      == cp.ones([10], dtype=cp.float32) * val
                  ).all()
              else:
                  assert (
                      results[i][j] == torch.ones([10]).cuda(j) * val
                  ).all()
  
file_path: ray/python/ray/util/collective/tests/distributed_multigpu_tests/test_distributed_multigpu_broadcast.py
imports: import pytest
  import cupy
  import ray
  from ray.util.collective.tests.util import create_collective_multigpu_workers
label_desc: Use image node_id 3 for calling a global function with example usage: test_broadcast_torch_cupy(ray_start_distributed_multigpu_2_nodes_4_gpus, src_rank, src_gpu_index) without return types