code: def stop_cluster(cls):
      logger.info("Stopping cluster %s.", cls._cluster_info)
      if not cls._cluster_info:  # pragma: no cover
          return
      pg_name = cls._cluster_info["pg_name"]
      pg = cls._cluster_info["pg_group"]
      for index, bundle_spec in enumerate(pg.bundle_specs):
          # Main pool took a process.
          # If supervisor is created in the same node with worker, it will take a process too.
          n_process = int(bundle_spec["CPU"]) + 2
          for process_index in reversed(range(n_process)):
              address = process_placement_to_address(
                  pg_name, index, process_index=process_index
              )
              try:
                  ray_actor = ray.get_actor(address)
                  if (
                      "COV_CORE_SOURCE" in os.environ
                  ):  # pragma: no cover
                      # must clean up first, or coverage info lost.
                      # must save the local reference until this is fixed:
                      # https://github.com/ray-project/ray/issues/7815
                      ray.get(ray_actor.cleanup.remote())
                  ray.kill(ray_actor, no_restart=True)
                  while True:
                      try:
                          ray.get(ray_actor.wait.remote(30))
                          logger.warning(
                              "Waiting actor %s to be killed.",
                              ray_actor,
                          )  # pragma: no cover
                      except ray.exceptions.RayActorError:
                          break
              except:  # noqa: E722  # nosec  # pylint: disable=bare-except
                  pass
      ray.util.remove_placement_group(pg)
      cls._cluster_info = dict()
      logger.info("Stopped cluster %s.", pg_name)
  
file_path: mars/mars/oscar/backends/ray/driver.py
imports: import copy
  import logging
  import os
  from numbers import Number
  from typing import Dict
  from ....utils import lazy_import
  from ...driver import BaseActorDriver
  from .utils import process_placement_to_address, addresses_to_placement_group_info
label_desc: Use image node_id 2 for calling the RayActorDriver obj's underlying member method code with example usage: obj.stop_cluster(cls) without return types