code: def start_api_server(
      include_dashboard: Optional[bool],
      raise_on_failure: bool,
      host: str,
      gcs_address: str,
      node_ip_address: str,
      temp_dir: str,
      logdir: str,
      session_dir: str,
      port: Optional[int] = None,
      dashboard_grpc_port: Optional[int] = None,
      fate_share: Optional[bool] = None,
      max_bytes: int = 0,
      backup_count: int = 0,
      redirect_logging: bool = True,
      stdout_file: Optional[IO[AnyStr]] = subprocess.DEVNULL,
      stderr_file: Optional[IO[AnyStr]] = subprocess.DEVNULL,
  ):
      """Start a API server process.
  
      Args:
          include_dashboard: If true, this will load all dashboard-related modules
              when starting the API server, or fail. If None, it will load all
              dashboard-related modules conditioned on dependencies being present.
              Otherwise, it will only start the modules that are not relevant to
              the dashboard.
          raise_on_failure: If true, this will raise an exception
              if we fail to start the API server. Otherwise it will print
              a warning if we fail to start the API server.
          host: The host to bind the dashboard web server to.
          gcs_address: The gcs address the dashboard should connect to
          node_ip_address: The IP address where this is running.
          temp_dir: The temporary directory used for log files and
              information for this Ray session.
          session_dir: The session directory under temp_dir.
              It is used as a identifier of individual cluster.
          logdir: The log directory used to generate dashboard log.
          port: The port to bind the dashboard web server to.
              Defaults to 8265.
          dashboard_grpc_port: The port which the dashboard listens for
              gRPC on. Defaults to a random, available port.
          max_bytes: Log rotation parameter. Corresponding to
              RotatingFileHandler's maxBytes.
          backup_count: Log rotation parameter. Corresponding to
              RotatingFileHandler's backupCount.
          redirect_logging: Whether we should redirect logging to
              the provided log directory.
          stdout_file: A file handle opened for writing to redirect stdout to. If
              no redirection should happen, then this should be None.
          stderr_file: A file handle opened for writing to redirect stderr to. If
              no redirection should happen, then this should be None.
  
      Returns:
          A tuple of :
              - Dashboard URL if dashboard enabled and started.
              - ProcessInfo for the process that was started.
      """
      try:
          # Make sure port is available.
          if port is None:
              port_retries = 50
              port = ray_constants.DEFAULT_DASHBOARD_PORT
          else:
              port_retries = 0
              port_test_socket = socket.socket()
              port_test_socket.setsockopt(
                  socket.SOL_SOCKET,
                  socket.SO_REUSEADDR,
                  1,
              )
              try:
                  port_test_socket.bind((host, port))
                  port_test_socket.close()
              except socket.error as e:
                  # 10013 on windows is a bit more broad than just
                  # "address in use": it can also indicate "permission denied".
                  # TODO: improve the error message?
                  if e.errno in {
                      48,
                      98,
                      10013,
                  }:  # address already in use.
                      raise ValueError(
                          f"Failed to bind to {host}:{port} because it's "
                          "already occupied. You can use `ray start "
                          "--dashboard-port ...` or `ray.init(dashboard_port=..."
                          ")` to select a different port."
                      )
                  else:
                      raise e
          # Make sure the process can start.
          minimal: bool = (
              not ray._private.utils.check_dashboard_dependencies_installed()
          )
  
          # Explicitly check here that when the user explicitly specifies
          # dashboard inclusion, the install is not minimal.
          if include_dashboard and minimal:
              logger.error(
                  "--include-dashboard is not supported when minimal ray is used."
                  "Download ray[default] to use the dashboard."
              )
              raise Exception(
                  "Cannot include dashboard with missing packages."
              )
  
          include_dash: bool = (
              True if include_dashboard is None else include_dashboard
          )
  
          # Start the dashboard process.
          dashboard_dir = "dashboard"
          dashboard_filepath = os.path.join(
              RAY_PATH, dashboard_dir, "dashboard.py"
          )
  
          command = [
              *_build_python_executable_command_memory_profileable(
                  ray_constants.PROCESS_TYPE_DASHBOARD,
                  session_dir,
                  unbuffered=False,
              ),
              dashboard_filepath,
              f"--host={host}",
              f"--port={port}",
              f"--port-retries={port_retries}",
              f"--temp-dir={temp_dir}",
              f"--log-dir={logdir}",
              f"--session-dir={session_dir}",
              f"--logging-rotate-bytes={max_bytes}",
              f"--logging-rotate-backup-count={backup_count}",
              f"--gcs-address={gcs_address}",
              f"--node-ip-address={node_ip_address}",
          ]
  
          if not redirect_logging:
              # If not redirecting logging to files, unset log filename.
              # This will cause log records to go to stderr.
              command.append("--logging-filename=")
              # Use stderr log format with the component name as a message prefix.
              logging_format = (
                  ray_constants.LOGGER_FORMAT_STDERR.format(
                      component=ray_constants.PROCESS_TYPE_DASHBOARD
                  )
              )
              command.append(f"--logging-format={logging_format}")
              # Inherit stdout/stderr streams so that
              # logs are redirected to stderr.
              stdout_file = None
              stderr_file = None
          if minimal:
              command.append("--minimal")
  
          if not include_dash:
              # If dashboard is not included, load modules
              # that are irrelevant to the dashboard.
              # TODO(sang): Modules like job or state APIs should be
              # loaded although dashboard is disabled. Fix it.
              command.append("--modules-to-load=UsageStatsHead")
              command.append("--disable-frontend")
  
          if dashboard_grpc_port is not None:
              command.append(f"--grpc-port={dashboard_grpc_port}")
  
          process_info = start_ray_process(
              command,
              ray_constants.PROCESS_TYPE_DASHBOARD,
              stdout_file=stdout_file,
              stderr_file=stderr_file,
              fate_share=fate_share,
          )
  
          # Retrieve the dashboard url
          gcs_client = GcsClient(address=gcs_address)
          ray.experimental.internal_kv._initialize_internal_kv(
              gcs_client
          )
          dashboard_url = None
          dashboard_returncode = None
          for _ in range(200):
              dashboard_url = (
                  ray.experimental.internal_kv._internal_kv_get(
                      ray_constants.DASHBOARD_ADDRESS,
                      namespace=ray_constants.KV_NAMESPACE_DASHBOARD,
                  )
              )
              if dashboard_url is not None:
                  dashboard_url = dashboard_url.decode("utf-8")
                  break
              dashboard_returncode = process_info.process.poll()
              if dashboard_returncode is not None:
                  break
              # This is often on the critical path of ray.init() and ray start,
              # so we need to poll often.
              time.sleep(0.1)
  
          # Dashboard couldn't be started.
          if dashboard_url is None:
              returncode_str = (
                  f", return code {dashboard_returncode}"
                  if dashboard_returncode is not None
                  else ""
              )
              logger.error(
                  f"Failed to start the dashboard {returncode_str}"
              )
  
              def read_log(filename, lines_to_read):
                  """Read a log file and return the last 20 lines."""
                  dashboard_log = os.path.join(logdir, filename)
                  # Read last n lines of dashboard log. The log file may be large.
                  lines_to_read = 20
                  lines = []
                  with open(dashboard_log, "rb") as f:
                      with mmap.mmap(
                          f.fileno(), 0, access=mmap.ACCESS_READ
                      ) as mm:
                          end = mm.size()
                          for _ in range(lines_to_read):
                              sep = mm.rfind(b"\n", 0, end - 1)
                              if sep == -1:
                                  break
                              lines.append(
                                  mm[sep + 1 : end].decode("utf-8")
                              )
                              end = sep
                  lines.append(
                      f"The last {lines_to_read} lines of {dashboard_log} "
                      "(it contains the error message from the dashboard): "
                  )
                  return lines
  
              if logdir:
                  lines_to_read = 20
                  logger.error(
                      "Error should be written to 'dashboard.log' or "
                      "'dashboard.err'. We are printing the last "
                      f"{lines_to_read} lines for you. See "
                      "'https://docs.ray.io/en/master/ray-observability/ray-logging.html#logging-directory-structure' "  # noqa
                      "to find where the log file is."
                  )
                  try:
                      lines = read_log(
                          "dashboard.log", lines_to_read=lines_to_read
                      )
                  except Exception as e:
                      logger.error(
                          f"Couldn't read dashboard.log file. Error: {e}. "
                          "It means the dashboard is broken even before it "
                          "initializes the logger (mostly dependency issues). "
                          "Reading the dashboard.err file which contains stdout/stderr."
                      )
                      # If we cannot read the .log file, we fallback to .err file.
                      # This is the case where dashboard couldn't be started at all
                      # and couldn't even initialize the logger to write logs to .log
                      # file.
                      try:
                          lines = read_log(
                              "dashboard.err",
                              lines_to_read=lines_to_read,
                          )
                      except Exception as e:
                          raise Exception(
                              f"Failed to read dashboard.err file: {e}. "
                              "It is unexpected. Please report an issue to "
                              "Ray github. "
                              "https://github.com/ray-project/ray/issues"
                          )
                  last_log_str = "\n" + "\n".join(
                      reversed(lines[-lines_to_read:])
                  )
                  raise Exception(last_log_str)
              else:
                  # Is it reachable?
                  raise Exception("Failed to start a dashboard.")
  
          if minimal or not include_dash:
              # If it is the minimal installation, the web url (dashboard url)
              # shouldn't be configured because it doesn't start a server.
              dashboard_url = ""
          return dashboard_url, process_info
      except Exception as e:
          if raise_on_failure:
              raise e from e
          else:
              logger.error(e)
              return None, None
  
imports: import base64
  import collections
  import errno
  import io
  import json
  import logging
  import mmap
  import multiprocessing
  import os
  import random
  import shutil
  import signal
  import socket
  import subprocess
  import sys
  import time
  from pathlib import Path
  from typing import List, Optional, IO, AnyStr
  import psutil
  from filelock import FileLock
  import ray
  import ray._private.ray_constants
  from ray._raylet import GcsClient, GcsClientOptions
  from ray.core.generated.common_pb2 import Language
  from ray._private.ray_constants import RAY_NODE_IP_FILENAME
label_desc: Use image node_id 31 for calling a global function with example usage: start_api_server(include_dashboard, raise_on_failure, host, gcs_address, node_ip_address, temp_dir, logdir, session_dir, port, dashboard_grpc_port, fate_share, max_bytes, backup_count, redirect_logging, stdout_file, stderr_file) and returns: dashboard_url,  process_info, lines, None,  None