code: def start(
      node_ip_address,
      address,
      port,
      node_name,
      redis_password,
      redis_shard_ports,
      object_manager_port,
      node_manager_port,
      gcs_server_port,
      min_worker_port,
      max_worker_port,
      worker_port_list,
      ray_client_server_port,
      memory,
      object_store_memory,
      redis_max_memory,
      num_cpus,
      num_gpus,
      resources,
      head,
      include_dashboard,
      dashboard_host,
      dashboard_port,
      dashboard_agent_listen_port,
      dashboard_agent_grpc_port,
      dashboard_grpc_port,
      runtime_env_agent_port,
      block,
      plasma_directory,
      autoscaling_config,
      no_redirect_output,
      plasma_store_socket_name,
      raylet_socket_name,
      temp_dir,
      storage,
      system_config,
      enable_object_reconstruction,
      metrics_export_port,
      no_monitor,
      tracing_startup_hook,
      ray_debugger_external,
      disable_usage_stats,
      labels,
  ):
      """Start Ray processes manually on the local machine."""
  
      if gcs_server_port is not None:
          cli_logger.error(
              "`{}` is deprecated and ignored. Use {} to specify "
              "GCS server port on head node.",
              cf.bold("--gcs-server-port"),
              cf.bold("--port"),
          )
      # Whether the original arguments include node_ip_address.
      include_node_ip_address = False
      if node_ip_address is not None:
          include_node_ip_address = True
          node_ip_address = services.resolve_ip_for_localhost(
              node_ip_address
          )
  
      resources = parse_resources_json(resources, cli_logger, cf)
      labels_dict = parse_node_labels_json(labels, cli_logger, cf)
  
      if plasma_store_socket_name is not None:
          warnings.warn(
              "plasma_store_socket_name is deprecated and will be removed. You are not "
              "supposed to specify this parameter as it's internal.",
              DeprecationWarning,
              stacklevel=2,
          )
      if raylet_socket_name is not None:
          warnings.warn(
              "raylet_socket_name is deprecated and will be removed. You are not "
              "supposed to specify this parameter as it's internal.",
              DeprecationWarning,
              stacklevel=2,
          )
      if temp_dir and not head:
          cli_logger.warning(
              f"`--temp-dir={temp_dir}` option will be ignored. "
              "`--head` is a required flag to use `--temp-dir`. "
              "temp_dir is only configurable from a head node. "
              "All the worker nodes will use the same temp_dir as a head node. "
          )
          temp_dir = None
  
      redirect_output = None if not no_redirect_output else True
  
      # no  client, no  port -> ok
      # no  port, has client -> default to 10001
      # has port, no  client -> value error
      # has port, has client -> ok, check port validity
      has_ray_client = check_ray_client_dependencies_installed()
      if has_ray_client and ray_client_server_port is None:
          ray_client_server_port = 10001
  
      ray_params = ray._private.parameter.RayParams(
          node_ip_address=node_ip_address,
          node_name=node_name if node_name else node_ip_address,
          min_worker_port=min_worker_port,
          max_worker_port=max_worker_port,
          worker_port_list=worker_port_list,
          ray_client_server_port=ray_client_server_port,
          object_manager_port=object_manager_port,
          node_manager_port=node_manager_port,
          memory=memory,
          object_store_memory=object_store_memory,
          redis_password=redis_password,
          redirect_output=redirect_output,
          num_cpus=num_cpus,
          num_gpus=num_gpus,
          resources=resources,
          labels=labels_dict,
          autoscaling_config=autoscaling_config,
          plasma_directory=plasma_directory,
          huge_pages=False,
          plasma_store_socket_name=plasma_store_socket_name,
          raylet_socket_name=raylet_socket_name,
          temp_dir=temp_dir,
          storage=storage,
          include_dashboard=include_dashboard,
          dashboard_host=dashboard_host,
          dashboard_port=dashboard_port,
          dashboard_agent_listen_port=dashboard_agent_listen_port,
          metrics_agent_port=dashboard_agent_grpc_port,
          dashboard_grpc_port=dashboard_grpc_port,
          runtime_env_agent_port=runtime_env_agent_port,
          _system_config=system_config,
          enable_object_reconstruction=enable_object_reconstruction,
          metrics_export_port=metrics_export_port,
          no_monitor=no_monitor,
          tracing_startup_hook=tracing_startup_hook,
          ray_debugger_external=ray_debugger_external,
      )
  
      if ray_constants.RAY_START_HOOK in os.environ:
          _load_class(os.environ[ray_constants.RAY_START_HOOK])(
              ray_params, head
          )
  
      if head:
          # Start head node.
  
          if disable_usage_stats:
              usage_lib.set_usage_stats_enabled_via_env_var(False)
          usage_lib.show_usage_stats_prompt(cli=True)
          cli_logger.newline()
  
          if port is None:
              port = ray_constants.DEFAULT_PORT
  
          # Set bootstrap port.
          assert ray_params.redis_port is None
          assert ray_params.gcs_server_port is None
          ray_params.gcs_server_port = port
  
          if os.environ.get("RAY_FAKE_CLUSTER"):
              ray_params.env_vars = {
                  "RAY_OVERRIDE_NODE_ID_FOR_TESTING": FAKE_HEAD_NODE_ID
              }
  
          num_redis_shards = None
          # Start Ray on the head node.
          if redis_shard_ports is not None and address is None:
              redis_shard_ports = redis_shard_ports.split(",")
              # Infer the number of Redis shards from the ports if the number is
              # not provided.
              num_redis_shards = len(redis_shard_ports)
  
          # This logic is deprecated and will be removed later.
          if address is not None:
              cli_logger.warning(
                  "Specifying {} for external Redis address is deprecated. "
                  "Please specify environment variable {}={} instead.",
                  cf.bold("--address"),
                  cf.bold("RAY_REDIS_ADDRESS"),
                  address,
              )
              external_addresses = address.split(",")
  
              # We reuse primary redis as sharding when there's only one
              # instance provided.
              if len(external_addresses) == 1:
                  external_addresses.append(external_addresses[0])
  
              ray_params.update_if_absent(
                  external_addresses=external_addresses
              )
              num_redis_shards = len(external_addresses) - 1
              if redis_password == ray_constants.REDIS_DEFAULT_PASSWORD:
                  cli_logger.warning(
                      "`{}` should not be specified as empty string if "
                      "external redis server(s) `{}` points to requires "
                      "password.",
                      cf.bold("--redis-password"),
                      cf.bold("--address"),
                  )
  
          # Get the node IP address if one is not provided.
          ray_params.update_if_absent(
              node_ip_address=services.get_node_ip_address()
          )
          cli_logger.labeled_value(
              "Local node IP", ray_params.node_ip_address
          )
  
          # Initialize Redis settings.
          ray_params.update_if_absent(
              redis_shard_ports=redis_shard_ports,
              redis_max_memory=redis_max_memory,
              num_redis_shards=num_redis_shards,
              redis_max_clients=None,
          )
  
          # Fail early when starting a new cluster when one is already running
          if address is None:
              default_address = f"{ray_params.node_ip_address}:{port}"
              bootstrap_address = services.find_bootstrap_address(
                  temp_dir
              )
              if (
                  default_address == bootstrap_address
                  and bootstrap_address in services.find_gcs_addresses()
              ):
                  # The default address is already in use by a local running GCS
                  # instance.
                  raise ConnectionError(
                      f"Ray is trying to start at {default_address}, "
                      f"but is already running at {bootstrap_address}. "
                      "Please specify a different port using the `--port`"
                      " flag of `ray start` command."
                  )
  
          node = ray._private.node.Node(
              ray_params,
              head=True,
              shutdown_at_exit=block,
              spawn_reaper=block,
          )
  
          bootstrap_address = node.address
  
          # this is a noop if new-style is not set, so the old logger calls
          # are still in place
          cli_logger.newline()
          startup_msg = "Ray runtime started."
          cli_logger.success("-" * len(startup_msg))
          cli_logger.success(startup_msg)
          cli_logger.success("-" * len(startup_msg))
          cli_logger.newline()
          with cli_logger.group("Next steps"):
              dashboard_url = node.address_info["webui_url"]
              if ray_constants.ENABLE_RAY_CLUSTER:
                  cli_logger.print(
                      "To add another node to this Ray cluster, run"
                  )
                  # NOTE(kfstorm): Java driver rely on this line to get the address
                  # of the cluster. Please be careful when updating this line.
                  cli_logger.print(
                      cf.bold(" {} ray start --address='{}'"),
                      f" {ray_constants.ENABLE_RAY_CLUSTERS_ENV_VAR}=1"
                      if ray_constants.IS_WINDOWS_OR_OSX
                      else "",
                      bootstrap_address,
                  )
  
              cli_logger.newline()
              cli_logger.print("To connect to this Ray cluster:")
              with cli_logger.indented():
                  cli_logger.print("{} ray", cf.magenta("import"))
                  cli_logger.print(
                      "ray{}init({})",
                      cf.magenta("."),
                      "_node_ip_address{}{}".format(
                          cf.magenta("="),
                          cf.yellow("'" + node_ip_address + "'"),
                      )
                      if include_node_ip_address
                      else "",
                  )
  
              if dashboard_url:
                  cli_logger.newline()
                  cli_logger.print(
                      "To submit a Ray job using the Ray Jobs CLI:"
                  )
                  cli_logger.print(
                      cf.bold(
                          "  RAY_ADDRESS='http://{}' ray job submit "
                          "--working-dir . "
                          "-- python my_script.py"
                      ),
                      dashboard_url,
                  )
                  cli_logger.newline()
                  cli_logger.print(
                      "See https://docs.ray.io/en/latest/cluster/running-applications"
                      "/job-submission/index.html "
                  )
                  cli_logger.print(
                      "for more information on submitting Ray jobs to the Ray cluster."
                  )
  
              cli_logger.newline()
              cli_logger.print("To terminate the Ray runtime, run")
              cli_logger.print(cf.bold("  ray stop"))
  
              cli_logger.newline()
              cli_logger.print("To view the status of the cluster, use")
              cli_logger.print("  {}".format(cf.bold("ray status")))
  
              if dashboard_url:
                  cli_logger.newline()
                  cli_logger.print(
                      "To monitor and debug Ray, view the dashboard at "
                  )
                  cli_logger.print(
                      "  {}".format(
                          cf.bold(dashboard_url),
                      )
                  )
  
                  cli_logger.newline()
                  cli_logger.print(
                      cf.underlined(
                          "If connection to the dashboard fails, check your "
                          "firewall settings and "
                          "network configuration."
                      )
                  )
          ray_params.gcs_address = bootstrap_address
      else:
          # Start worker node.
          if not ray_constants.ENABLE_RAY_CLUSTER:
              cli_logger.abort(
                  "Multi-node Ray clusters are not supported on Windows and OSX. "
                  "Restart the Ray cluster with the environment variable `{}=1` "
                  "to proceed anyway.",
                  cf.bold(ray_constants.ENABLE_RAY_CLUSTERS_ENV_VAR),
              )
              raise Exception(
                  "Multi-node Ray clusters are not supported on Windows and OSX. "
                  "Restart the Ray cluster with the environment variable "
                  f"`{ray_constants.ENABLE_RAY_CLUSTERS_ENV_VAR}=1` to proceed "
                  "anyway.",
              )
  
          # Ensure `--address` flag is specified.
          if address is None:
              cli_logger.abort(
                  "`{}` is a required flag unless starting a head node with `{}`.",
                  cf.bold("--address"),
                  cf.bold("--head"),
              )
              raise Exception(
                  "`--address` is a required flag unless starting a "
                  "head node with `--head`."
              )
  
          # Raise error if any head-only flag are specified.
          head_only_flags = {
              "--port": port,
              "--redis-shard-ports": redis_shard_ports,
              "--include-dashboard": include_dashboard,
          }
          for flag, val in head_only_flags.items():
              if val is None:
                  continue
              cli_logger.abort(
                  "`{}` should only be specified when starting head node with `{}`.",
                  cf.bold(flag),
                  cf.bold("--head"),
              )
              raise ValueError(
                  f"{flag} should only be specified when starting head node "
                  "with `--head`."
              )
  
          # Start Ray on a non-head node.
          bootstrap_address = services.canonicalize_bootstrap_address(
              address, temp_dir=temp_dir
          )
  
          if bootstrap_address is None:
              cli_logger.abort(
                  "Cannot canonicalize address `{}={}`.",
                  cf.bold("--address"),
                  cf.bold(address),
              )
              raise Exception(
                  "Cannot canonicalize address "
                  f"`--address={address}`."
              )
  
          ray_params.gcs_address = bootstrap_address
  
          # Get the node IP address if one is not provided.
          ray_params.update_if_absent(
              node_ip_address=services.get_node_ip_address(
                  bootstrap_address
              )
          )
  
          cli_logger.labeled_value(
              "Local node IP", ray_params.node_ip_address
          )
  
          node = ray._private.node.Node(
              ray_params,
              head=False,
              shutdown_at_exit=block,
              spawn_reaper=block,
          )
          temp_dir = node.get_temp_dir_path()
  
          # Ray and Python versions should probably be checked before
          # initializing Node.
          node.check_version_info()
  
          cli_logger.newline()
          startup_msg = "Ray runtime started."
          cli_logger.success("-" * len(startup_msg))
          cli_logger.success(startup_msg)
          cli_logger.success("-" * len(startup_msg))
          cli_logger.newline()
          cli_logger.print("To terminate the Ray runtime, run")
          cli_logger.print(cf.bold("  ray stop"))
          cli_logger.flush()
  
      assert ray_params.gcs_address is not None
      ray._private.utils.write_ray_address(
          ray_params.gcs_address, temp_dir
      )
  
      if block:
          cli_logger.newline()
          with cli_logger.group(cf.bold("--block")):
              cli_logger.print(
                  "This command will now block forever until terminated by a signal."
              )
              cli_logger.print(
                  "Running subprocesses are monitored and a message will be "
                  "printed if any of them terminate unexpectedly. Subprocesses "
                  "exit with SIGTERM will be treated as graceful, thus NOT reported."
              )
              cli_logger.flush()
  
          while True:
              time.sleep(1)
              deceased = node.dead_processes()
  
              # Report unexpected exits of subprocesses with unexpected return codes.
              # We are explicitly expecting SIGTERM because this is how `ray stop` sends
              # shutdown signal to subprocesses, i.e. log_monitor, raylet...
              # NOTE(rickyyx): We are treating 128+15 as an expected return code since
              # this is what autoscaler/_private/monitor.py does upon SIGTERM
              # handling.
              expected_return_codes = [
                  0,
                  signal.SIGTERM,
                  -1 * signal.SIGTERM,
                  128 + signal.SIGTERM,
              ]
              unexpected_deceased = [
                  (process_type, process)
                  for process_type, process in deceased
                  if process.returncode not in expected_return_codes
              ]
              if len(unexpected_deceased) > 0:
                  cli_logger.newline()
                  cli_logger.error(
                      "Some Ray subprocesses exited unexpectedly:"
                  )
  
                  with cli_logger.indented():
                      for process_type, process in unexpected_deceased:
                          cli_logger.error(
                              "{}",
                              cf.bold(str(process_type)),
                              _tags={
                                  "exit code": str(process.returncode)
                              },
                          )
  
                  cli_logger.newline()
                  cli_logger.error(
                      "Remaining processes will be killed."
                  )
                  # explicitly kill all processes since atexit handlers
                  # will not exit with errors.
                  node.kill_all_processes(
                      check_alive=False, allow_graceful=False
                  )
                  os._exit(1)
  
