code: def test_allreduce_destroy(
      ray_start_distributed_2_nodes, backend, group_name="default"
  ):
      world_size = 8
      actors, _ = create_collective_workers(world_size, backend=backend)
  
      results = ray.get([a.do_allreduce.remote() for a in actors])
      assert (
          results[0] == np.ones((10,), dtype=np.float32) * world_size
      ).all()
      assert (
          results[1] == np.ones((10,), dtype=np.float32) * world_size
      ).all()
  
      # destroy the group and try do work, should fail
      ray.get([a.destroy_group.remote() for a in actors])
      with pytest.raises(RuntimeError):
          results = ray.get([a.do_allreduce.remote() for a in actors])
  
      # reinit the same group and all reduce
      ray.get(
          [
              actor.init_group.remote(
                  world_size, i, backend, group_name
              )
              for i, actor in enumerate(actors)
          ]
      )
      results = ray.get([a.do_allreduce.remote() for a in actors])
      assert (
          results[0]
          == np.ones((10,), dtype=np.float32) * world_size * world_size
      ).all()
      assert (
          results[1]
          == np.ones((10,), dtype=np.float32) * world_size * world_size
      ).all()
  
file_path: ray/python/ray/util/collective/tests/distributed_cpu_tests/test_distributed_allreduce.py
imports: import pytest
  import ray
  from ray.util.collective.types import ReduceOp
  import numpy
  import torch
  from ray.util.collective.types import Backend
  from ray.util.collective.tests.cpu_util import create_collective_workers
label_desc: Use image node_id 3 for calling a global function with example usage: test_allreduce_destroy(ray_start_distributed_2_nodes, backend, group_name) without return types