code: def _setup_ray_cluster(
      *,
      max_worker_nodes: int,
      min_worker_nodes: int,
      num_cpus_worker_node: int,
      num_cpus_head_node: int,
      num_gpus_worker_node: int,
      num_gpus_head_node: int,
      using_stage_scheduling: bool,
      heap_memory_worker_node: int,
      heap_memory_head_node: int,
      object_store_memory_worker_node: int,
      object_store_memory_head_node: int,
      head_node_options: Dict,
      worker_node_options: Dict,
      ray_temp_root_dir: str,
      collect_log_to_path: str,
      autoscale_upscaling_speed: float,
      autoscale_idle_timeout_minutes: float,
      is_global: bool,
  ) -> Type[RayClusterOnSpark]:
      """
      The public API `ray.util.spark.setup_ray_cluster` does some argument
      validation and then pass validated arguments to this interface.
      and it returns a `RayClusterOnSpark` instance.
  
      The returned instance can be used to connect to, disconnect from and shutdown the
      ray cluster. This instance can also be used as a context manager (used by
      encapsulating operations within `with _setup_ray_cluster(...):`). Upon entering the
      managed scope, the ray cluster is initiated and connected to. When exiting the
      scope, the ray cluster is disconnected and shut down.
  
      Note: This function interface is stable and can be used for
      instrumentation logging patching.
      """
      from pyspark.util import inheritable_thread_target
      import fcntl
  
      start_hook = _create_hook_entry(is_global)
      spark = get_spark_session()
  
      ray_head_ip = socket.gethostbyname(
          get_spark_application_driver_host(spark)
      )
      ray_head_port = get_random_unused_port(
          ray_head_ip, min_port=9000, max_port=10000
      )
      port_exclude_list = [ray_head_port]
  
      # Make a copy for head_node_options to avoid changing original dict in user code.
      head_node_options = head_node_options.copy()
      include_dashboard = head_node_options.pop(
          "include_dashboard", None
      )
      ray_dashboard_port = head_node_options.pop("dashboard_port", None)
  
      if is_global:
          ray_client_server_port = 10001
      else:
          ray_client_server_port = get_random_unused_port(
              ray_head_ip,
              min_port=9000,
              max_port=10000,
              exclude_list=port_exclude_list,
          )
  
      port_exclude_list.append(ray_client_server_port)
  
      autoscale = min_worker_nodes < max_worker_nodes
      if autoscale:
          spark_job_server_port = get_random_unused_port(
              ray_head_ip,
              min_port=9000,
              max_port=10000,
              exclude_list=port_exclude_list,
          )
          port_exclude_list.append(spark_job_server_port)
      else:
          spark_job_server_port = None
  
      if include_dashboard is None or include_dashboard is True:
          if ray_dashboard_port is None:
              ray_dashboard_port = get_random_unused_port(
                  ray_head_ip,
                  min_port=9000,
                  max_port=10000,
                  exclude_list=port_exclude_list,
              )
              port_exclude_list.append(ray_dashboard_port)
          ray_dashboard_agent_port = get_random_unused_port(
              ray_head_ip,
              min_port=9000,
              max_port=10000,
              exclude_list=port_exclude_list,
          )
          port_exclude_list.append(ray_dashboard_agent_port)
  
          dashboard_options = [
              "--dashboard-host=0.0.0.0",
              f"--dashboard-port={ray_dashboard_port}",
              f"--dashboard-agent-listen-port={ray_dashboard_agent_port}",
          ]
          # If include_dashboard is None, we don't set `--include-dashboard` option,
          # in this case Ray will decide whether dashboard can be started
          # (e.g. checking any missing dependencies).
          if include_dashboard is True:
              dashboard_options += ["--include-dashboard=true"]
      else:
          dashboard_options = [
              "--include-dashboard=false",
          ]
  
      _logger.info(
          f"Ray head hostname: {ray_head_ip}, port: {ray_head_port}, "
          f"ray client server port: {ray_client_server_port}."
      )
  
      cluster_unique_id = uuid.uuid4().hex[:8]
  
      if is_global:
          # global mode enabled
          # for global mode, Ray always uses default temp dir
          # so that local Ray client can discover it without specifying
          # head node address.
          if ray_temp_root_dir is not None:
              raise ValueError(
                  "Ray on spark global mode cluster does not allow you to set "
                  "'ray_temp_root_dir' argument."
              )
  
          # We only allow user to launch one active Ray on spark global cluster
          # at a time. So acquiring a global file lock before setting up a new
          # Ray on spark global cluster.
          global_cluster_lock_fd = os.open(
              "/tmp/ray_on_spark_global_cluster.lock",
              os.O_RDWR | os.O_CREAT | os.O_TRUNC,
          )
  
          try:
              # acquiring exclusive lock to ensure copy logs and removing dir safely.
              fcntl.flock(
                  global_cluster_lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB
              )
          except BlockingIOError:
              # acquiring global lock failed.
              raise ValueError(
                  "Acquiring global lock failed for setting up new global mode Ray on "
                  "spark cluster. If there is an active global mode Ray on spark "
                  "cluster, please shut down it before you create a new one."
              )
  
          ray_temp_dir = None
          ray_default_tmp_dir = _get_default_ray_tmp_dir()
          os.makedirs(ray_default_tmp_dir, exist_ok=True)
          object_spilling_dir = os.path.join(
              ray_default_tmp_dir, "spill"
          )
      else:
          global_cluster_lock_fd = None
          if ray_temp_root_dir is None:
              ray_temp_root_dir = start_hook.get_default_temp_root_dir()
          ray_temp_dir = os.path.join(
              ray_temp_root_dir,
              f"ray-{ray_head_port}-{cluster_unique_id}",
          )
          os.makedirs(ray_temp_dir, exist_ok=True)
          object_spilling_dir = os.path.join(ray_temp_dir, "spill")
  
      os.makedirs(object_spilling_dir, exist_ok=True)
  
      head_node_options = _append_default_spilling_dir_config(
          head_node_options, object_spilling_dir
      )
  
      if autoscale:
          from ray.autoscaler._private.spark.spark_job_server import (
              _start_spark_job_server,
          )
  
          spark_job_server = _start_spark_job_server(
              ray_head_ip, spark_job_server_port, spark
          )
          autoscaling_cluster = AutoscalingCluster(
              head_resources={
                  "CPU": num_cpus_head_node,
                  "GPU": num_gpus_head_node,
                  "memory": heap_memory_head_node,
                  "object_store_memory": object_store_memory_head_node,
              },
              worker_node_types={
                  "ray.worker": {
                      "resources": {
                          "CPU": num_cpus_worker_node,
                          "GPU": num_gpus_worker_node,
                          "memory": heap_memory_worker_node,
                          "object_store_memory": object_store_memory_worker_node,
                      },
                      "node_config": {},
                      "min_workers": min_worker_nodes,
                      "max_workers": max_worker_nodes,
                  },
              },
              extra_provider_config={
                  "ray_head_ip": ray_head_ip,
                  "ray_head_port": ray_head_port,
                  "cluster_unique_id": cluster_unique_id,
                  "using_stage_scheduling": using_stage_scheduling,
                  "ray_temp_dir": ray_temp_dir,
                  "worker_node_options": worker_node_options,
                  "collect_log_to_path": collect_log_to_path,
                  "spark_job_server_port": spark_job_server_port,
              },
              upscaling_speed=autoscale_upscaling_speed,
              idle_timeout_minutes=autoscale_idle_timeout_minutes,
          )
          ray_head_proc, tail_output_deque = autoscaling_cluster.start(
              ray_head_ip,
              ray_head_port,
              ray_client_server_port,
              ray_temp_dir,
              dashboard_options,
              head_node_options,
              collect_log_to_path,
          )
          ray_head_node_cmd = autoscaling_cluster.ray_head_node_cmd
      else:
          (
              worker_port_range_begin,
              worker_port_range_end,
          ) = _preallocate_ray_worker_port_range()
  
          ray_head_node_cmd = [
              sys.executable,
              "-m",
              "ray.util.spark.start_ray_node",
              "--block",
              "--head",
              f"--node-ip-address={ray_head_ip}",
              f"--port={ray_head_port}",
              f"--ray-client-server-port={ray_client_server_port}",
              f"--num-cpus={num_cpus_head_node}",
              f"--num-gpus={num_gpus_head_node}",
              f"--memory={heap_memory_head_node}",
              f"--object-store-memory={object_store_memory_head_node}",
              f"--min-worker-port={worker_port_range_begin}",
              f"--max-worker-port={worker_port_range_end - 1}",
              *dashboard_options,
              *_convert_ray_node_options(head_node_options),
          ]
          if ray_temp_dir is not None:
              ray_head_node_cmd.append(f"--temp-dir={ray_temp_dir}")
  
          _logger.info(
              f"Starting Ray head, command: {' '.join(ray_head_node_cmd)}"
          )
  
          ray_head_proc, tail_output_deque = exec_cmd(
              ray_head_node_cmd,
              synchronous=False,
              extra_env={
                  RAY_ON_SPARK_COLLECT_LOG_TO_PATH: collect_log_to_path
                  or "",
                  RAY_ON_SPARK_START_RAY_PARENT_PID: str(os.getpid()),
                  **start_hook.custom_environment_variables(),
              },
          )
          spark_job_server = None
  
      # wait ray head node spin up.
      time.sleep(_RAY_HEAD_STARTUP_TIMEOUT)
  
      if not is_port_in_use(ray_head_ip, ray_head_port):
          if ray_head_proc.poll() is None:
              # Ray head GCS service is down. Kill ray head node.
              ray_head_proc.terminate()
              # wait killing complete.
              time.sleep(0.5)
  
          cmd_exec_failure_msg = gen_cmd_exec_failure_msg(
              ray_head_node_cmd,
              ray_head_proc.returncode,
              tail_output_deque,
          )
          raise RuntimeError(
              "Start Ray head node failed!\n" + cmd_exec_failure_msg
          )
  
      _logger.info("Ray head node started.")
  
      cluster_address = f"{ray_head_ip}:{ray_head_port}"
      # Set RAY_ADDRESS environment variable to the cluster address.
      os.environ["RAY_ADDRESS"] = cluster_address
  
      ray_cluster_handler = RayClusterOnSpark(
          autoscale=autoscale,
          address=cluster_address,
          head_proc=ray_head_proc,
          spark_job_group_id=None,
          min_worker_nodes=min_worker_nodes,
          max_worker_nodes=max_worker_nodes,
          temp_dir=ray_temp_dir,
          cluster_unique_id=cluster_unique_id,
          start_hook=start_hook,
          ray_dashboard_port=ray_dashboard_port,
          spark_job_server=spark_job_server,
          global_cluster_lock_fd=global_cluster_lock_fd,
          ray_client_server_port=ray_client_server_port,
      )
  
      if not autoscale:
          spark_job_group_id = (
              f"ray-cluster-{ray_head_port}-{cluster_unique_id}"
          )
          ray_cluster_handler.spark_job_group_id = spark_job_group_id
  
          def background_job_thread_fn():
              try:
                  _start_ray_worker_nodes(
                      spark=spark,
                      spark_job_group_id=spark_job_group_id,
                      spark_job_group_desc=(
                          "This job group is for spark job which runs the Ray cluster "
                          f"with ray head node {ray_head_ip}:{ray_head_port}"
                      ),
                      num_worker_nodes=max_worker_nodes,
                      using_stage_scheduling=using_stage_scheduling,
                      ray_head_ip=ray_head_ip,
                      ray_head_port=ray_head_port,
                      ray_temp_dir=ray_temp_dir,
                      num_cpus_per_node=num_cpus_worker_node,
                      num_gpus_per_node=num_gpus_worker_node,
                      heap_memory_per_node=heap_memory_worker_node,
                      object_store_memory_per_node=object_store_memory_worker_node,
                      worker_node_options=worker_node_options,
                      collect_log_to_path=collect_log_to_path,
                      autoscale_mode=False,
                      spark_job_server_port=spark_job_server_port,
                  )
              except Exception as e:
                  # NB:
                  # The background spark job is designed to running forever until it is
                  # killed, The exception might be raised in following cases:
                  #  1. The background job raises unexpected exception (i.e. ray cluster
                  #     dies unexpectedly)
                  #  2. User explicitly orders shutting down the ray cluster.
                  #  3. On Databricks runtime, when a notebook is detached, it triggers
                  #     python REPL `onCancel` event, cancelling the background running
                  #     spark job.
                  #  For case 1 and 3, only ray workers are killed, but driver side ray
                  #  head might still be running and the ray context might be in
                  #  connected status.
                  #  In order to disconnect and kill the ray head node, a call to
                  #  `ray_cluster_handler.shutdown()` is performed.
                  if not ray_cluster_handler.spark_job_is_canceled:
                      # Set `background_job_exception` attribute before calling
                      # `shutdown` so inside `shutdown` we can get exception information
                      # easily.
                      ray_cluster_handler.background_job_exception = e
                      ray_cluster_handler.shutdown(
                          cancel_background_job=False
                      )
  
          try:
              threading.Thread(
                  target=inheritable_thread_target(
                      background_job_thread_fn
                  ),
                  args=(),
              ).start()
  
              # Call hook immediately after spark job started.
              start_hook.on_cluster_created(ray_cluster_handler)
  
              # wait background spark task starting.
              for _ in range(_BACKGROUND_JOB_STARTUP_WAIT):
                  time.sleep(1)
                  if (
                      ray_cluster_handler.background_job_exception
                      is not None
                  ):
                      raise RuntimeError(
                          "Ray workers failed to start."
                      ) from ray_cluster_handler.background_job_exception
  
          except Exception:
              # If driver side setup ray-cluster routine raises exception, it might
              # result in part of ray processes has been launched (e.g. ray head or
              # some ray workers have been launched), calling
              # `ray_cluster_handler.shutdown()` to kill them and clean status.
              ray_cluster_handler.shutdown()
              raise
  
      return ray_cluster_handler
  
imports: import copy
  import yaml
  import json
  import os
  import socket
  import sys
  import time
  import threading
  import logging
  import uuid
  import warnings
  import requests
  from packaging.version import Version
  from typing import Optional, Dict, Tuple, Type
  import ray
  import ray._private.services
  from ray.autoscaler._private.spark.node_provider import HEAD_NODE_ID
  from ray.util.annotations import DeveloperAPI, PublicAPI
  from ray._private.storage import _load_class
  from .utils import exec_cmd, is_port_in_use, get_random_unused_port, get_spark_session, get_spark_application_driver_host, is_in_databricks_runtime, get_spark_task_assigned_physical_gpus, get_avail_mem_per_ray_worker_node, get_max_num_concurrent_tasks, gen_cmd_exec_failure_msg, calc_mem_ray_head_node, _wait_service_up
  from .start_hook_base import RayOnSparkStartHook
  from .databricks_hook import DefaultDatabricksRayOnSparkStartHook
  from threading import Event
label_desc: Use image node_id 9 for calling a global function with example usage: _setup_ray_cluster() and returns: ray_cluster_handler