code: def test_allreduce_different_op(ray_start_single_node_2_gpus):
      world_size = 2
      actors, _ = create_collective_workers(world_size)
  
      # check product
      ray.wait(
          [
              a.set_buffer.remote(
                  cp.ones(10, dtype=cp.float32) * (i + 2)
              )
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [a.do_allreduce.remote(op=ReduceOp.PRODUCT) for a in actors]
      )
      assert (results[0] == cp.ones((10,), dtype=cp.float32) * 6).all()
      assert (results[1] == cp.ones((10,), dtype=cp.float32) * 6).all()
  
      # check min
      ray.wait(
          [
              a.set_buffer.remote(
                  cp.ones(10, dtype=cp.float32) * (i + 2)
              )
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [a.do_allreduce.remote(op=ReduceOp.MIN) for a in actors]
      )
      assert (results[0] == cp.ones((10,), dtype=cp.float32) * 2).all()
      assert (results[1] == cp.ones((10,), dtype=cp.float32) * 2).all()
  
      # check max
      ray.wait(
          [
              a.set_buffer.remote(
                  cp.ones(10, dtype=cp.float32) * (i + 2)
              )
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [a.do_allreduce.remote(op=ReduceOp.MAX) for a in actors]
      )
      assert (results[0] == cp.ones((10,), dtype=cp.float32) * 3).all()
      assert (results[1] == cp.ones((10,), dtype=cp.float32) * 3).all()
  
file_path: ray/python/ray/util/collective/tests/single_node_gpu_tests/test_allreduce.py
imports: import cupy
  import pytest
  import ray
  import torch
  from ray.util.collective.tests.util import create_collective_workers
  from ray.util.collective.types import ReduceOp
label_desc: Use image node_id 5 for calling a global function with example usage: test_allreduce_different_op(ray_start_single_node_2_gpus) without return types