imports: import copy
  import json
  import logging
  import os
  import platform
  import signal
  import subprocess
  import sys
  import time
  import traceback
  import urllib
  import urllib.parse
  import warnings
  import shutil
  from datetime import datetime
  from typing import Optional, Set, List, Tuple
  import click
  import psutil
  import yaml
  import ray
  import ray._private.ray_constants
  import ray._private.services
  from ray._private.utils import check_ray_client_dependencies_installed, parse_resources_json, parse_node_labels_json
  from ray._private.internal_api import memory_summary
  from ray._private.storage import _load_class
  from ray._private.usage import usage_lib
  from ray.autoscaler._private.cli_logger import add_click_logging_options, cf, cli_logger
  from ray.autoscaler._private.commands import RUN_ENV_TYPES, attach_cluster, create_or_update_cluster, debug_status, exec_cluster, get_cluster_dump_archive, get_head_node_ip, get_local_dump_archive, get_worker_node_ips, kill_node, monitor_cluster, rsync, teardown_cluster
  from ray.autoscaler._private.constants import RAY_PROCESSES
  from ray.autoscaler._private.fake_multi_node.node_provider import FAKE_HEAD_NODE_ID
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 6 for calling a global function with example usage: start(node_ip_address, address, port, node_name, redis_password, redis_shard_ports, object_manager_port, node_manager_port, gcs_server_port, min_worker_port, max_worker_port, worker_port_list, ray_client_server_port, memory, object_store_memory, redis_max_memory, num_cpus, num_gpus, resources, head, include_dashboard, dashboard_host, dashboard_port, dashboard_agent_listen_port, dashboard_agent_grpc_port, dashboard_grpc_port, runtime_env_agent_port, block, plasma_directory, autoscaling_config, no_redirect_output, plasma_store_socket_name, raylet_socket_name, temp_dir, storage, system_config, enable_object_reconstruction, metrics_export_port, no_monitor, tracing_startup_hook, ray_debugger_external, disable_usage_stats, labels) without return types