code: def stop(force: bool, grace_period: int):
      """Stop Ray processes manually on the local machine."""
      is_linux = sys.platform.startswith("linux")
      total_procs_found = 0
      total_procs_stopped = 0
      procs_not_gracefully_killed = []
  
      def kill_procs(
          force: bool, grace_period: int, processes_to_kill: List[str]
      ) -> Tuple[int, int, List[psutil.Process]]:
          """Find all processes from `processes_to_kill` and terminate them.
  
          Unless `force` is specified, it gracefully kills processes. If
          processes are not cleaned within `grace_period`, it force kill all
          remaining processes.
  
          Returns:
              total_procs_found: Total number of processes found from
                  `processes_to_kill` is added.
              total_procs_stopped: Total number of processes gracefully
                  stopped from `processes_to_kill` is added.
              procs_not_gracefully_killed: If processes are not killed
                  gracefully, they are added here.
          """
          process_infos = []
          for proc in psutil.process_iter(["name", "cmdline"]):
              try:
                  process_infos.append(
                      (proc, proc.name(), proc.cmdline())
                  )
              except psutil.Error:
                  pass
  
          stopped = []
          for keyword, filter_by_cmd in processes_to_kill:
              if filter_by_cmd and is_linux and len(keyword) > 15:
                  # getting here is an internal bug, so we do not use cli_logger
                  msg = (
                      "The filter string should not be more than {} "
                      "characters. Actual length: {}. Filter: {}"
                  ).format(15, len(keyword), keyword)
                  raise ValueError(msg)
  
              found = []
              for candidate in process_infos:
                  proc, proc_cmd, proc_args = candidate
                  corpus = (
                      proc_cmd
                      if filter_by_cmd
                      else subprocess.list2cmdline(proc_args)
                  )
                  if keyword in corpus:
                      found.append(candidate)
              for proc, proc_cmd, proc_args in found:
                  proc_string = str(subprocess.list2cmdline(proc_args))
                  try:
                      if force:
                          proc.kill()
                      else:
                          # TODO(mehrdadn): On Windows, this is forceful termination.
                          # We don't want CTRL_BREAK_EVENT, because that would
                          # terminate the entire process group. What to do?
                          proc.terminate()
  
                      if force:
                          cli_logger.verbose(
                              "Killed `{}` {} ",
                              cf.bold(proc_string),
                              cf.dimmed("(via SIGKILL)"),
                          )
                      else:
                          cli_logger.verbose(
                              "Send termination request to `{}` {}",
                              cf.bold(proc_string),
                              cf.dimmed("(via SIGTERM)"),
                          )
  
                      stopped.append(proc)
                  except psutil.NoSuchProcess:
                      cli_logger.verbose(
                          "Attempted to stop `{}`, but process was already dead.",
                          cf.bold(proc_string),
                      )
                  except (psutil.Error, OSError) as ex:
                      cli_logger.error(
                          "Could not terminate `{}` due to {}",
                          cf.bold(proc_string),
                          str(ex),
                      )
  
          # Wait for the processes to actually stop.
          # Dedup processes.
          stopped, alive = psutil.wait_procs(stopped, timeout=0)
          procs_to_kill = stopped + alive
          total_found = len(procs_to_kill)
  
          # Wait for grace period to terminate processes.
          gone_procs = set()
  
          def on_terminate(proc):
              gone_procs.add(proc)
              cli_logger.print(
                  f"{len(gone_procs)}/{total_found} stopped.", end="\r"
              )
  
          stopped, alive = psutil.wait_procs(
              procs_to_kill, timeout=grace_period, callback=on_terminate
          )
          total_stopped = len(stopped)
  
          # For processes that are not killed within the grace period,
          # we send force termination signals.
          for proc in alive:
              proc.kill()
          # Wait a little bit to make sure processes are killed forcefully.
          psutil.wait_procs(alive, timeout=2)
          return total_found, total_stopped, alive
  
      # Process killing procedure: we put processes into 3 buckets.
      # Bucket 1: raylet
      # Bucket 2: all other processes, e.g. dashboard, runtime env agents
      # Bucket 3: gcs_server.
      #
      # For each bucket, we send sigterm to all processes, then wait for 30s, then if
      # they are still alive, send sigkill.
      processes_to_kill = RAY_PROCESSES
      # Raylet should exit before all other processes exit.
      # Otherwise, fate-sharing agents will complain and suicide.
      assert processes_to_kill[0][0] == "raylet"
  
      # GCS should exit after all other processes exit.
      # Otherwise, some of processes may exit with an unexpected
      # exit code which breaks ray start --block.
      assert processes_to_kill[-1][0] == "gcs_server"
  
      buckets = [
          [processes_to_kill[0]],
          processes_to_kill[1:-1],
          [processes_to_kill[-1]],
      ]
  
      for bucket in buckets:
          found, stopped, alive = kill_procs(
              force, grace_period / len(buckets), bucket
          )
          total_procs_found += found
          total_procs_stopped += stopped
          procs_not_gracefully_killed.extend(alive)
  
      # Print the termination result.
      if total_procs_found == 0:
          cli_logger.print("Did not find any active Ray processes.")
      else:
          if total_procs_stopped == total_procs_found:
              cli_logger.success(
                  "Stopped all {} Ray processes.", total_procs_stopped
              )
          else:
              cli_logger.warning(
                  f"Stopped only {total_procs_stopped} out of {total_procs_found} "
                  f"Ray processes within the grace period {grace_period} seconds. "
                  f"Set `{cf.bold('-v')}` to see more details. "
                  f"Remaining processes {procs_not_gracefully_killed} "
                  "will be forcefully terminated.",
              )
              cli_logger.warning(
                  f"You can also use `{cf.bold('--force')}` to forcefully terminate "
                  "processes or set higher `--grace-period` to wait longer time for "
                  "proper termination."
              )
  
      # NOTE(swang): This will not reset the cluster address for a user-defined
      # temp_dir. This is fine since it will get overwritten the next time we
      # call `ray start`.
      ray._private.utils.reset_ray_address()
  
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 7 for calling a global function with example usage: stop(force, grace_period) and returns: total_found,  total_stopped,  alive