class_docstr: This class is the type of instance returned by the `_setup_ray_cluster` interface.
  Its main functionality is to:
  Connect to, disconnect from, and shutdown the Ray cluster running on Apache Spark.
  Serve as a Python context manager for the `RayClusterOnSpark` instance.
  
  Args
      address: The url for the ray head node (defined as the hostname and unused
               port on Spark driver node)
      head_proc: Ray head process
      spark_job_group_id: The Spark job id for a submitted ray job
      num_workers_node: The number of workers in the ray cluster.
class_docstr_tok: ["This","class","is","the","type","of","instance","returned","by","the","`","_setup_ray_cluster","`","interface",".","Its","main","functionality","is","to",":","Connect","to",",","disconnect","from",",","and","shutdown","the","Ray","cluster","running","on","Apache","Spark",".","Serve","as","a","Python","context","manager","for","the","`","RayClusterOnSpark","`","instance",".","Args","address",":","The","url","for","the","ray","head","node","(","defined","as","the","hostname","and","unused","port","on","Spark","driver","node",")","head_proc",":","Ray","head","process","spark_job_group_id",":","The","Spark","job","id","for","a","submitted","ray","job","num_workers_node",":","The","number","of","workers","in","the","ray","cluster","."]
code: def wait_until_ready(self):
      import ray
  
      if self.is_shutdown:
          raise RuntimeError(
              "The ray cluster has been shut down or it failed to start."
          )
  
      try:
          ray.init(address=self.address)
  
          if self.ray_dashboard_port is not None and _wait_service_up(
              self.address.split(":")[0],
              self.ray_dashboard_port,
              _RAY_DASHBOARD_STARTUP_TIMEOUT,
          ):
              self.start_hook.on_ray_dashboard_created(
                  self.ray_dashboard_port
              )
          else:
              try:
                  __import__("ray.dashboard.optional_deps")
              except ModuleNotFoundError:
                  _logger.warning(
                      "Dependencies to launch the optional dashboard API "
                      "server cannot be found. They can be installed with "
                      "pip install ray[default]."
                  )
  
          if self.autoscale:
              return
  
          last_alive_worker_count = 0
          last_progress_move_time = time.time()
          while True:
              time.sleep(
                  _RAY_CLUSTER_STARTUP_PROGRESS_CHECKING_INTERVAL
              )
  
              # Inside the waiting ready loop,
              # checking `self.background_job_exception`, if it is not None,
              # it means the background spark job has failed,
              # in this case, raise error directly.
              if self.background_job_exception is not None:
                  raise RuntimeError(
                      "Ray workers failed to start."
                  ) from self.background_job_exception
  
              cur_alive_worker_count = (
                  len([node for node in ray.nodes() if node["Alive"]])
                  - 1
              )  # Minus 1 means excluding the head node.
  
              if cur_alive_worker_count >= self.max_worker_nodes:
                  return
  
              if cur_alive_worker_count > last_alive_worker_count:
                  last_alive_worker_count = cur_alive_worker_count
                  last_progress_move_time = time.time()
                  _logger.info(
                      "Ray worker nodes are starting. Progress: "
                      f"({cur_alive_worker_count} / {self.max_worker_nodes})"
                  )
              else:
                  if (
                      time.time() - last_progress_move_time
                      > _RAY_CONNECT_CLUSTER_POLL_PROGRESS_TIMEOUT
                  ):
                      if cur_alive_worker_count == 0:
                          raise RuntimeError(
                              "Current spark cluster has no resources to launch "
                              "Ray worker nodes."
                          )
                      _logger.warning(
                          "Timeout in waiting for all ray workers to start. "
                          "Started / Total requested: "
                          f"({cur_alive_worker_count} / {self.max_worker_nodes}). "
                          "Current spark cluster does not have sufficient resources "
                          "to launch requested number of Ray worker nodes."
                      )
                      return
      finally:
          ray.shutdown()
  
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 3 for calling the RayClusterOnSpark obj's underlying member method code with example usage: obj.wait_until_ready() without return types