image-match/python/py/Lib/site-packages/ray/_private/utils.py

import contextlib
import importlib
import json
import logging
import multiprocessing
import os
import platform
import re
import signal
import subprocess
import sys
import threading
import time
from collections import defaultdict
from pathlib import Path
from subprocess import list2cmdline
from typing import (
    TYPE_CHECKING,
    Dict,
    List,
    Mapping,
    Optional,
    Tuple,
    Union,
)

from google.protobuf import json_format

import ray
import ray._private.ray_constants as ray_constants
from ray._common.utils import (
    PLACEMENT_GROUP_BUNDLE_RESOURCE_NAME,
    get_ray_address_file,
    get_system_memory,
)
from ray.core.generated.runtime_environment_pb2 import (
    RuntimeEnvInfo as ProtoRuntimeEnvInfo,
)

# Import psutil after ray so the packaged version is used.
import psutil

if TYPE_CHECKING:
    from ray.runtime_env import RuntimeEnv


INT32_MAX = (2**31) - 1


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

ENV_DISABLE_DOCKER_CPU_WARNING = "RAY_DISABLE_DOCKER_CPU_WARNING" in os.environ

# This global variable is used for testing only
_CALLED_FREQ = defaultdict(lambda: 0)
_CALLED_FREQ_LOCK = threading.Lock()

PLACEMENT_GROUP_INDEXED_BUNDLED_RESOURCE_PATTERN = re.compile(
    r"(.+)_group_(\d+)_([0-9a-zA-Z]+)"
)
PLACEMENT_GROUP_WILDCARD_RESOURCE_PATTERN = re.compile(r"(.+)_group_([0-9a-zA-Z]+)")


def write_ray_address(ray_address: str, temp_dir: Optional[str] = None):
    address_file = get_ray_address_file(temp_dir)
    if os.path.exists(address_file):
        with open(address_file, "r") as f:
            prev_address = f.read()
        if prev_address == ray_address:
            return

        logger.info(
            f"Overwriting previous Ray address ({prev_address}). "
            "Running ray.init() on this node will now connect to the new "
            f"instance at {ray_address}. To override this behavior, pass "
            f"address={prev_address} to ray.init()."
        )

    with open(address_file, "w+") as f:
        f.write(ray_address)


def read_ray_address(temp_dir: Optional[str] = None) -> str:
    address_file = get_ray_address_file(temp_dir)
    if not os.path.exists(address_file):
        return None
    with open(address_file, "r") as f:
        return f.read().strip()


