code: def setup_global_ray_cluster(
      *,
      max_worker_nodes: int,
      is_blocking: bool = True,
      min_worker_nodes: Optional[int] = None,
      num_cpus_worker_node: Optional[int] = None,
      num_cpus_head_node: Optional[int] = None,
      num_gpus_worker_node: Optional[int] = None,
      num_gpus_head_node: Optional[int] = None,
      object_store_memory_worker_node: Optional[int] = None,
      object_store_memory_head_node: Optional[int] = None,
      head_node_options: Optional[Dict] = None,
      worker_node_options: Optional[Dict] = None,
      strict_mode: bool = False,
      collect_log_to_path: Optional[str] = None,
      autoscale_upscaling_speed: Optional[float] = 1.0,
      autoscale_idle_timeout_minutes: Optional[float] = 1.0,
  ):
      """
      Set up a global mode cluster.
      The global Ray on spark cluster means:
      - You can only create one active global Ray on spark cluster at a time.
      On databricks cluster, the global Ray cluster can be used by all users,
      - as contrast, non-global Ray cluster can only be used by current notebook
      user.
      - It is up persistently without automatic shutdown.
      - On databricks notebook, you can connect to the global cluster by calling
      ``ray.init()`` without specifying its address, it will discover the
      global cluster automatically if it is up.
  
      For global mode, the ``ray_temp_root_dir`` argument is not supported.
      Global model Ray cluster always use the default Ray temporary directory
      path.
  
      All arguments are the same with ``setup_ray_cluster`` API except that:
      - the ``ray_temp_root_dir`` argument is not supported.
      Global model Ray cluster always use the default Ray temporary directory
      path.
      - A new argument "is_blocking" (default ``True``) is added.
      If "is_blocking" is True,
      then keep the call blocking until it is interrupted.
      once the call is interrupted, the global Ray on spark cluster is shut down and
      `serve_global_ray_cluster` call terminates.
      If "is_blocking" is False,
      once Ray cluster setup completes, return immediately.
      """
  
      cluster_address = _setup_ray_cluster_internal(
          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,
          object_store_memory_worker_node=object_store_memory_worker_node,
          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=None,
          strict_mode=strict_mode,
          collect_log_to_path=collect_log_to_path,
          autoscale_upscaling_speed=autoscale_upscaling_speed,
          autoscale_idle_timeout_minutes=autoscale_idle_timeout_minutes,
          is_global=True,
      )
  
      if not is_blocking:
          return cluster_address
  
      global _global_ray_cluster_cancel_event
      try:
          _global_ray_cluster_cancel_event = Event()
          # serve forever until user cancel the command.
          _global_ray_cluster_cancel_event.wait()
      finally:
          _global_ray_cluster_cancel_event = None
          # once the program is interrupted,
          # or the corresponding databricks notebook command is interrupted
          # shut down the Ray cluster.
          shutdown_ray_cluster()
  
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 14 for calling a global function with example usage: setup_global_ray_cluster() and returns: cluster_address