text: import binascii
  import errno
  import functools
  import hashlib
  import importlib
  import logging
  import multiprocessing
  import os
  import signal
  import subprocess
  import sys
  import tempfile
  import threading
  import time
  from typing import Optional, Sequence, Tuple, Any, Union, Dict
  import uuid
  import grpc
  import warnings
  
  try:
      from grpc import aio as aiogrpc
  except ImportError:
      from grpc.experimental import aio as aiogrpc
  
  import inspect
  from inspect import signature
  from pathlib import Path
  import numpy as np
  
  import ray
  from ray.core.generated.gcs_pb2 import ErrorTableData
  import ray.ray_constants as ray_constants
  from ray._private.tls_utils import load_certs_from_env
  
  # Import psutil after ray so the packaged version is used.
  import psutil
  
  pwd = None
  if sys.platform != "win32":
      import pwd
  
  logger = logging.getLogger(__name__)
  
  # Linux can bind child processes' lifetimes to that of their parents via prctl.
  # prctl support is detected dynamically once, and assumed thereafter.
  linux_prctl = None
  
  # Windows can bind processes' lifetimes to that of kernel-level "job objects".
  # We keep a global job object to tie its lifetime to that of our own process.
  win32_job = None
  win32_AssignProcessToJobObject = None
  
  
  def get_user_temp_dir():
      if "RAY_TMPDIR" in os.environ:
          return os.environ["RAY_TMPDIR"]
      elif sys.platform.startswith("linux") and "TMPDIR" in os.environ:
          return os.environ["TMPDIR"]
      elif sys.platform.startswith("darwin") or sys.platform.startswith("linux"):
          # Ideally we wouldn't need this fallback, but keep it for now for
          # for compatibility
          tempdir = os.path.join(os.sep, "tmp")
      else:
          tempdir = tempfile.gettempdir()
      return tempdir
  
  
  def get_ray_temp_dir():
      return os.path.join(get_user_temp_dir(), "ray")
  
  
  def _random_string():
      id_hash = hashlib.shake_128()
      id_hash.update(uuid.uuid4().bytes)
      id_bytes = id_hash.digest(ray_constants.ID_SIZE)
      assert len(id_bytes) == ray_constants.ID_SIZE
      return id_bytes
  
  
  def format_error_message(exception_message, task_exception=False):
      """Improve the formatting of an exception thrown by a remote function.
  
      This method takes a traceback from an exception and makes it nicer by
      removing a few uninformative lines and adding some space to indent the
      remaining lines nicely.
  
      Args:
          exception_message (str): A message generated by traceback.format_exc().
  
      Returns:
          A string of the formatted exception message.
      """
      lines = exception_message.split("\n")
      if task_exception:
          # For errors that occur inside of tasks, remove lines 1 and 2 which are
          # always the same, they just contain information about the worker code.
          lines = lines[0:1] + lines[3:]
          pass
      return "\n".join(lines)
  
  
  def push_error_to_driver(worker, error_type, message, job_id=None):
      """Push an error message to the driver to be printed in the background.
  
      Args:
          worker: The worker to use.
          error_type (str): The type of the error.
          message (str): The message that will be printed in the background
              on the driver.
          job_id: The ID of the driver to push the error message to. If this
              is None, then the message will be pushed to all drivers.
      """
      if job_id is None:
          job_id = ray.JobID.nil()
      assert isinstance(job_id, ray.JobID)
      worker.core_worker.push_error(job_id, error_type, message, time.time())
  
  
  def construct_error_message(job_id, error_type, message, timestamp):
      """Construct an ErrorTableData object.
  
      Args:
          job_id: The ID of the job that the error should go to. If this is
              nil, then the error will go to all drivers.
          error_type: The type of the error.
          message: The error message.
          timestamp: The time of the error.
  
      Returns:
          The ErrorTableData object.
      """
      data = ErrorTableData()
      data.job_id = job_id.binary()
      data.type = error_type
      data.error_message = message
      data.timestamp = timestamp
      return data
  
  
  def publish_error_to_driver(
      error_type,
      message,
      gcs_publisher,
      job_id=None,
  ):
      """Push an error message to the driver to be printed in the background.
  
      Normally the push_error_to_driver function should be used. However, in some
      instances, the raylet client is not available, e.g., because the
      error happens in Python before the driver or worker has connected to the
      backend processes.
  
      Args:
          error_type (str): The type of the error.
          message (str): The message that will be printed in the background
              on the driver.
          gcs_publisher: The GCS publisher to use.
          job_id: The ID of the driver to push the error message to. If this
              is None, then the message will be pushed to all drivers.
      """
      if job_id is None:
          job_id = ray.JobID.nil()
      assert isinstance(job_id, ray.JobID)
      error_data = construct_error_message(job_id, error_type, message, time.time())
      gcs_publisher.publish_error(job_id.hex().encode(), error_data)
  
  
  def random_string():
      """Generate a random string to use as an ID.
  
      Note that users may seed numpy, which could cause this function to generate
      duplicate IDs. Therefore, we need to seed numpy ourselves, but we can't
      interfere with the state of the user's random number generator, so we
      extract the state of the random number generator and reset it after we are
      done.
  
      TODO(rkn): If we want to later guarantee that these are generated in a
      deterministic manner, then we will need to make some changes here.
  
      Returns:
          A random byte string of length ray_constants.ID_SIZE.
      """
      # Get the state of the numpy random number generator.
      numpy_state = np.random.get_state()
      # Try to use true randomness.
      np.random.seed(None)
      # Generate the random ID.
      random_id = np.random.bytes(ray_constants.ID_SIZE)
      # Reset the state of the numpy random number generator.
      np.random.set_state(numpy_state)
      return random_id
  
  
  def decode(byte_str: str, allow_none: bool = False, encode_type: str = "utf-8"):
      """Make this unicode in Python 3, otherwise leave it as bytes.
  
      Args:
          byte_str: The byte string to decode.
          allow_none: If true, then we will allow byte_str to be None in which
              case we will return an empty string. TODO(rkn): Remove this flag.
              This is only here to simplify upgrading to flatbuffers 1.10.0.
  
      Returns:
          A byte string in Python 2 and a unicode string in Python 3.
      """
      if byte_str is None and allow_none:
          return ""
  
      if not isinstance(byte_str, bytes):
          raise ValueError(f"The argument {byte_str} must be a bytes object.")
      if sys.version_info >= (3, 0):
          return byte_str.decode(encode_type)
      else:
          return byte_str
  
  
  def ensure_str(s, encoding="utf-8", errors="strict"):
      """Coerce *s* to `str`.
  
      - `str` -> `str`
      - `bytes` -> decoded to `str`
      """
      if isinstance(s, str):
          return s
      else:
          assert isinstance(s, bytes)
          return s.decode(encoding, errors)
  
  
  def binary_to_object_ref(binary_object_ref):
      return ray.ObjectRef(binary_object_ref)
  
  
  def binary_to_task_id(binary_task_id):
      return ray.TaskID(binary_task_id)
  
  
  def binary_to_hex(identifier):
      hex_identifier = binascii.hexlify(identifier)
      if sys.version_info >= (3, 0):
          hex_identifier = hex_identifier.decode()
      return hex_identifier
  
  
  def hex_to_binary(hex_identifier):
      return binascii.unhexlify(hex_identifier)
  
  
  # TODO(qwang): Remove these hepler functions
  # once we separate `WorkerID` from `UniqueID`.
  def compute_job_id_from_driver(driver_id):
      assert isinstance(driver_id, ray.WorkerID)
      return ray.JobID(driver_id.binary()[0 : ray.JobID.size()])
  
  
  def compute_driver_id_from_job(job_id):
      assert isinstance(job_id, ray.JobID)
      rest_length = ray_constants.ID_SIZE - job_id.size()
      driver_id_str = job_id.binary() + (rest_length * b"\xff")
      return ray.WorkerID(driver_id_str)
  
  
  def get_cuda_visible_devices():
      """Get the device IDs in the CUDA_VISIBLE_DEVICES environment variable.
  
      Returns:
          devices (List[str]): If CUDA_VISIBLE_DEVICES is set, returns a
              list of strings representing the IDs of the visible GPUs.
              If it is not set or is set to NoDevFiles, returns empty list.
      """
      gpu_ids_str = os.environ.get("CUDA_VISIBLE_DEVICES", None)
  
      if gpu_ids_str is None:
          return None
  
      if gpu_ids_str == "":
          return []
  
      if gpu_ids_str == "NoDevFiles":
          return []
  
      # GPU identifiers are given as strings representing integers or UUIDs.
      return list(gpu_ids_str.split(","))
  
  
  last_set_gpu_ids = None
  
  
  def set_cuda_visible_devices(gpu_ids):
      """Set the CUDA_VISIBLE_DEVICES environment variable.
  
      Args:
          gpu_ids (List[str]): List of strings representing GPU IDs.
      """
  
      if os.environ.get("RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES"):
          return
  
      global last_set_gpu_ids
      if last_set_gpu_ids == gpu_ids:
          return  # optimization: already set
  
      os.environ["CUDA_VISIBLE_DEVICES"] = ",".join([str(i) for i in gpu_ids])
      last_set_gpu_ids = gpu_ids
  
  
  def resources_from_ray_options(options_dict: Dict[str, Any]) -> Dict[str, Any]:
      """Determine a task's resource requirements.
  
      Args:
          options_dict: The dictionary that contains resources requirements.
  
      Returns:
          A dictionary of the resource requirements for the task.
      """
      resources = (options_dict.get("resources") or {}).copy()
  
      if "CPU" in resources or "GPU" in resources:
          raise ValueError(
              "The resources dictionary must not contain the key 'CPU' or 'GPU'"
          )
      elif "memory" in resources or "object_store_memory" in resources:
          raise ValueError(
              "The resources dictionary must not "
              "contain the key 'memory' or 'object_store_memory'"
          )
  
      num_cpus = options_dict.get("num_cpus")
      num_gpus = options_dict.get("num_gpus")
      memory = options_dict.get("memory")
      object_store_memory = options_dict.get("object_store_memory")
      accelerator_type = options_dict.get("accelerator_type")
  
      if num_cpus is not None:
          resources["CPU"] = num_cpus
      if num_gpus is not None:
          resources["GPU"] = num_gpus
      if memory is not None:
          resources["memory"] = ray_constants.to_memory_units(memory, round_up=True)
      if object_store_memory is not None:
          resources["object_store_memory"] = ray_constants.to_memory_units(
              object_store_memory, round_up=True
          )
      if accelerator_type is not None:
          resources[
              f"{ray_constants.RESOURCE_CONSTRAINT_PREFIX}{accelerator_type}"
          ] = 0.001
  
      return resources
  
  
  class Unbuffered(object):
      """There's no "built-in" solution to programatically disabling buffering of
      text files. Ray expects stdout/err to be text files, so creating an
      unbuffered binary file is unacceptable.
  
      See
      https://mail.python.org/pipermail/tutor/2003-November/026645.html.
      https://docs.python.org/3/library/functions.html#open
  
      """
  
      def __init__(self, stream):
          self.stream = stream
  
      def write(self, data):
          self.stream.write(data)
          self.stream.flush()
  
      def writelines(self, datas):
          self.stream.writelines(datas)
          self.stream.flush()
  
      def __getattr__(self, attr):
          return getattr(self.stream, attr)
  
  
  def open_log(path, unbuffered=False, **kwargs):
      """
      Opens the log file at `path`, with the provided kwargs being given to
      `open`.
      """
      # Disable buffering, see test_advanced_3.py::test_logging_to_driver
      kwargs.setdefault("buffering", 1)
      kwargs.setdefault("mode", "a")
      kwargs.setdefault("encoding", "utf-8")
      stream = open(path, **kwargs)
      if unbuffered:
          return Unbuffered(stream)
      else:
          return stream
  
  
  def get_system_memory(
      # For cgroups v1:
      memory_limit_filename="/sys/fs/cgroup/memory/memory.limit_in_bytes",
      # For cgroups v2:
      memory_limit_filename_v2="/sys/fs/cgroup/memory.max",
  ):
      """Return the total amount of system memory in bytes.
  
      Returns:
          The total amount of system memory in bytes.
      """
      # Try to accurately figure out the memory limit if we are in a docker
      # container. Note that this file is not specific to Docker and its value is
      # often much larger than the actual amount of memory.
      docker_limit = None
      if os.path.exists(memory_limit_filename):
          with open(memory_limit_filename, "r") as f:
              docker_limit = int(f.read())
      elif os.path.exists(memory_limit_filename_v2):
          with open(memory_limit_filename_v2, "r") as f:
              max_file = f.read()
              if max_file.isnumeric():
                  docker_limit = int(max_file)
              else:
                  # max_file is "max", i.e. is unset.
                  docker_limit = None
  
      # Use psutil if it is available.
      psutil_memory_in_bytes = psutil.virtual_memory().total
  
      if docker_limit is not None:
          # We take the min because the cgroup limit is very large if we aren't
          # in Docker.
          return min(docker_limit, psutil_memory_in_bytes)
  
      return psutil_memory_in_bytes
  
  
  def _get_docker_cpus(
      cpu_quota_file_name="/sys/fs/cgroup/cpu/cpu.cfs_quota_us",
      cpu_period_file_name="/sys/fs/cgroup/cpu/cpu.cfs_period_us",
      cpuset_file_name="/sys/fs/cgroup/cpuset/cpuset.cpus",
      cpu_max_file_name="/sys/fs/cgroup/cpu.max",
  ) -> Optional[float]:
      # TODO (Alex): Don't implement this logic oursleves.
      # Docker has 2 underyling ways of implementing CPU limits:
      # https://docs.docker.com/config/containers/resource_constraints/#configure-the-default-cfs-scheduler
      # 1. --cpuset-cpus 2. --cpus or --cpu-quota/--cpu-period (--cpu-shares is a
      # soft limit so we don't worry about it). For Ray's purposes, if we use
      # docker, the number of vCPUs on a machine is whichever is set (ties broken
      # by smaller value).
  
      cpu_quota = None
      # See: https://bugs.openjdk.java.net/browse/JDK-8146115
      if os.path.exists(cpu_quota_file_name) and os.path.exists(cpu_period_file_name):
          try:
              with open(cpu_quota_file_name, "r") as quota_file, open(
                  cpu_period_file_name, "r"
              ) as period_file:
                  cpu_quota = float(quota_file.read()) / float(period_file.read())
          except Exception:
              logger.exception("Unexpected error calculating docker cpu quota.")
      # Look at cpu.max for cgroups v2
      elif os.path.exists(cpu_max_file_name):
          try:
              max_file = open(cpu_max_file_name).read()
              quota_str, period_str = max_file.split()
              if quota_str.isnumeric() and period_str.isnumeric():
                  cpu_quota = float(quota_str) / float(period_str)
              else:
                  # quota_str is "max" meaning the cpu quota is unset
                  cpu_quota = None
          except Exception:
              logger.exception("Unexpected error calculating docker cpu quota.")
      if (cpu_quota is not None) and (cpu_quota < 0):
          cpu_quota = None
      elif cpu_quota == 0:
          # Round up in case the cpu limit is less than 1.
          cpu_quota = 1
  
      cpuset_num = None
      if os.path.exists(cpuset_file_name):
          try:
              with open(cpuset_file_name) as cpuset_file:
                  ranges_as_string = cpuset_file.read()
                  ranges = ranges_as_string.split(",")
                  cpu_ids = []
                  for num_or_range in ranges:
                      if "-" in num_or_range:
                          start, end = num_or_range.split("-")
                          cpu_ids.extend(list(range(int(start), int(end) + 1)))
                      else:
                          cpu_ids.append(int(num_or_range))
                  cpuset_num = len(cpu_ids)
          except Exception:
              logger.exception("Unexpected error calculating docker cpuset ids.")
      # Possible to-do: Parse cgroups v2's cpuset.cpus.effective for the number
      # of accessible CPUs.
  
      if cpu_quota and cpuset_num:
          return min(cpu_quota, cpuset_num)
      return cpu_quota or cpuset_num
  
  
  def get_num_cpus() -> int:
      cpu_count = multiprocessing.cpu_count()
      if os.environ.get("RAY_USE_MULTIPROCESSING_CPU_COUNT"):
          logger.info(
              "Detected RAY_USE_MULTIPROCESSING_CPU_COUNT=1: Using "
              "multiprocessing.cpu_count() to detect the number of CPUs. "
              "This may be inconsistent when used inside docker. "
              "To correctly detect CPUs, unset the env var: "
              "`RAY_USE_MULTIPROCESSING_CPU_COUNT`."
          )
          return cpu_count
      try:
          # Not easy to get cpu count in docker, see:
          # https://bugs.python.org/issue36054
          docker_count = _get_docker_cpus()
          if docker_count is not None and docker_count != cpu_count:
              # Don't log this warning if we're on K8s or if the warning is
              # explicitly disabled.
              if (
                  "RAY_DISABLE_DOCKER_CPU_WARNING" not in os.environ
                  and "KUBERNETES_SERVICE_HOST" not in os.environ
              ):
                  logger.warning(
                      "Detecting docker specified CPUs. In "
                      "previous versions of Ray, CPU detection in containers "
                      "was incorrect. Please ensure that Ray has enough CPUs "
                      "allocated. As a temporary workaround to revert to the "
                      "prior behavior, set "
                      "`RAY_USE_MULTIPROCESSING_CPU_COUNT=1` as an env var "
                      "before starting Ray. Set the env var: "
                      "`RAY_DISABLE_DOCKER_CPU_WARNING=1` to mute this warning."
                  )
              # TODO (Alex): We should probably add support for fractional cpus.
              if int(docker_count) != float(docker_count):
                  logger.warning(
                      f"Ray currently does not support initializing Ray"
                      f"with fractional cpus. Your num_cpus will be "
                      f"truncated from {docker_count} to "
                      f"{int(docker_count)}."
                  )
              docker_count = int(docker_count)
              cpu_count = docker_count
  
      except Exception:
          # `nproc` and cgroup are linux-only. If docker only works on linux
          # (will run in a linux VM on other platforms), so this is fine.
          pass
  
      return cpu_count
  
  
  def get_used_memory():
      """Return the currently used system memory in bytes
  
      Returns:
          The total amount of used memory
      """
      # Try to accurately figure out the memory usage if we are in a docker
      # container.
      docker_usage = None
      # For cgroups v1:
      memory_usage_filename = "/sys/fs/cgroup/memory/memory.usage_in_bytes"
      # For cgroups v2:
      memory_usage_filename_v2 = "/sys/fs/cgroup/memory.current"
      if os.path.exists(memory_usage_filename):
          with open(memory_usage_filename, "r") as f:
              docker_usage = int(f.read())
      elif os.path.exists(memory_usage_filename_v2):
          with open(memory_usage_filename_v2, "r") as f:
              docker_usage = int(f.read())
  
      # Use psutil if it is available.
      psutil_memory_in_bytes = psutil.virtual_memory().used
  
      if docker_usage is not None:
          # We take the min because the cgroup limit is very large if we aren't
          # in Docker.
          return min(docker_usage, psutil_memory_in_bytes)
  
      return psutil_memory_in_bytes
  
  
  def estimate_available_memory():
      """Return the currently available amount of system memory in bytes.
  
      Returns:
          The total amount of available memory in bytes. Based on the used
          and total memory.
  
      """
      return get_system_memory() - get_used_memory()
  
  
  def get_shared_memory_bytes():
      """Get the size of the shared memory file system.
  
      Returns:
          The size of the shared memory file system in bytes.
      """
      # Make sure this is only called on Linux.
      assert sys.platform == "linux" or sys.platform == "linux2"
  
      shm_fd = os.open("/dev/shm", os.O_RDONLY)
      try:
          shm_fs_stats = os.fstatvfs(shm_fd)
          # The value shm_fs_stats.f_bsize is the block size and the
          # value shm_fs_stats.f_bavail is the number of available
          # blocks.
          shm_avail = shm_fs_stats.f_bsize * shm_fs_stats.f_bavail
      finally:
          os.close(shm_fd)
  
      return shm_avail
  
  
  def check_oversized_function(
      pickled: bytes, name: str, obj_type: str, worker: "ray.Worker"
  ) -> None:
      """Send a warning message if the pickled function is too large.
  
      Args:
          pickled: the pickled function.
          name: name of the pickled object.
          obj_type: type of the pickled object, can be 'function',
              'remote function', or 'actor'.
          worker: the worker used to send warning message. message will be logged
              locally if None.
      """
      length = len(pickled)
      if length <= ray_constants.FUNCTION_SIZE_WARN_THRESHOLD:
          return
      elif length < ray_constants.FUNCTION_SIZE_ERROR_THRESHOLD:
          warning_message = (
              "The {} {} is very large ({} MiB). "
              "Check that its definition is not implicitly capturing a large "
              "array or other object in scope. Tip: use ray.put() to put large "
              "objects in the Ray object store."
          ).format(obj_type, name, length // (1024 * 1024))
          if worker:
              push_error_to_driver(
                  worker,
                  ray_constants.PICKLING_LARGE_OBJECT_PUSH_ERROR,
                  "Warning: " + warning_message,
                  job_id=worker.current_job_id,
              )
      else:
          error = (
              "The {} {} is too large ({} MiB > FUNCTION_SIZE_ERROR_THRESHOLD={}"
              " MiB). Check that its definition is not implicitly capturing a "
              "large array or other object in scope. Tip: use ray.put() to "
              "put large objects in the Ray object store."
          ).format(
              obj_type,
              name,
              length // (1024 * 1024),
              ray_constants.FUNCTION_SIZE_ERROR_THRESHOLD // (1024 * 1024),
          )
          raise ValueError(error)
  
  
  def is_main_thread():
      return threading.current_thread().getName() == "MainThread"
  
  
  def detect_fate_sharing_support_win32():
      global win32_job, win32_AssignProcessToJobObject
      if win32_job is None and sys.platform == "win32":
          import ctypes
  
          try:
              from ctypes.wintypes import BOOL, DWORD, HANDLE, LPVOID, LPCWSTR
  
              kernel32 = ctypes.WinDLL("kernel32")
              kernel32.CreateJobObjectW.argtypes = (LPVOID, LPCWSTR)
              kernel32.CreateJobObjectW.restype = HANDLE
              sijo_argtypes = (HANDLE, ctypes.c_int, LPVOID, DWORD)
              kernel32.SetInformationJobObject.argtypes = sijo_argtypes
              kernel32.SetInformationJobObject.restype = BOOL
              kernel32.AssignProcessToJobObject.argtypes = (HANDLE, HANDLE)
              kernel32.AssignProcessToJobObject.restype = BOOL
              kernel32.IsDebuggerPresent.argtypes = ()
              kernel32.IsDebuggerPresent.restype = BOOL
          except (AttributeError, TypeError, ImportError):
              kernel32 = None
          job = kernel32.CreateJobObjectW(None, None) if kernel32 else None
          job = subprocess.Handle(job) if job else job
          if job:
              from ctypes.wintypes import DWORD, LARGE_INTEGER, ULARGE_INTEGER
  
              class JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure):
                  _fields_ = [
                      ("PerProcessUserTimeLimit", LARGE_INTEGER),
                      ("PerJobUserTimeLimit", LARGE_INTEGER),
                      ("LimitFlags", DWORD),
                      ("MinimumWorkingSetSize", ctypes.c_size_t),
                      ("MaximumWorkingSetSize", ctypes.c_size_t),
                      ("ActiveProcessLimit", DWORD),
                      ("Affinity", ctypes.c_size_t),
                      ("PriorityClass", DWORD),
                      ("SchedulingClass", DWORD),
                  ]
  
              class IO_COUNTERS(ctypes.Structure):
                  _fields_ = [
                      ("ReadOperationCount", ULARGE_INTEGER),
                      ("WriteOperationCount", ULARGE_INTEGER),
                      ("OtherOperationCount", ULARGE_INTEGER),
                      ("ReadTransferCount", ULARGE_INTEGER),
                      ("WriteTransferCount", ULARGE_INTEGER),
                      ("OtherTransferCount", ULARGE_INTEGER),
                  ]
  
              class JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure):
                  _fields_ = [
                      ("BasicLimitInformation", JOBOBJECT_BASIC_LIMIT_INFORMATION),
                      ("IoInfo", IO_COUNTERS),
                      ("ProcessMemoryLimit", ctypes.c_size_t),
                      ("JobMemoryLimit", ctypes.c_size_t),
                      ("PeakProcessMemoryUsed", ctypes.c_size_t),
                      ("PeakJobMemoryUsed", ctypes.c_size_t),
                  ]
  
              debug = kernel32.IsDebuggerPresent()
  
              # Defined in <WinNT.h>; also available here:
              # https://docs.microsoft.com/en-us/windows/win32/api/jobapi2/nf-jobapi2-setinformationjobobject
              JobObjectExtendedLimitInformation = 9
              JOB_OBJECT_LIMIT_BREAKAWAY_OK = 0x00000800
              JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION = 0x00000400
              JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000
              buf = JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
              buf.BasicLimitInformation.LimitFlags = (
                  (0 if debug else JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE)
                  | JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION
                  | JOB_OBJECT_LIMIT_BREAKAWAY_OK
              )
              infoclass = JobObjectExtendedLimitInformation
              if not kernel32.SetInformationJobObject(
                  job, infoclass, ctypes.byref(buf), ctypes.sizeof(buf)
              ):
                  job = None
          win32_AssignProcessToJobObject = (
              kernel32.AssignProcessToJobObject if kernel32 is not None else False
          )
          win32_job = job if job else False
      return bool(win32_job)
  
  
  def detect_fate_sharing_support_linux():
      global linux_prctl
      if linux_prctl is None and sys.platform.startswith("linux"):
          try:
              from ctypes import c_int, c_ulong, CDLL
  
              prctl = CDLL(None).prctl
              prctl.restype = c_int
              prctl.argtypes = [c_int, c_ulong, c_ulong, c_ulong, c_ulong]
          except (AttributeError, TypeError):
              prctl = None
          linux_prctl = prctl if prctl else False
      return bool(linux_prctl)
  
  
  def detect_fate_sharing_support():
      result = None
      if sys.platform == "win32":
          result = detect_fate_sharing_support_win32()
      elif sys.platform.startswith("linux"):
          result = detect_fate_sharing_support_linux()
      return result
  
  
  def set_kill_on_parent_death_linux():
      """Ensures this process dies if its parent dies (fate-sharing).
  
      Linux-only. Must be called in preexec_fn (i.e. by the child).
      """
      if detect_fate_sharing_support_linux():
          import signal
  
          PR_SET_PDEATHSIG = 1
          if linux_prctl(PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) != 0:
              import ctypes
  
              raise OSError(ctypes.get_errno(), "prctl(PR_SET_PDEATHSIG) failed")
      else:
          assert False, "PR_SET_PDEATHSIG used despite being unavailable"
  
  
  def set_kill_child_on_death_win32(child_proc):
      """Ensures the child process dies if this process dies (fate-sharing).
  
      Windows-only. Must be called by the parent, after spawning the child.
  
      Args:
          child_proc: The subprocess.Popen or subprocess.Handle object.
      """
  
      if isinstance(child_proc, subprocess.Popen):
          child_proc = child_proc._handle
      assert isinstance(child_proc, subprocess.Handle)
  
      if detect_fate_sharing_support_win32():
          if not win32_AssignProcessToJobObject(win32_job, int(child_proc)):
              import ctypes
  
              raise OSError(ctypes.get_last_error(), "AssignProcessToJobObject() failed")
      else:
          assert False, "AssignProcessToJobObject used despite being unavailable"
  
  
  def set_sigterm_handler(sigterm_handler):
      """Registers a handler for SIGTERM in a platform-compatible manner."""
      if sys.platform == "win32":
          # Note that these signal handlers only work for console applications.
          # TODO(mehrdadn): implement graceful process termination mechanism
          # SIGINT is Ctrl+C, SIGBREAK is Ctrl+Break.
          signal.signal(signal.SIGBREAK, sigterm_handler)
      else:
          signal.signal(signal.SIGTERM, sigterm_handler)
  
  
  def try_make_directory_shared(directory_path):
      try:
          os.chmod(directory_path, 0o0777)
      except OSError as e:
          # Silently suppress the PermissionError that is thrown by the chmod.
          # This is done because the user attempting to change the permissions
          # on a directory may not own it. The chmod is attempted whether the
          # directory is new or not to avoid race conditions.
          # ray-project/ray/#3591
          if e.errno in [errno.EACCES, errno.EPERM]:
              pass
          else:
              raise
  
  
  def try_to_create_directory(directory_path):
      """Attempt to create a directory that is globally readable/writable.
  
      Args:
          directory_path: The path of the directory to create.
      """
      directory_path = os.path.expanduser(directory_path)
      os.makedirs(directory_path, exist_ok=True)
      # Change the log directory permissions so others can use it. This is
      # important when multiple people are using the same machine.
      try_make_directory_shared(directory_path)
  
  
  def try_to_symlink(symlink_path, target_path):
      """Attempt to create a symlink.
  
      If the symlink path exists and isn't a symlink, the symlink will not be
      created. If a symlink exists in the path, it will be attempted to be
      removed and replaced.
  
      Args:
          symlink_path: The path at which to create the symlink.
          target_path: The path the symlink should point to.
      """
      symlink_path = os.path.expanduser(symlink_path)
      target_path = os.path.expanduser(target_path)
  
      if os.path.exists(symlink_path):
          if os.path.islink(symlink_path):
              # Try to remove existing symlink.
              try:
                  os.remove(symlink_path)
              except OSError:
                  return
          else:
              # There's an existing non-symlink file, don't overwrite it.
              return
  
      try:
          os.symlink(target_path, symlink_path)
      except OSError:
          return
  
  
  def get_user():
      if pwd is None:
          return ""
      try:
          return pwd.getpwuid(os.getuid()).pw_name
      except Exception:
          return ""
  
  
  def get_function_args(callable):
      all_parameters = frozenset(signature(callable).parameters)
      return list(all_parameters)
  
  
  def get_conda_bin_executable(executable_name):
      """
      Return path to the specified executable, assumed to be discoverable within
      the 'bin' subdirectory of a conda installation.  Adapted from
      https://github.com/mlflow/mlflow.
      """
  
      # Use CONDA_EXE as per https://github.com/conda/conda/issues/7126
      if "CONDA_EXE" in os.environ:
          conda_bin_dir = os.path.dirname(os.environ["CONDA_EXE"])
          return os.path.join(conda_bin_dir, executable_name)
      return executable_name
  
  
  def get_conda_env_dir(env_name):
      """Find and validate the conda directory for a given conda environment.
  
      For example, given the environment name `tf1`, this function checks
      the existence of the corresponding conda directory, e.g.
      `/Users/scaly/anaconda3/envs/tf1`, and returns it.
      """
      conda_prefix = os.environ.get("CONDA_PREFIX")
      if conda_prefix is None:
          # The caller is neither in a conda env or in (base) env.  This is rare
          # because by default, new terminals start in (base), but we can still
          # support this case.
          conda_exe = os.environ.get("CONDA_EXE")
          if conda_exe is None:
              raise ValueError(
                  "Cannot find environment variables set by conda. "
                  "Please verify conda is installed."
              )
          # Example: CONDA_EXE=$HOME/anaconda3/bin/python
          # Strip out /bin/python by going up two parent directories.
          conda_prefix = str(Path(conda_exe).parent.parent)
  
      # There are two cases:
      # 1. We are in a conda (base) env: CONDA_DEFAULT_ENV=base and
      #    CONDA_PREFIX=$HOME/anaconda3
      # 2. We are in a user-created conda env: CONDA_DEFAULT_ENV=$env_name and
      #    CONDA_PREFIX=$HOME/anaconda3/envs/$current_env_name
      if os.environ.get("CONDA_DEFAULT_ENV") == "base":
          # Caller's curent environment is (base).
          # Not recommended by conda, but we can still support it.
          if env_name == "base":
              # Desired environment is (base), located at e.g. $HOME/anaconda3
              env_dir = conda_prefix
          else:
              # Desired environment is user-created, e.g.
              # $HOME/anaconda3/envs/$env_name
              env_dir = os.path.join(conda_prefix, "envs", env_name)
      else:
          # Now `conda_prefix` should be something like
          # $HOME/anaconda3/envs/$current_env_name
          # We want to replace the last component with the desired env name.
          conda_envs_dir = os.path.split(conda_prefix)[0]
          env_dir = os.path.join(conda_envs_dir, env_name)
      if not os.path.isdir(env_dir):
          raise ValueError(
              "conda env "
              + env_name
              + " not found in conda envs directory. Run `conda env list` to "
              + "verify the name is correct."
          )
      return env_dir
  
  
  def get_call_location(back=1):
      """
      Get the location (filename and line number) of a function caller, `back`
      frames up the stack.
  
      Args:
          back (int): The number of frames to go up the stack, not including this
              function.
      """
      stack = inspect.stack()
      try:
          frame = stack[back + 1]
          return f"{frame.filename}:{frame.lineno}"
      except IndexError:
          return "UNKNOWN"
  
  
  # Used to only print a deprecation warning once for a given function if we
  # don't wish to spam the caller.
  _PRINTED_WARNING = set()
  
  
  # The following is inspired by
  # https://github.com/tensorflow/tensorflow/blob/dec8e0b11f4f87693b67e125e67dfbc68d26c205/tensorflow/python/util/deprecation.py#L274-L329
  def deprecated(
      instructions=None, removal_release=None, removal_date=None, warn_once=True
  ):
      """
      Creates a decorator for marking functions as deprecated. The decorator
      will log a deprecation warning on the first (or all, see `warn_once` arg)
      invocations, and will otherwise leave the wrapped function unchanged.
  
      Args:
          instructions (str): Instructions for the caller to update their code.
          removal_release (str): The release in which this deprecated function
              will be removed. Only one of removal_release and removal_date
              should be specified. If neither is specfieid, we'll warning that
              the function will be removed "in a future release".
          removal_date (str): The date on which this deprecated function will be
              removed. Only one of removal_release and removal_date should be
              specified. If neither is specfieid, we'll warning that
              the function will be removed "in a future release".
          warn_once (bool): If true, the deprecation warning will only be logged
              on the first invocation. Otherwise, the deprecation warning will
              be logged on every invocation. Defaults to True.
  
      Returns:
          A decorator to be used for wrapping deprecated functions.
      """
      if removal_release is not None and removal_date is not None:
          raise ValueError(
              "Only one of removal_release and removal_date should be specified."
          )
  
      def deprecated_wrapper(func):
          @functools.wraps(func)
          def new_func(*args, **kwargs):
              global _PRINTED_WARNING
              if func not in _PRINTED_WARNING:
                  if warn_once:
                      _PRINTED_WARNING.add(func)
                  msg = (
                      "From {}: {} (from {}) is deprecated and will ".format(
                          get_call_location(), func.__name__, func.__module__
                      )
                      + "be removed "
                      + (
                          f"in version {removal_release}."
                          if removal_release is not None
                          else f"after {removal_date}"
                          if removal_date is not None
                          else "in a future version"
                      )
                      + (f" {instructions}" if instructions is not None else "")
                  )
                  warnings.warn(msg)
              return func(*args, **kwargs)
  
          return new_func
  
      return deprecated_wrapper
  
  
  def import_attr(full_path: str):
      """Given a full import path to a module attr, return the imported attr.
  
      For example, the following are equivalent:
          MyClass = import_attr("module.submodule:MyClass")
          MyClass = import_attr("module.submodule.MyClass")
          from module.submodule import MyClass
  
      Returns:
          Imported attr
      """
      if full_path is None:
          raise TypeError("import path cannot be None")
  
      if ":" in full_path:
          if full_path.count(":") > 1:
              raise ValueError(
                  f'Got invalid import path "{full_path}". An '
                  "import path may have at most one colon."
              )
          module_name, attr_name = full_path.split(":")
      else:
          last_period_idx = full_path.rfind(".")
          module_name = full_path[:last_period_idx]
          attr_name = full_path[last_period_idx + 1 :]
  
      module = importlib.import_module(module_name)
      return getattr(module, attr_name)
  
  
  def get_wheel_filename(
      sys_platform: str = sys.platform,
      ray_version: str = ray.__version__,
      py_version: str = f"{sys.version_info.major}{sys.version_info.minor}",
  ) -> str:
      """Returns the filename used for the nightly Ray wheel.
  
      Args:
          sys_platform (str): The platform as returned by sys.platform. Examples:
              "darwin", "linux", "win32"
          ray_version (str): The Ray version as returned by ray.__version__ or
              `ray --version`.  Examples: "2.0.0.dev0"
          py_version (str):
              The major and minor Python versions concatenated.  Examples: "36",
              "37", "38", "39"
      Returns:
          The wheel file name.  Examples:
              ray-2.0.0.dev0-cp38-cp38-manylinux2014_x86_64.whl
      """
      assert py_version in ["36", "37", "38", "39"], py_version
  
      os_strings = {
          "darwin": "macosx_10_15_x86_64"
          if py_version in ["38", "39"]
          else "macosx_10_15_intel",
          "linux": "manylinux2014_x86_64",
          "win32": "win_amd64",
      }
  
      assert sys_platform in os_strings, sys_platform
  
      wheel_filename = (
          f"ray-{ray_version}-cp{py_version}-"
          f"cp{py_version}{'m' if py_version in ['36', '37'] else ''}"
          f"-{os_strings[sys_platform]}.whl"
      )
  
      return wheel_filename
  
  
  def get_master_wheel_url(
      ray_commit: str = ray.__commit__,
      sys_platform: str = sys.platform,
      ray_version: str = ray.__version__,
      py_version: str = f"{sys.version_info.major}{sys.version_info.minor}",
  ) -> str:
      """Return the URL for the wheel from a specific commit."""
      filename = get_wheel_filename(
          sys_platform=sys_platform, ray_version=ray_version, py_version=py_version
      )
      return (
          f"https://s3-us-west-2.amazonaws.com/ray-wheels/master/"
          f"{ray_commit}/{filename}"
      )
  
  
  def get_release_wheel_url(
      ray_commit: str = ray.__commit__,
      sys_platform: str = sys.platform,
      ray_version: str = ray.__version__,
      py_version: str = f"{sys.version_info.major}{sys.version_info.minor}",
  ) -> str:
      """Return the URL for the wheel for a specific release."""
      filename = get_wheel_filename(
          sys_platform=sys_platform, ray_version=ray_version, py_version=py_version
      )
      return (
          f"https://ray-wheels.s3-us-west-2.amazonaws.com/releases/"
          f"{ray_version}/{ray_commit}/{filename}"
      )
      # e.g. https://ray-wheels.s3-us-west-2.amazonaws.com/releases/1.4.0rc1/e7c7
      # f6371a69eb727fa469e4cd6f4fbefd143b4c/ray-1.4.0rc1-cp36-cp36m-manylinux201
      # 4_x86_64.whl
  
  
  def validate_namespace(namespace: str):
      if not isinstance(namespace, str):
          raise TypeError("namespace must be None or a string.")
      elif namespace == "":
          raise ValueError(
              '"" is not a valid namespace. ' "Pass None to not specify a namespace."
          )
  
  
  def init_grpc_channel(
      address: str,
      options: Optional[Sequence[Tuple[str, Any]]] = None,
      asynchronous: bool = False,
  ):
      grpc_module = aiogrpc if asynchronous else grpc
      if os.environ.get("RAY_USE_TLS", "0").lower() in ("1", "true"):
          server_cert_chain, private_key, ca_cert = load_certs_from_env()
          credentials = grpc.ssl_channel_credentials(
              certificate_chain=server_cert_chain,
              private_key=private_key,
              root_certificates=ca_cert,
          )
          channel = grpc_module.secure_channel(address, credentials, options=options)
      else:
          channel = grpc_module.insecure_channel(address, options=options)
  
      return channel
  
  
  def check_dashboard_dependencies_installed() -> bool:
      """Returns True if Ray Dashboard dependencies are installed.
  
      Checks to see if we should start the dashboard agent or not based on the
      Ray installation version the user has installed (ray vs. ray[default]).
      Unfortunately there doesn't seem to be a cleaner way to detect this other
      than just blindly importing the relevant packages.
  
      """
      try:
          import ray.dashboard.optional_deps  # noqa: F401
  
          return True
      except ImportError:
          return False
  
  
  def internal_kv_list_with_retry(gcs_client, prefix, namespace, num_retries=20):
      result = None
      if isinstance(prefix, str):
          prefix = prefix.encode()
      if isinstance(namespace, str):
          namespace = namespace.encode()
      for _ in range(num_retries):
          try:
              result = gcs_client.internal_kv_keys(prefix, namespace)
          except Exception as e:
              if isinstance(e, grpc.RpcError) and e.code() in (
                  grpc.StatusCode.UNAVAILABLE,
                  grpc.StatusCode.UNKNOWN,
              ):
                  logger.warning(
                      f"Unable to connect to GCS at {gcs_client.address}. "
                      "Check that (1) Ray GCS with matching version started "
                      "successfully at the specified address, and (2) there is "
                      "no firewall setting preventing access."
                  )
              else:
                  logger.exception("Internal KV List failed")
              result = None
  
          if result is not None:
              break
          else:
              logger.debug(f"Fetched {prefix}=None from KV. Retrying.")
              time.sleep(2)
      if result is None:
          raise RuntimeError(
              f"Could not list '{prefix}' from GCS. Did GCS start successfully?"
          )
      return result
  
  
  def internal_kv_get_with_retry(gcs_client, key, namespace, num_retries=20):
      result = None
      if isinstance(key, str):
          key = key.encode()
      for _ in range(num_retries):
          try:
              result = gcs_client.internal_kv_get(key, namespace)
          except Exception as e:
              if isinstance(e, grpc.RpcError) and e.code() in (
                  grpc.StatusCode.UNAVAILABLE,
                  grpc.StatusCode.UNKNOWN,
              ):
                  logger.warning(
                      f"Unable to connect to GCS at {gcs_client.address}. "
                      "Check that (1) Ray GCS with matching version started "
                      "successfully at the specified address, and (2) there is "
                      "no firewall setting preventing access."
                  )
              else:
                  logger.exception("Internal KV Get failed")
              result = None
  
          if result is not None:
              break
          else:
              logger.debug(f"Fetched {key}=None from KV. Retrying.")
              time.sleep(2)
      if not result:
          raise RuntimeError(
              f"Could not read '{key.decode()}' from GCS. Did GCS start successfully?"
          )
      return result
  
  
  def internal_kv_put_with_retry(gcs_client, key, value, namespace, num_retries=20):
      if isinstance(key, str):
          key = key.encode()
      if isinstance(value, str):
          value = value.encode()
      if isinstance(namespace, str):
          namespace = namespace.encode()
      error = None
      for _ in range(num_retries):
          try:
              return gcs_client.internal_kv_put(
                  key, value, overwrite=True, namespace=namespace
              )
          except grpc.RpcError as e:
              if e.code() in (
                  grpc.StatusCode.UNAVAILABLE,
                  grpc.StatusCode.UNKNOWN,
              ):
                  logger.warning(
                      f"Unable to connect to GCS at {gcs_client.address}. "
                      "Check that (1) Ray GCS with matching version started "
                      "successfully at the specified address, and (2) there is "
                      "no firewall setting preventing access."
                  )
              else:
                  logger.exception("Internal KV Put failed")
              time.sleep(2)
              error = e
      # Reraise the last grpc.RpcError.
      raise error
  
  
  def compute_version_info():
      """Compute the versions of Python, and Ray.
  
      Returns:
          A tuple containing the version information.
      """
      ray_version = ray.__version__
      python_version = ".".join(map(str, sys.version_info[:3]))
      return ray_version, python_version
  
  
  def get_directory_size_bytes(path: Union[str, Path] = ".") -> int:
      """Get the total size of a directory in bytes, including subdirectories."""
      total_size_bytes = 0
      for dirpath, dirnames, filenames in os.walk(path):
          for f in filenames:
              fp = os.path.join(dirpath, f)
              # skip if it is a symbolic link or a .pyc file
              if not os.path.islink(fp) and not f.endswith(".pyc"):
                  total_size_bytes += os.path.getsize(fp)
  
      return total_size_bytes
  
  
  def check_version_info(cluster_metadata):
      """Check if the Python and Ray versions stored in GCS matches this process.
      Args:
          cluster_metadata: Ray cluster metadata from GCS.
  
      Raises:
          Exception: An exception is raised if there is a version mismatch.
      """
      cluster_version_info = (
          cluster_metadata["ray_version"],
          cluster_metadata["python_version"],
      )
      version_info = compute_version_info()
      if version_info != cluster_version_info:
          node_ip_address = ray._private.services.get_node_ip_address()
          error_message = (
              "Version mismatch: The cluster was started with:\n"
              "    Ray: " + cluster_version_info[0] + "\n"
              "    Python: " + cluster_version_info[1] + "\n"
              "This process on node " + node_ip_address + " was started with:" + "\n"
              "    Ray: " + version_info[0] + "\n"
              "    Python: " + version_info[1] + "\n"
          )
          raise RuntimeError(error_message)
  