def format_error_message(exception_message: str, task_exception: bool = 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: 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: str, message: str, job_id: Optional[str] = None
):
    """Push an error message to the driver to be printed in the background.

    Args:
        worker: The worker to use.
        error_type: The type of the error.
        message: 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 publish_error_to_driver(
    error_type: str,
    message: str,
    gcs_client,
    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: The type of the error.
        message: The message that will be printed in the background
            on the driver.
        gcs_client: The GCS client 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)
    try:
        gcs_client.publish_error(job_id.hex().encode(), error_type, message, job_id, 60)
    except Exception:
        logger.exception(f"Failed to publish error: {message} [type {error_type}]")


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), f"Expected str or bytes, got {type(s)}"
        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)


# 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_visible_accelerator_ids() -> Mapping[str, Optional[List[str]]]:
    """Get the mapping from accelerator resource name
    to the visible ids."""

    from ray._private.accelerators import (
        get_accelerator_manager_for_resource,
        get_all_accelerator_resource_names,
    )

    return {
        accelerator_resource_name: get_accelerator_manager_for_resource(
            accelerator_resource_name
        ).get_current_process_visible_accelerator_ids()
        for accelerator_resource_name in get_all_accelerator_resource_names()
    }


def set_omp_num_threads_if_unset() -> bool:
    """Set the OMP_NUM_THREADS to default to num cpus assigned to the worker

    This function sets the environment variable OMP_NUM_THREADS for the worker,
    if the env is not previously set and it's running in worker (WORKER_MODE).

    Returns True if OMP_NUM_THREADS is set in this function.

    """
    num_threads_from_env = os.environ.get("OMP_NUM_THREADS")
    if num_threads_from_env is not None:
        # No ops if it's set
        return False

    # If unset, try setting the correct CPU count assigned.
    runtime_ctx = ray.get_runtime_context()
    if runtime_ctx.worker.mode != ray._private.worker.WORKER_MODE:
        # Non worker mode, no ops.
        return False

    num_assigned_cpus = runtime_ctx.get_assigned_resources().get("CPU")

    if num_assigned_cpus is None:
        # This is an actor task w/o any num_cpus specified, set it to 1
        logger.debug(
            "[ray] Forcing OMP_NUM_THREADS=1 to avoid performance "
            "degradation with many workers (issue #6998). You can override this "
            "by explicitly setting OMP_NUM_THREADS, or changing num_cpus."
        )
        num_assigned_cpus = 1

    import math

    # For num_cpu < 1: Set to 1.
    # For num_cpus >= 1: Set to the floor of the actual assigned cpus.
    omp_num_threads = max(math.floor(num_assigned_cpus), 1)
    os.environ["OMP_NUM_THREADS"] = str(omp_num_threads)
    return True


def set_visible_accelerator_ids() -> Mapping[str, Optional[str]]:
    """Set (CUDA_VISIBLE_DEVICES, ONEAPI_DEVICE_SELECTOR, HIP_VISIBLE_DEVICES,
    NEURON_RT_VISIBLE_CORES, TPU_VISIBLE_CHIPS , HABANA_VISIBLE_MODULES ,...)
    environment variables based on the accelerator runtime. Return the original
    environment variables.
    """
    from ray._private.ray_constants import env_bool

    original_visible_accelerator_env_vars = {}
    override_on_zero = env_bool(
        ray._private.accelerators.RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO_ENV_VAR,
        True,
    )
    for resource_name, accelerator_ids in (
        ray.get_runtime_context().get_accelerator_ids().items()
    ):
        # If no accelerator ids are set, skip overriding the environment variable.
        if not override_on_zero and len(accelerator_ids) == 0:
            continue
        env_var = ray._private.accelerators.get_accelerator_manager_for_resource(
            resource_name
        ).get_visible_accelerator_ids_env_var()
        original_visible_accelerator_env_vars[env_var] = os.environ.get(env_var, None)
        ray._private.accelerators.get_accelerator_manager_for_resource(
            resource_name
        ).set_current_process_visible_accelerator_ids(accelerator_ids)

    return original_visible_accelerator_env_vars


def reset_visible_accelerator_env_vars(
    original_visible_accelerator_env_vars: Mapping[str, Optional[str]]
) -> None:
    """Reset the visible accelerator env vars to the original values."""
    for env_var, env_value in original_visible_accelerator_env_vars.items():
        if env_value is None:
            os.environ.pop(env_var, None)
        else:
            os.environ[env_var] = env_value


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):
        # Avoid endless loop when get `stream` attribute
        if attr == "stream":
            return super().__getattribute__("stream")
        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_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(
    override_docker_cpu_warning: bool = ENV_DISABLE_DOCKER_CPU_WARNING,
    truncate: bool = True,
) -> float:
    """
    Get the number of CPUs available on this node.
    Depending on the situation, use multiprocessing.cpu_count() or cgroups.

    Args:
        override_docker_cpu_warning: An extra flag to explicitly turn off the Docker
            warning. Setting this flag True has the same effect as setting the env
            RAY_DISABLE_DOCKER_CPU_WARNING. By default, whether or not to log
            the warning is determined by the env variable
            RAY_DISABLE_DOCKER_CPU_WARNING.
        truncate: truncates the return value and drops the decimal part.
    """
    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 (
                "KUBERNETES_SERVICE_HOST" not in os.environ
                and not ENV_DISABLE_DOCKER_CPU_WARNING
                and not override_docker_cpu_warning
            ):
                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)}."
                )
            if truncate:
                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


# TODO(clarng): merge code with c++
def get_cgroup_used_memory(
    memory_stat_filename: str,
    memory_usage_filename: str,
    inactive_file_key: str,
    active_file_key: str,
):
    """
    The calculation logic is the same with `GetCGroupMemoryUsedBytes`
    in `memory_monitor.cc` file.
    """
    inactive_file_bytes = -1
    active_file_bytes = -1
    with open(memory_stat_filename, "r") as f:
        lines = f.readlines()
        for line in lines:
            if f"{inactive_file_key} " in line:
                inactive_file_bytes = int(line.split()[1])
            elif f"{active_file_key} " in line:
                active_file_bytes = int(line.split()[1])

    with open(memory_usage_filename, "r") as f:
        lines = f.readlines()
        cgroup_usage_in_bytes = int(lines[0].strip())

    if (
        inactive_file_bytes == -1
        or cgroup_usage_in_bytes == -1
        or active_file_bytes == -1
    ):
        return None

    return cgroup_usage_in_bytes - inactive_file_bytes - active_file_bytes


def resolve_object_store_memory(
    available_memory_bytes: int,
    object_store_memory: Optional[int] = None,
) -> int:
    """Resolve the object store memory size.

    This function determines the appropriate object store memory size based on
    the provided value or calculates a default based on available system memory.

    Args:
        available_memory_bytes: The memory available for this node.
        object_store_memory: The user-specified object store memory size in bytes.
            If None, a default size will be calculated.

    Returns:
        The resolved object store memory size in bytes.
    """
    # Derive default object store memory if not specified
    if object_store_memory is None:
        object_store_memory_cap = ray_constants.DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES

        # Cap by shm size by default to avoid low performance, but don't
        # go lower than REQUIRE_SHM_SIZE_THRESHOLD.
        if sys.platform == "linux" or sys.platform == "linux2":
            # Multiple by 0.95 to give a bit of wiggle-room.
            # https://github.com/ray-project/ray/pull/23034/files
            shm_avail = get_shared_memory_bytes() * 0.95
            shm_cap = max(ray_constants.REQUIRE_SHM_SIZE_THRESHOLD, shm_avail)

            object_store_memory_cap = min(object_store_memory_cap, shm_cap)

        object_store_memory = int(
            available_memory_bytes
            * ray_constants.DEFAULT_OBJECT_STORE_MEMORY_PROPORTION
        )

        # Set the object_store_memory size to 2GB on Mac
        # to avoid degraded performance.
        # (https://github.com/ray-project/ray/issues/20388)
        if sys.platform == "darwin":
            object_store_memory = min(
                object_store_memory, ray_constants.MAC_DEGRADED_PERF_MMAP_SIZE_LIMIT
            )

        # Cap memory to avoid memory waste and perf issues on large nodes
        if object_store_memory > object_store_memory_cap:
            logger.debug(
                "Warning: Capping object memory store to {}GB. ".format(
                    object_store_memory_cap // 1e9
                )
                + "To increase this further, specify `object_store_memory` "
                "when calling ray.init() or ray start."
            )
            object_store_memory = object_store_memory_cap

    return object_store_memory


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_v1 = "/sys/fs/cgroup/memory/memory.usage_in_bytes"
    memory_stat_filename_v1 = "/sys/fs/cgroup/memory/memory.stat"
    # For cgroups v2:
    memory_usage_filename_v2 = "/sys/fs/cgroup/memory.current"
    memory_stat_filename_v2 = "/sys/fs/cgroup/memory.stat"
    if os.path.exists(memory_usage_filename_v1) and os.path.exists(
        memory_stat_filename_v1
    ):
        docker_usage = get_cgroup_used_memory(
            memory_stat_filename_v1,
            memory_usage_filename_v1,
            "total_inactive_file",
            "total_active_file",
        )
    elif os.path.exists(memory_usage_filename_v2) and os.path.exists(
        memory_stat_filename_v2
    ):
        docker_usage = get_cgroup_used_memory(
            memory_stat_filename_v2,
            memory_usage_filename_v2,
            "inactive_file",
            "active_file",
        )

    if docker_usage is not None:
        return docker_usage
    return psutil.virtual_memory().used


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, LPCWSTR, LPVOID

            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 CDLL, c_int, c_ulong

            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_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_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_ray_doc_version():
    """Get the docs.ray.io version corresponding to the ray.__version__."""
    # The ray.__version__ can be official Ray release (such as 1.12.0), or
    # dev (3.0.0dev0) or release candidate (2.0.0rc0). For the later we map
    # to the master doc version at docs.ray.io.
    if re.match(r"^\d+\.\d+\.\d+$", ray.__version__) is None:
        return "master"
    # For the former (official Ray release), we have corresponding doc version
    # released as well.
    return f"releases-{ray.__version__}"


# Used to only print a deprecation warning once for a given function if we
# don't wish to spam the caller.
def get_wheel_filename(
    sys_platform: str = sys.platform,
    ray_version: str = ray.__version__,
    py_version: Tuple[int, int] = (sys.version_info.major, sys.version_info.minor),
    architecture: Optional[str] = None,
) -> str:
    """Returns the filename used for the nightly Ray wheel.

    Args:
        sys_platform: The platform as returned by sys.platform. Examples:
            "darwin", "linux", "win32"
        ray_version: The Ray version as returned by ray.__version__ or
            `ray --version`.  Examples: "3.0.0.dev0"
        py_version: The Python version as returned by sys.version_info. A
            tuple of (major, minor). Examples: (3, 8)
        architecture: Architecture, e.g. ``x86_64`` or ``aarch64``. If None, will
            be determined by calling ``platform.processor()``.

    Returns:
        The wheel file name.  Examples:
            ray-3.0.0.dev0-cp38-cp38-manylinux2014_x86_64.whl
    """
    assert py_version in ray_constants.RUNTIME_ENV_CONDA_PY_VERSIONS, py_version

    py_version_str = "".join(map(str, py_version))

    architecture = architecture or platform.processor()

    assert sys_platform in ["darwin", "linux", "win32"], sys_platform

    if sys_platform == "darwin":
        if architecture == "x86_64":
            os_string = "macosx_12_0_x86_64"
        else:
            os_string = "macosx_12_0_arm64"
    elif sys_platform == "linux":
        if architecture == "aarch64" or architecture == "arm64":
            os_string = "manylinux2014_aarch64"
        else:
            os_string = "manylinux2014_x86_64"
    elif sys_platform == "win32":
        os_string = "win_amd64"

    wheel_filename = (
        f"ray-{ray_version}-cp{py_version_str}-"
        f"cp{py_version_str}{'m' if py_version_str in ['37'] else ''}"
        f"-{os_string}.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: Tuple[int, int] = sys.version_info[:2],
) -> 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: Tuple[int, int] = sys.version_info[:2],
) -> 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 get_dashboard_dependency_error() -> Optional[ImportError]:
    """Returns the exception error if Ray Dashboard dependencies are not installed.
    None if they 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 None
    except ImportError as e:
        return e


