code: def _ray_large_cluster(request):  # pragma: no cover
      param = getattr(request, "param", {})
      num_nodes = param.get("num_nodes", 3)
      num_cpus = param.get("num_cpus", 16)
      from ray.cluster_utils import Cluster
  
      cluster = Cluster()
      remote_nodes = []
      for i in range(num_nodes):
          remote_nodes.append(
              cluster.add_node(
                  num_cpus=num_cpus, memory=num_cpus * 2 * 1024**3
              )
          )
          if len(remote_nodes) == 1:
              try:
                  job_config = ray.job_config.JobConfig(
                      total_memory_mb=num_nodes * 32 * 1024**3
                  )
              except TypeError:
                  job_config = None
              ray.init(address=cluster.address, job_config=job_config)
      try:
          yield cluster
      finally:
          Router.set_instance(None)
          RayServer.clear()
          ray.shutdown()
          cluster.shutdown()
          if "COV_CORE_SOURCE" in os.environ:
              # Remove this when https://github.com/ray-project/ray/issues/16802 got fixed
              subprocess.check_call(["ray", "stop", "--force"])
  
imports: import concurrent.futures
  import os
  import subprocess
  import psutil
  import pytest
  from mars.config import option_context
  from mars.core.mode import is_kernel_mode, is_build_mode
  from mars.lib.aio.lru import clear_all_alru_caches
  from mars.oscar.backends.router import Router
  from mars.oscar.backends.ray.communication import RayServer
  from mars.utils import lazy_import
label_desc: Use image node_id 9 for calling a global function with example usage: _ray_large_cluster(request) without return types