code: def _setup_ray_cluster_internal(
      max_worker_nodes: int,
      min_worker_nodes: Optional[int],
      num_cpus_worker_node: Optional[int],
      num_cpus_head_node: Optional[int],
      num_gpus_worker_node: Optional[int],
      num_gpus_head_node: Optional[int],
      object_store_memory_worker_node: Optional[int],
      object_store_memory_head_node: Optional[int],
      head_node_options: Optional[Dict],
      worker_node_options: Optional[Dict],
      ray_temp_root_dir: Optional[str],
      strict_mode: bool,
      collect_log_to_path: Optional[str],
      autoscale_upscaling_speed: Optional[float],
      autoscale_idle_timeout_minutes: Optional[float],
      is_global: bool,
      **kwargs,
  ) -> Tuple[str, str]:
      global _active_ray_cluster
  
      _check_system_environment()
  
      head_node_options = head_node_options or {}
      worker_node_options = worker_node_options or {}
  
      _verify_node_options(
          head_node_options,
          _head_node_option_block_keys,
          "Ray head node on spark",
      )
      _verify_node_options(
          worker_node_options,
          _worker_node_option_block_keys,
          "Ray worker node on spark",
      )
  
      if _active_ray_cluster is not None:
          raise RuntimeError(
              "Current active ray cluster on spark haven't shut down. Please call "
              "`ray.util.spark.shutdown_ray_cluster()` before initiating a new Ray "
              "cluster on spark."
          )
  
      if ray.is_initialized():
          raise RuntimeError(
              "Current python process already initialized Ray, Please shut down it "
              "by `ray.shutdown()` before initiating a Ray cluster on spark."
          )
  
      spark = get_spark_session()
  
      spark_master = spark.sparkContext.master
  
      is_spark_local_mode = (
          spark_master == "local" or spark_master.startswith("local[")
      )
  
      if not (
          spark_master.startswith("spark://")
          or spark_master.startswith("local-cluster[")
          or is_spark_local_mode
      ):
          raise RuntimeError(
              "Ray on Spark only supports spark cluster in standalone mode, "
              "local-cluster mode or spark local mode."
          )
  
      if is_spark_local_mode:
          support_stage_scheduling = False
      elif (
          is_in_databricks_runtime()
          and Version(os.environ["DATABRICKS_RUNTIME_VERSION"]).major
          >= 12
      ):
          support_stage_scheduling = True
      else:
          import pyspark
  
          if Version(pyspark.__version__).release >= (3, 4, 0):
              support_stage_scheduling = True
          else:
              support_stage_scheduling = False
  
      if "num_cpus_per_node" in kwargs:
          if num_cpus_worker_node is not None:
              raise ValueError(
                  "'num_cpus_per_node' and 'num_cpus_worker_node' arguments are "
                  "equivalent. Only set 'num_cpus_worker_node'."
              )
          num_cpus_worker_node = kwargs["num_cpus_per_node"]
          warnings.warn(
              "'num_cpus_per_node' argument is deprecated, please use "
              "'num_cpus_worker_node' argument instead.",
              DeprecationWarning,
          )
  
      if "num_gpus_per_node" in kwargs:
          if num_gpus_worker_node is not None:
              raise ValueError(
                  "'num_gpus_per_node' and 'num_gpus_worker_node' arguments are "
                  "equivalent. Only set 'num_gpus_worker_node'."
              )
          num_gpus_worker_node = kwargs["num_gpus_per_node"]
          warnings.warn(
              "'num_gpus_per_node' argument is deprecated, please use "
              "'num_gpus_worker_node' argument instead.",
              DeprecationWarning,
          )
  
      if "object_store_memory_per_node" in kwargs:
          if object_store_memory_worker_node is not None:
              raise ValueError(
                  "'object_store_memory_per_node' and 'object_store_memory_worker_node' "
                  "arguments  are equivalent. Only set "
                  "'object_store_memory_worker_node'."
              )
          object_store_memory_worker_node = kwargs[
              "object_store_memory_per_node"
          ]
          warnings.warn(
              "'object_store_memory_per_node' argument is deprecated, please use "
              "'object_store_memory_worker_node' argument instead.",
              DeprecationWarning,
          )
  
      # Environment configurations within the Spark Session that dictate how many cpus
      # and gpus to use for each submitted spark task.
      num_spark_task_cpus = int(
          spark.sparkContext.getConf().get("spark.task.cpus", "1")
      )
  
      if num_cpus_worker_node is not None and num_cpus_worker_node <= 0:
          raise ValueError(
              "Argument `num_cpus_worker_node` value must be > 0."
          )
  
      num_spark_task_gpus = int(
          spark.sparkContext.getConf().get(
              "spark.task.resource.gpu.amount", "0"
          )
      )
  
      if num_gpus_worker_node is not None and num_gpus_worker_node < 0:
          raise ValueError(
              "Argument `num_gpus_worker_node` value must be >= 0."
          )
  
      if (
          num_cpus_worker_node is not None
          or num_gpus_worker_node is not None
      ):
          if support_stage_scheduling:
              num_cpus_worker_node = (
                  num_cpus_worker_node or num_spark_task_cpus
              )
              num_gpus_worker_node = (
                  num_gpus_worker_node or num_spark_task_gpus
              )
  
              using_stage_scheduling = True
              res_profile = _create_resource_profile(
                  num_cpus_worker_node, num_gpus_worker_node
              )
          else:
              raise ValueError(
                  "Current spark version does not support stage scheduling, so that "
                  "you cannot set the argument `num_cpus_worker_node` and "
                  "`num_gpus_worker_node` values. Without setting the 2 arguments, "
                  "per-Ray worker node will be assigned with number of "
                  f"'spark.task.cpus' (equals to {num_spark_task_cpus}) cpu cores "
                  "and number of 'spark.task.resource.gpu.amount' "
                  f"(equals to {num_spark_task_gpus}) GPUs. To enable spark stage "
                  "scheduling, you need to upgrade spark to 3.4 version or use "
                  "Databricks Runtime 12.x, and you cannot use spark local mode."
              )
      else:
          using_stage_scheduling = False
          res_profile = None
  
          num_cpus_worker_node = num_spark_task_cpus
          num_gpus_worker_node = num_spark_task_gpus
  
      (
          ray_worker_node_heap_mem_bytes,
          ray_worker_node_object_store_mem_bytes,
      ) = get_avail_mem_per_ray_worker_node(
          spark,
          object_store_memory_worker_node,
          num_cpus_worker_node,
          num_gpus_worker_node,
      )
  
      if "num_worker_nodes" in kwargs:
          raise ValueError(
              "'num_worker_nodes' argument is removed, please set "
              "'max_worker_nodes' and 'min_worker_nodes' argument instead."
          )
  
      if max_worker_nodes == MAX_NUM_WORKER_NODES:
          if min_worker_nodes is not None:
              raise ValueError(
                  "If you set 'max_worker_nodes' to 'MAX_NUM_WORKER_NODES', autoscaling "
                  "is not supported, so that you cannot set 'min_worker_nodes' argument "
                  "and 'min_worker_nodes' is automatically set to be equal to "
                  "'max_worker_nodes'."
              )
  
          # max_worker_nodes=MAX_NUM_WORKER_NODES represents using all available
          # spark task slots
          max_worker_nodes = get_max_num_concurrent_tasks(
              spark.sparkContext, res_profile
          )
          min_worker_nodes = max_worker_nodes
      elif max_worker_nodes <= 0:
          raise ValueError(
              "The value of 'max_worker_nodes' argument must be either a positive "
              "integer or 'ray.util.spark.MAX_NUM_WORKER_NODES'."
          )
  
      if "autoscale" in kwargs:
          raise ValueError(
              "'autoscale' argument is removed. You can set 'min_worker_nodes' argument "
              "to be less than 'max_worker_nodes' to make autoscaling enabled."
          )
  
      if min_worker_nodes is None:
          min_worker_nodes = max_worker_nodes
      elif not (0 <= min_worker_nodes <= max_worker_nodes):
          raise ValueError(
              "The value of 'max_worker_nodes' argument must be an integer >= 0 "
              "and <= 'max_worker_nodes'"
          )
  
      insufficient_resources = []
  
      if num_cpus_worker_node < 4:
          insufficient_resources.append(
              "The provided CPU resources for each ray worker are inadequate to start "
              "a ray cluster. Based on the total cpu resources available and the "
              "configured task sizing, each ray worker node would start with "
              f"{num_cpus_worker_node} CPU cores. This is less than the recommended "
              "value of `4` CPUs per worker. On spark version >= 3.4 or Databricks "
              "Runtime 12.x, you can set the argument `num_cpus_worker_node` to "
              "a value >= 4 to address it, otherwise you need to increase the spark "
              "application configuration 'spark.task.cpus' to a minimum of `4` to "
              "address it."
          )
  
      if ray_worker_node_heap_mem_bytes < 10 * 1024 * 1024 * 1024:
          insufficient_resources.append(
              "The provided memory resources for each ray worker node are inadequate. "
              "Based on the total memory available on the spark cluster and the "
              "configured task sizing, each ray worker would start with "
              f"{ray_worker_node_heap_mem_bytes} bytes heap memory. This is less than "
              "the recommended value of 10GB. The ray worker node heap memory size is "
              "calculated by "
              "(SPARK_WORKER_NODE_PHYSICAL_MEMORY / num_local_spark_task_slots * 0.8) - "
              "object_store_memory_worker_node. To increase the heap space available, "
              "increase the memory in the spark cluster by using instance types with "
              "larger memory, or increase number of CPU/GPU per Ray worker node "
              "(so it leads to less Ray worker node slots per spark worker node), "
              "or apply a lower `object_store_memory_worker_node`."
          )
      if insufficient_resources:
          if strict_mode:
              raise ValueError(
                  "You are creating ray cluster on spark with strict mode (it can be "
                  "disabled by setting argument 'strict_mode=False' when calling API "
                  "'setup_ray_cluster'), strict mode requires the spark cluster config "
                  "satisfying following criterion: "
                  "\n".join(insufficient_resources)
              )
          else:
              _logger.warning("\n".join(insufficient_resources))
  
      if num_cpus_head_node is None:
          num_cpus_head_node = 0
      else:
          if num_cpus_head_node < 0:
              raise ValueError(
                  "Argument `num_cpus_head_node` value must be >= 0. "
                  f"Current value is {num_cpus_head_node}."
              )
  
      if num_gpus_head_node is None:
          num_gpus_head_node = 0
      else:
          if num_gpus_head_node < 0:
              raise ValueError(
                  "Argument `num_gpus_head_node` value must be >= 0."
                  f"Current value is {num_gpus_head_node}."
              )
  
      if (
          num_cpus_head_node == 0
          and num_gpus_head_node == 0
          and object_store_memory_head_node is None
      ):
          # Because tasks that require CPU or GPU resources are not scheduled to Ray
          # head node, and user does not set `object_store_memory_head_node` explicitly,
          # limit the heap memory and object store memory allocation to the
          # head node, in order to save spark driver memory.
          heap_memory_head_node = 1024 * 1024 * 1024
          object_store_memory_head_node = 1024 * 1024 * 1024
      else:
          (
              heap_memory_head_node,
              object_store_memory_head_node,
          ) = calc_mem_ray_head_node(object_store_memory_head_node)
  
      with _active_ray_cluster_rwlock:
          cluster = _setup_ray_cluster(
              max_worker_nodes=max_worker_nodes,
              min_worker_nodes=min_worker_nodes,
              num_cpus_worker_node=num_cpus_worker_node,
              num_cpus_head_node=num_cpus_head_node,
              num_gpus_worker_node=num_gpus_worker_node,
              num_gpus_head_node=num_gpus_head_node,
              using_stage_scheduling=using_stage_scheduling,
              heap_memory_worker_node=ray_worker_node_heap_mem_bytes,
              heap_memory_head_node=heap_memory_head_node,
              object_store_memory_worker_node=ray_worker_node_object_store_mem_bytes,
              object_store_memory_head_node=object_store_memory_head_node,
              head_node_options=head_node_options,
              worker_node_options=worker_node_options,
              ray_temp_root_dir=ray_temp_root_dir,
              collect_log_to_path=collect_log_to_path,
              autoscale_upscaling_speed=autoscale_upscaling_speed,
              autoscale_idle_timeout_minutes=autoscale_idle_timeout_minutes,
              is_global=is_global,
          )
  
          cluster.wait_until_ready()  # NB: this line might raise error.
  
          # If connect cluster successfully, set global _active_ray_cluster to be the
          # started cluster.
          _active_ray_cluster = cluster
  
      head_ip = cluster.address.split(":")[0]
      remote_connection_address = (
          f"ray://{head_ip}:{cluster.ray_client_server_port}"
      )
      return cluster.address, remote_connection_address
  
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 12 for calling a global function with example usage: _setup_ray_cluster_internal(max_worker_nodes, min_worker_nodes, num_cpus_worker_node, num_cpus_head_node, num_gpus_worker_node, num_gpus_head_node, object_store_memory_worker_node, object_store_memory_head_node, head_node_options, worker_node_options, ray_temp_root_dir, strict_mode, collect_log_to_path, autoscale_upscaling_speed, autoscale_idle_timeout_minutes, is_global) and returns: cluster,  remote_connection_address