def get_ray_client_dependency_error() -> Optional[ImportError]:
    """Returns the exception error if Ray Client dependencies are not installed.
    None if they are installed.
    """
    try:
        import grpc  # noqa: F401

        return None
    except ImportError as e:
        return e


connect_error = (
    "Unable to connect to GCS (ray head) at {}. "
    "Check that (1) Ray with matching version started "
    "successfully at the specified address, (2) this "
    "node can reach the specified address, and (3) there is "
    "no firewall setting preventing access."
)


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, ray.exceptions.RpcError) and e.rpc_code in (
                ray._raylet.GRPC_STATUS_CODE_UNAVAILABLE,
                ray._raylet.GRPC_STATUS_CODE_UNKNOWN,
            ):
                logger.warning(connect_error.format(gcs_client.address))
            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 ConnectionError(
            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, ray.exceptions.RpcError) and e.rpc_code in (
                ray._raylet.GRPC_STATUS_CODE_UNAVAILABLE,
                ray._raylet.GRPC_STATUS_CODE_UNKNOWN,
            ):
                logger.warning(connect_error.format(gcs_client.address))
            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 ConnectionError(
            f"Could not read '{key.decode()}' from GCS. Did GCS start successfully?"
        )
    return result


def parse_resources_json(
    resources: str, cli_logger, cf, command_arg="--resources"
) -> Dict[str, float]:
    try:
        resources = json.loads(resources)
        if not isinstance(resources, dict):
            raise ValueError("The format after deserialization is not a dict")
    except Exception as e:
        cli_logger.error(
            "`{}` is not a valid JSON string, detail error:{}",
            cf.bold(f"{command_arg}={resources}"),
            str(e),
        )
        cli_logger.abort(
            "Valid values look like this: `{}`",
            cf.bold(
                f'{command_arg}=\'{{"CustomResource3": 1, "CustomResource2": 2}}\''
            ),
        )
    return resources


