code: def test_reduce_different_op(
      ray_start_single_node, dst_rank, backend
  ):
      world_size = 2
      actors, _ = create_collective_workers(world_size, backend=backend)
  
      # check product
      ray.wait(
          [
              a.set_buffer.remote(
                  np.ones(10, dtype=np.float32) * (i + 2)
              )
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [
              a.do_reduce.remote(dst_rank=dst_rank, op=ReduceOp.PRODUCT)
              for a in actors
          ]
      )
      for i in range(world_size):
          if i == dst_rank:
              assert (
                  results[i] == np.ones((10,), dtype=np.float32) * 6
              ).all()
          else:
              assert (
                  results[i]
                  == np.ones((10,), dtype=np.float32) * (i + 2)
              ).all()
  
      # check min
      ray.wait(
          [
              a.set_buffer.remote(
                  np.ones(10, dtype=np.float32) * (i + 2)
              )
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [
              a.do_reduce.remote(dst_rank=dst_rank, op=ReduceOp.MIN)
              for a in actors
          ]
      )
      for i in range(world_size):
          if i == dst_rank:
              assert (
                  results[i] == np.ones((10,), dtype=np.float32) * 2
              ).all()
          else:
              assert (
                  results[i]
                  == np.ones((10,), dtype=np.float32) * (i + 2)
              ).all()
  
      # check max
      ray.wait(
          [
              a.set_buffer.remote(
                  np.ones(10, dtype=np.float32) * (i + 2)
              )
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [
              a.do_reduce.remote(dst_rank=dst_rank, op=ReduceOp.MAX)
              for a in actors
          ]
      )
      for i in range(world_size):
          if i == dst_rank:
              assert (
                  results[i] == np.ones((10,), dtype=np.float32) * 3
              ).all()
          else:
              assert (
                  results[i]
                  == np.ones((10,), dtype=np.float32) * (i + 2)
              ).all()
  
file_path: ray/python/ray/util/collective/tests/single_node_cpu_tests/test_reduce.py
imports: import pytest
  import numpy
  import ray
  from ray.util.collective.types import Backend, ReduceOp
  from ray.util.collective.tests.cpu_util import create_collective_workers
label_desc: Use image node_id 4 for calling a global function with example usage: test_reduce_different_op(ray_start_single_node, dst_rank, backend) without return types