docstr: Initializes all services of a Ray cluster.
  
  Args:
      initialize_head: Automatically start a Ray cluster
          by initializing the head node. Defaults to False.
      connect: If `initialize_head=True` and `connect=True`,
          ray.init will be called with the address of this cluster
          passed in.
      head_node_args: Arguments to be passed into
          `start_ray_head` via `self.add_node`.
      shutdown_at_exit: If True, registers an exit hook
          for shutting down all started processes.
docstr_tok: ["Initializes","all","services","of","a","Ray","cluster",".","Args",":","initialize_head",":","Automatically","start","a","Ray","cluster","by","initializing","the","head","node",".","Defaults","to","False",".","connect",":","If","`","initialize_head=True","`","and","`","connect=True","`",",","ray.init","will","be","called","with","the","address","of","this","cluster","passed","in",".","head_node_args",":","Arguments","to","be","passed","into","`","start_ray_head","`","via","`","self.add_node","`",".","shutdown_at_exit",":","If","True",",","registers","an","exit","hook","for","shutting","down","all","started","processes","."]
code: def __init__(
      self,
      initialize_head: bool = False,
      connect: bool = False,
      head_node_args: dict = None,
      shutdown_at_exit: bool = True,
  ):
      """Initializes all services of a Ray cluster.
  
      Args:
          initialize_head: Automatically start a Ray cluster
              by initializing the head node. Defaults to False.
          connect: If `initialize_head=True` and `connect=True`,
              ray.init will be called with the address of this cluster
              passed in.
          head_node_args: Arguments to be passed into
              `start_ray_head` via `self.add_node`.
          shutdown_at_exit: If True, registers an exit hook
              for shutting down all started processes.
      """
      if cluster_not_supported:
          logger.warning(
              "Ray cluster mode is currently experimental and untested on "
              "Windows. If you are using it and running into issues please "
              "file a report at https://github.com/ray-project/ray/issues."
          )
      self.head_node = None
      self.worker_nodes = set()
      self.redis_address = None
      self.connected = False
      # Create a new global state accessor for fetching GCS table.
      self.global_state = ray._private.state.GlobalState()
      self._shutdown_at_exit = shutdown_at_exit
      if not initialize_head and connect:
          raise RuntimeError("Cannot connect to uninitialized cluster.")
  
      if initialize_head:
          head_node_args = head_node_args or {}
          self.add_node(**head_node_args)
          if connect:
              self.connect()
  
imports: import copy
  import json
  import logging
  import os
  import subprocess
  import tempfile
  import time
  from typing import Dict, Optional
  import yaml
  import ray
  import ray._private.services
  from ray._private import ray_constants
  from ray._private.client_mode_hook import disable_client_hook
  from ray._raylet import GcsClientOptions
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 1 to create a new Cluster object  with example: obj = Cluster(initialize_head, connect, head_node_args, shutdown_at_exit)