def parse_metadata_json(
    metadata: str, cli_logger, cf, command_arg="--metadata-json"
) -> Dict[str, str]:
    try:
        metadata = json.loads(metadata)
        if not isinstance(metadata, dict):
            raise ValueError("The format after deserialization is not a dict")
    except Exception as e:
        cli_logger.error(
            "`{}` is not a valid JSON string, detail error:{}",
            cf.bold(f"{command_arg}={metadata}"),
            str(e),
        )
        cli_logger.abort(
            "Valid values look like this: `{}`",
            cf.bold(f'{command_arg}=\'{{"key1": "value1", "key2": "value2"}}\''),
        )
    return metadata


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 ray.exceptions.RpcError as e:
            if e.rpc_code in (
                ray._raylet.GRPC_STATUS_CODE_UNAVAILABLE,
                ray._raylet.GRPC_STATUS_CODE_UNKNOWN,
            ):
                logger.warning(connect_error.format(gcs_client.address))
            else:
                logger.exception("Internal KV Put failed")
            time.sleep(2)
            error = e
    # Reraise the last error.
    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,
    this_process_address,
    raise_on_mismatch=True,
    python_version_match_level=None,
):
    """Check if the Python and Ray versions stored in GCS matches this process.
    Args:
        cluster_metadata: Ray cluster metadata from GCS.
        this_process_address: Informational only. The address of this process.
            e.g. "node address:port" or "Ray Client".
        raise_on_mismatch: Raise an exception on True, log a warning otherwise.
        python_version_match_level: "minor" or "patch". To which python version level we
            try to match. Note if "minor" and the patch is different, we will still log
            a warning. Default value is `RAY_DEFAULT_PYTHON_VERSION_MATCH_LEVEL` if it
            exists, otherwise "patch"

    Behavior:
        - We raise or log a warning, based on raise_on_mismatch, if:
            - Ray versions do not match; OR
            - Python (major, minor) versions do not match,
                if python_version_match_level == 'minor'; OR
            - Python (major, minor, patch) versions do not match,
                if python_version_match_level == 'patch'.
        - We also log a warning if:
            - Python (major, minor) versions match, AND
            - Python patch versions do not match, AND
            - python_version_match_level == 'minor' AND
            - raise_on_mismatch == False.

    Raises:
        Exception: An exception is raised if there is a version mismatch.
    """
    if python_version_match_level is None:
        python_version_match_level = os.environ.get(
            "RAY_DEFAULT_PYTHON_VERSION_MATCH_LEVEL", "patch"
        )

    cluster_version_info = (
        cluster_metadata["ray_version"],
        cluster_metadata["python_version"],
    )
    my_version_info = compute_version_info()

    # Calculate: ray_matches, python_matches, python_full_matches
    ray_matches = cluster_version_info[0] == my_version_info[0]
    python_full_matches = cluster_version_info[1] == my_version_info[1]
    if python_version_match_level == "patch":
        python_matches = cluster_version_info[1] == my_version_info[1]
    elif python_version_match_level == "minor":
        my_python_versions = my_version_info[1].split(".")
        cluster_python_versions = cluster_version_info[1].split(".")
        python_matches = my_python_versions[:2] == cluster_python_versions[:2]
    else:
        raise ValueError(
            f"Invalid python_version_match_level: {python_version_match_level}, "
            "want: 'minor' or 'patch'"
        )

    mismatch_msg = (
        "The cluster was started with:\n"
        f"    Ray: {cluster_version_info[0]}\n"
        f"    Python: {cluster_version_info[1]}\n"
        f"This process on {this_process_address} was started with:\n"
        f"    Ray: {my_version_info[0]}\n"
        f"    Python: {my_version_info[1]}\n"
    )

    if ray_matches and python_matches:
        if not python_full_matches:
            logger.warning(f"Python patch version mismatch: {mismatch_msg}")
    else:
        error_message = f"Version mismatch: {mismatch_msg}"
        if raise_on_mismatch:
            raise RuntimeError(error_message)
        else:
            logger.warning(error_message)


def get_runtime_env_info(
    runtime_env: "RuntimeEnv",
    *,
    is_job_runtime_env: bool = False,
    serialize: bool = False,
):
    """Create runtime env info from runtime env.

    In the user interface, the argument `runtime_env` contains some fields
    which not contained in `ProtoRuntimeEnv` but in `ProtoRuntimeEnvInfo`,
    such as `eager_install`. This function will extract those fields from
    `RuntimeEnv` and create a new `ProtoRuntimeEnvInfo`, and serialize it
    into json format.
    """
    from ray.runtime_env import RuntimeEnvConfig

    proto_runtime_env_info = ProtoRuntimeEnvInfo()

    if runtime_env.working_dir_uri():
        proto_runtime_env_info.uris.working_dir_uri = runtime_env.working_dir_uri()
    if len(runtime_env.py_modules_uris()) > 0:
        proto_runtime_env_info.uris.py_modules_uris[:] = runtime_env.py_modules_uris()

    # TODO(Catch-Bull): overload `__setitem__` for `RuntimeEnv`, change the
    # runtime_env of all internal code from dict to RuntimeEnv.

    runtime_env_config = runtime_env.get("config")
    if runtime_env_config is None:
        runtime_env_config = RuntimeEnvConfig.default_config()
    else:
        runtime_env_config = RuntimeEnvConfig.parse_and_validate_runtime_env_config(
            runtime_env_config
        )

    proto_runtime_env_info.runtime_env_config.CopyFrom(
        runtime_env_config.build_proto_runtime_env_config()
    )

    # Normally, `RuntimeEnv` should guarantee the accuracy of field eager_install,
    # but so far, the internal code has not completely prohibited direct
    # modification of fields in RuntimeEnv, so we should check it for insurance.
    eager_install = (
        runtime_env_config.get("eager_install")
        if runtime_env_config is not None
        else None
    )
    if is_job_runtime_env or eager_install is not None:
        if eager_install is None:
            eager_install = True
        elif not isinstance(eager_install, bool):
            raise TypeError(
                f"eager_install must be a boolean. got {type(eager_install)}"
            )
        proto_runtime_env_info.runtime_env_config.eager_install = eager_install

    proto_runtime_env_info.serialized_runtime_env = runtime_env.serialize()

    if not serialize:
        return proto_runtime_env_info

    return json_format.MessageToJson(proto_runtime_env_info)


def parse_runtime_env_for_task_or_actor(
    runtime_env: Optional[Union[Dict, "RuntimeEnv"]]
):
    from ray.runtime_env import RuntimeEnv
    from ray.runtime_env.runtime_env import _validate_no_local_paths

    # Parse local pip/conda config files here. If we instead did it in
    # .remote(), it would get run in the Ray Client server, which runs on
    # a remote node where the files aren't available.
    if runtime_env:
        if isinstance(runtime_env, dict):
            runtime_env = RuntimeEnv(**(runtime_env or {}))
            _validate_no_local_paths(runtime_env)
            return runtime_env
        raise TypeError(
            "runtime_env must be dict or RuntimeEnv, ",
            f"but got: {type(runtime_env)}",
        )
    else:
        # Keep the new_runtime_env as None.  In .remote(), we need to know
        # if runtime_env is None to know whether or not to fall back to the
        # runtime_env specified in the @ray.remote decorator.
        return None


def split_address(address: str) -> Tuple[str, str]:
    """Splits address into a module string (scheme) and an inner_address.

    We use a custom splitting function instead of urllib because
    PEP allows "underscores" in a module names, while URL schemes do not
    allow them.

    Args:
        address: The address to split.

    Returns:
        A tuple of (scheme, inner_address).

    Raises:
        ValueError: If the address does not contain '://'.

    Examples:
        >>> split_address("ray://my_cluster")
        ('ray', 'my_cluster')
    """
    if "://" not in address:
        raise ValueError("Address must contain '://'")

    module_string, inner_address = address.split("://", maxsplit=1)
    return (module_string, inner_address)


def get_entrypoint_name():
    """Get the entrypoint of the current script."""
    prefix = ""
    try:
        curr = psutil.Process()
        # Prepend `interactive_shell` for interactive shell scripts.
        # https://stackoverflow.com/questions/2356399/tell-if-python-is-in-interactive-mode # noqa
        if hasattr(sys, "ps1"):
            prefix = "(interactive_shell) "

        return prefix + list2cmdline(curr.cmdline())
    except Exception:
        return "unknown"


class DeferSigint(contextlib.AbstractContextManager):
    """Context manager that defers SIGINT signals until the context is left."""

    # This is used by Ray's task cancellation to defer cancellation interrupts during
    # problematic areas, e.g. task argument deserialization.
    def __init__(self):
        # Whether a SIGINT signal was received during the context.
        self.signal_received = False
        # The overridden SIGINT handler
        self.overridden_sigint_handler = None
        # The original signal method.
        self.orig_signal = None

    @classmethod
    def create_if_main_thread(cls) -> contextlib.AbstractContextManager:
        """Creates a DeferSigint context manager if running on the main thread,
        returns a no-op context manager otherwise.
        """
        if threading.current_thread() == threading.main_thread():
            return cls()
        else:
            return contextlib.nullcontext()

    def _set_signal_received(self, signum, frame):
        """SIGINT handler that defers the signal."""
        self.signal_received = True

    def _signal_monkey_patch(self, signum, handler):
        """Monkey patch for signal.signal that defers the setting of new signal
        handler after the DeferSigint context exits."""
        # Only handle it in the main thread because if setting a handler in a non-main
        # thread, signal.signal will raise an error because Python doesn't allow it.
        if (
            threading.current_thread() == threading.main_thread()
            and signum == signal.SIGINT
        ):
            orig_sigint_handler = self.overridden_sigint_handler
            self.overridden_sigint_handler = handler
            return orig_sigint_handler
        return self.orig_signal(signum, handler)

    def __enter__(self):
        # Save original SIGINT handler for later restoration.
        self.overridden_sigint_handler = signal.getsignal(signal.SIGINT)
        # Set SIGINT signal handler that defers the signal.
        signal.signal(signal.SIGINT, self._set_signal_received)
        # Monkey patch signal.signal to raise an error if a SIGINT handler is registered
        # within the context.
        self.orig_signal = signal.signal
        signal.signal = self._signal_monkey_patch
        return self

    def __exit__(self, exc_type, exc, exc_tb):
        assert self.overridden_sigint_handler is not None
        assert self.orig_signal is not None
        # Restore original signal.signal function.
        signal.signal = self.orig_signal
        # Restore overridden SIGINT handler.
        signal.signal(signal.SIGINT, self.overridden_sigint_handler)
        if exc_type is None and self.signal_received:
            # No exception raised in context, call the original SIGINT handler.
            # By default, this means raising KeyboardInterrupt.
            self.overridden_sigint_handler(signal.SIGINT, None)
        else:
            # If exception was raised in context, returning False will cause it to be
            # reraised.
            return False


def try_import_each_module(module_names_to_import: List[str]) -> None:
    """
    Make a best-effort attempt to import each named Python module.
    This is used by the Python default_worker.py to preload modules.
    """
    for module_to_preload in module_names_to_import:
        try:
            importlib.import_module(module_to_preload)
        except ImportError:
            logger.exception(f'Failed to preload the module "{module_to_preload}"')


def remove_ray_internal_flags_from_env(env: dict):
    """
    Remove Ray internal flags from `env`.
    Defined in ray/common/ray_internal_flag_def.h
    """
    for flag in ray_constants.RAY_INTERNAL_FLAGS:
        env.pop(flag, None)


def update_envs(env_vars: Dict[str, str]):
    """
    When updating the environment variable, if there is ${X},
    it will be replaced with the current environment variable.
    """
    if not env_vars:
        return

    for key, value in env_vars.items():
        expanded = os.path.expandvars(value)
        # Replace non-existant env vars with an empty string.
        result = re.sub(r"\$\{[A-Z0-9_]+\}", "", expanded)
        os.environ[key] = result


def parse_pg_formatted_resources_to_original(
    pg_formatted_resources: Dict[str, float]
) -> Dict[str, float]:
    original_resources = {}

    for key, value in pg_formatted_resources.items():
        result = PLACEMENT_GROUP_INDEXED_BUNDLED_RESOURCE_PATTERN.match(key)
        if result and len(result.groups()) == 3:
            # Filter out resources that have bundle_group_[pg_id] since
            # it is an implementation detail.
            # This resource is automatically added to the resource
            # request for all tasks that require placement groups.
            if result.group(1) == PLACEMENT_GROUP_BUNDLE_RESOURCE_NAME:
                continue

            original_resources[result.group(1)] = value
            continue

        result = PLACEMENT_GROUP_WILDCARD_RESOURCE_PATTERN.match(key)
        if result and len(result.groups()) == 2:
            if result.group(1) == "bundle":
                continue

            original_resources[result.group(1)] = value
            continue
        original_resources[key] = value

    return original_resources


def validate_actor_state_name(actor_state_name):
    if actor_state_name is None:
        return
    actor_state_names = [
        "DEPENDENCIES_UNREADY",
        "PENDING_CREATION",
        "ALIVE",
        "RESTARTING",
        "DEAD",
    ]
    if actor_state_name not in actor_state_names:
        raise ValueError(
            f'"{actor_state_name}" is not a valid actor state name, '
            'it must be one of the following: "DEPENDENCIES_UNREADY", '
            '"PENDING_CREATION", "ALIVE", "RESTARTING", or "DEAD"'
        )


def get_current_node_cpu_model_name() -> Optional[str]:
    if not sys.platform.startswith("linux"):
        return None

    try:
        """
        /proc/cpuinfo content example:

        processor	: 0
        vendor_id	: GenuineIntel
        cpu family	: 6
        model		: 85
        model name	: Intel(R) Xeon(R) Platinum 8259CL CPU @ 2.50GHz
        stepping	: 7
        """
        with open("/proc/cpuinfo", "r") as f:
            for line in f:
                if line.startswith("model name"):
                    return line.split(":")[1].strip()
        return None
    except Exception:
        logger.debug("Failed to get CPU model name", exc_info=True)
        return None


def validate_socket_filepath(filepath: str):
    """
    Validate the provided filename is a valid Unix socket filename.
    """
    # Don't check for Windows as it doesn't support Unix sockets.
    if sys.platform == "win32":
        return
    is_mac = sys.platform.startswith("darwin")
    maxlen = (104 if is_mac else 108) - 1
    if len(filepath.encode("utf-8")) > maxlen:
        raise OSError(
            f"validate_socket_filename failed: AF_UNIX path length cannot exceed {maxlen} bytes: {filepath}"
        )


# Whether we're currently running in a test, either local or CI.
in_test = None


def is_in_test():
    global in_test

    if in_test is None:
        in_test = any(
            env_var in os.environ
            # These environment variables are always set by pytest and Buildkite,
            # respectively.
            for env_var in ("PYTEST_CURRENT_TEST", "BUILDKITE")
        )
    return in_test