Add files using upload-large-folder tool

lib/python3.12/site-packages/nvidia/cufft/include/cufftXt.h

@@ -0,0 +1,259 @@
+
+ /* Copyright 2005-2021 NVIDIA Corporation.  All rights reserved.
+  *
+  * NOTICE TO LICENSEE:
+  *
+  * The source code and/or documentation ("Licensed Deliverables") are
+  * subject to NVIDIA intellectual property rights under U.S. and
+  * international Copyright laws.
+  *
+  * The Licensed Deliverables contained herein are PROPRIETARY and
+  * CONFIDENTIAL to NVIDIA and are being provided under the terms and
+  * conditions of a form of NVIDIA software license agreement by and
+  * between NVIDIA and Licensee ("License Agreement") or electronically
+  * accepted by Licensee.  Notwithstanding any terms or conditions to
+  * the contrary in the License Agreement, reproduction or disclosure
+  * of the Licensed Deliverables to any third party without the express
+  * written consent of NVIDIA is prohibited.
+  *
+  * NOTWITHSTANDING ANY TERMS OR CONDITIONS TO THE CONTRARY IN THE
+  * LICENSE AGREEMENT, NVIDIA MAKES NO REPRESENTATION ABOUT THE
+  * SUITABILITY OF THESE LICENSED DELIVERABLES FOR ANY PURPOSE.  THEY ARE
+  * PROVIDED "AS IS" WITHOUT EXPRESS OR IMPLIED WARRANTY OF ANY KIND.
+  * NVIDIA DISCLAIMS ALL WARRANTIES WITH REGARD TO THESE LICENSED
+  * DELIVERABLES, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY,
+  * NONINFRINGEMENT, AND FITNESS FOR A PARTICULAR PURPOSE.
+  * NOTWITHSTANDING ANY TERMS OR CONDITIONS TO THE CONTRARY IN THE
+  * LICENSE AGREEMENT, IN NO EVENT SHALL NVIDIA BE LIABLE FOR ANY
+  * SPECIAL, INDIRECT, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, OR ANY
+  * DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
+  * WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
+  * ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
+  * OF THESE LICENSED DELIVERABLES.
+  *
+  * U.S. Government End Users.  These Licensed Deliverables are a
+  * "commercial item" as that term is defined at 48 C.F.R. 2.101 (OCT
+  * 1995), consisting of "commercial computer software" and "commercial
+  * computer software documentation" as such terms are used in 48
+  * C.F.R. 12.212 (SEPT 1995) and are provided to the U.S. Government
+  * only as a commercial end item.  Consistent with 48 C.F.R.12.212 and
+  * 48 C.F.R. 227.7202-1 through 227.7202-4 (JUNE 1995), all
+  * U.S. Government End Users acquire the Licensed Deliverables with
+  * only those rights set forth herein.
+  *
+  * Any use of the Licensed Deliverables in individual and commercial
+  * software must include, in the user documentation and internal
+  * comments to the code, the above Disclaimer and U.S. Government End
+  * Users Notice.
+  */
+
+/*!
+* \file cufftXt.h
+* \brief Public header file for the NVIDIA CUDA FFT library (CUFFT)
+*/
+
+#ifndef _CUFFTXT_H_
+#define _CUFFTXT_H_
+#include "cudalibxt.h"
+#include "cufft.h"
+
+
+#ifndef CUFFTAPI
+#ifdef _WIN32
+#define CUFFTAPI __stdcall
+#else
+#define CUFFTAPI
+#endif
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+//
+// cufftXtSubFormat identifies the data layout of
+// a memory descriptor owned by cufft.
+// note that multi GPU cufft does not yet support out-of-place transforms
+//
+
+typedef enum cufftXtSubFormat_t {
+    CUFFT_XT_FORMAT_INPUT = 0x00,              //by default input is in linear order across GPUs
+    CUFFT_XT_FORMAT_OUTPUT = 0x01,             //by default output is in scrambled order depending on transform
+    CUFFT_XT_FORMAT_INPLACE = 0x02,            //by default inplace is input order, which is linear across GPUs
+    CUFFT_XT_FORMAT_INPLACE_SHUFFLED = 0x03,   //shuffled output order after execution of the transform
+    CUFFT_XT_FORMAT_1D_INPUT_SHUFFLED = 0x04,  //shuffled input order prior to execution of 1D transforms
+    CUFFT_XT_FORMAT_DISTRIBUTED_INPUT = 0x05,
+    CUFFT_XT_FORMAT_DISTRIBUTED_OUTPUT = 0x06,
+    CUFFT_FORMAT_UNDEFINED = 0x07
+} cufftXtSubFormat;
+
+//
+// cufftXtCopyType specifies the type of copy for cufftXtMemcpy
+//
+typedef enum cufftXtCopyType_t {
+    CUFFT_COPY_HOST_TO_DEVICE = 0x00,
+    CUFFT_COPY_DEVICE_TO_HOST = 0x01,
+    CUFFT_COPY_DEVICE_TO_DEVICE = 0x02,
+    CUFFT_COPY_UNDEFINED = 0x03
+} cufftXtCopyType;
+
+//
+// cufftXtQueryType specifies the type of query for cufftXtQueryPlan
+//
+typedef enum cufftXtQueryType_t {
+    CUFFT_QUERY_1D_FACTORS = 0x00,
+    CUFFT_QUERY_UNDEFINED = 0x01
+} cufftXtQueryType;
+
+typedef struct cufftXt1dFactors_t {
+    long long int size;
+    long long int stringCount;
+    long long int stringLength;
+    long long int substringLength;
+    long long int factor1;
+    long long int factor2;
+    long long int stringMask;
+    long long int substringMask;
+    long long int factor1Mask;
+    long long int factor2Mask;
+    int stringShift;
+    int substringShift;
+    int factor1Shift;
+    int factor2Shift;
+} cufftXt1dFactors;
+
+//
+// cufftXtWorkAreaPolicy specifies policy for cufftXtSetWorkAreaPolicy
+//
+typedef enum cufftXtWorkAreaPolicy_t {
+    CUFFT_WORKAREA_MINIMAL = 0, /* maximum reduction */
+    CUFFT_WORKAREA_USER = 1, /* use workSize parameter as limit */
+    CUFFT_WORKAREA_PERFORMANCE = 2, /* default - 1x overhead or more, maximum performance */
+} cufftXtWorkAreaPolicy;
+
+// multi-GPU routines
+cufftResult CUFFTAPI cufftXtSetGPUs(cufftHandle handle, int nGPUs, int *whichGPUs);
+
+cufftResult CUFFTAPI cufftXtMalloc(cufftHandle plan,
+                                   cudaLibXtDesc ** descriptor,
+                                   cufftXtSubFormat format);
+
+cufftResult CUFFTAPI cufftXtMemcpy(cufftHandle plan,
+                                   void *dstPointer,
+                                   void *srcPointer,
+                                   cufftXtCopyType type);
+
+cufftResult CUFFTAPI cufftXtFree(cudaLibXtDesc *descriptor);
+
+cufftResult CUFFTAPI cufftXtSetWorkArea(cufftHandle plan, void **workArea);
+
+cufftResult CUFFTAPI cufftXtExecDescriptorC2C(cufftHandle plan,
+                                              cudaLibXtDesc *input,
+                                              cudaLibXtDesc *output,
+                                              int direction);
+
+cufftResult CUFFTAPI cufftXtExecDescriptorR2C(cufftHandle plan,
+                                              cudaLibXtDesc *input,
+                                              cudaLibXtDesc *output);
+
+cufftResult CUFFTAPI cufftXtExecDescriptorC2R(cufftHandle plan,
+                                              cudaLibXtDesc *input,
+                                              cudaLibXtDesc *output);
+
+cufftResult CUFFTAPI cufftXtExecDescriptorZ2Z(cufftHandle plan,
+                                              cudaLibXtDesc *input,
+                                              cudaLibXtDesc *output,
+                                              int direction);
+
+cufftResult CUFFTAPI cufftXtExecDescriptorD2Z(cufftHandle plan,
+                                              cudaLibXtDesc *input,
+                                              cudaLibXtDesc *output);
+
+cufftResult CUFFTAPI cufftXtExecDescriptorZ2D(cufftHandle plan,
+                                              cudaLibXtDesc *input,
+                                              cudaLibXtDesc *output);
+
+// Utility functions
+
+cufftResult CUFFTAPI cufftXtQueryPlan(cufftHandle plan, void *queryStruct, cufftXtQueryType queryType);
+
+
+// callbacks
+
+
+typedef enum cufftXtCallbackType_t {
+    CUFFT_CB_LD_COMPLEX = 0x0,
+    CUFFT_CB_LD_COMPLEX_DOUBLE = 0x1,
+    CUFFT_CB_LD_REAL = 0x2,
+    CUFFT_CB_LD_REAL_DOUBLE = 0x3,
+    CUFFT_CB_ST_COMPLEX = 0x4,
+    CUFFT_CB_ST_COMPLEX_DOUBLE = 0x5,
+    CUFFT_CB_ST_REAL = 0x6,
+    CUFFT_CB_ST_REAL_DOUBLE = 0x7,
+    CUFFT_CB_UNDEFINED = 0x8
+
+} cufftXtCallbackType;
+
+typedef cufftComplex (*cufftCallbackLoadC)(void *dataIn, size_t offset, void *callerInfo, void *sharedPointer);
+typedef cufftDoubleComplex (*cufftCallbackLoadZ)(void *dataIn, size_t offset, void *callerInfo, void *sharedPointer);
+typedef cufftReal (*cufftCallbackLoadR)(void *dataIn, size_t offset, void *callerInfo, void *sharedPointer);
+typedef cufftDoubleReal(*cufftCallbackLoadD)(void *dataIn, size_t offset, void *callerInfo, void *sharedPointer);
+
+typedef void (*cufftCallbackStoreC)(void *dataOut, size_t offset, cufftComplex element, void *callerInfo, void *sharedPointer);
+typedef void (*cufftCallbackStoreZ)(void *dataOut, size_t offset, cufftDoubleComplex element, void *callerInfo, void *sharedPointer);
+typedef void (*cufftCallbackStoreR)(void *dataOut, size_t offset, cufftReal element, void *callerInfo, void *sharedPointer);
+typedef void (*cufftCallbackStoreD)(void *dataOut, size_t offset, cufftDoubleReal element, void *callerInfo, void *sharedPointer);
+
+
+cufftResult CUFFTAPI cufftXtSetCallback(cufftHandle plan, void **callback_routine, cufftXtCallbackType cbType, void **caller_info);
+cufftResult CUFFTAPI cufftXtClearCallback(cufftHandle plan, cufftXtCallbackType cbType);
+cufftResult CUFFTAPI cufftXtSetCallbackSharedSize(cufftHandle plan, cufftXtCallbackType cbType, size_t sharedSize);
+
+cufftResult CUFFTAPI cufftXtMakePlanMany(cufftHandle plan,
+                                         int rank,
+                                         long long int *n,
+                                         long long int *inembed,
+                                         long long int istride,
+                                         long long int idist,
+                                         cudaDataType inputtype,
+                                         long long int *onembed,
+                                         long long int ostride,
+                                         long long int odist,
+                                         cudaDataType outputtype,
+                                         long long int batch,
+                                         size_t *workSize,
+                                       	 cudaDataType executiontype);
+
+cufftResult CUFFTAPI cufftXtGetSizeMany(cufftHandle plan,
+                                        int rank,
+                                        long long int *n,
+                                        long long int *inembed,
+                                        long long int istride,
+                                        long long int idist,
+                                        cudaDataType inputtype,
+                                        long long int *onembed,
+                                        long long int ostride,
+                                        long long int odist,
+                                        cudaDataType outputtype,
+                                        long long int batch,
+                                        size_t *workSize,
+                                        cudaDataType executiontype);
+
+
+cufftResult CUFFTAPI cufftXtExec(cufftHandle plan,
+                                 void *input,
+                                 void *output,
+                                 int direction);
+
+cufftResult CUFFTAPI cufftXtExecDescriptor(cufftHandle plan,
+                                           cudaLibXtDesc *input,
+                                           cudaLibXtDesc *output,
+                                           int direction);
+
+cufftResult CUFFTAPI cufftXtSetWorkAreaPolicy(cufftHandle plan, cufftXtWorkAreaPolicy policy, size_t *workSize);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

lib/python3.12/site-packages/ray/_common/constants.py

@@ -0,0 +1,7 @@
+# Prefix for the node id resource that is automatically added to each node.
+# For example, a node may have id `node:172.23.42.1`.
+NODE_ID_PREFIX = "node:"
+# The system resource that head node has.
+HEAD_NODE_RESOURCE_NAME = NODE_ID_PREFIX + "__internal_head__"
+
+RAY_WARN_BLOCKING_GET_INSIDE_ASYNC_ENV_VAR = "RAY_WARN_BLOCKING_GET_INSIDE_ASYNC"

lib/python3.12/site-packages/ray/_common/deprecation.py

@@ -0,0 +1,141 @@
+import inspect
+import logging
+from typing import Optional, Union
+
+from ray.util import log_once
+from ray.util.annotations import _mark_annotated
+
+logger = logging.getLogger(__name__)
+
+# A constant to use for any configuration that should be deprecated
+# (to check, whether this config has actually been assigned a proper value or
+# not).
+DEPRECATED_VALUE = -1
+
+
+def deprecation_warning(
+    old: str,
+    new: Optional[str] = None,
+    *,
+    help: Optional[str] = None,
+    error: Optional[Union[bool, Exception]] = None,
+    stacklevel: int = 2,
+) -> None:
+    """Warns (via the `logger` object) or throws a deprecation warning/error.
+
+    Args:
+        old: A description of the "thing" that is to be deprecated.
+        new: A description of the new "thing" that replaces it.
+        help: An optional help text to tell the user, what to
+            do instead of using `old`.
+        error: Whether or which exception to raise. If True, raise ValueError.
+            If False, just warn. If `error` is-a subclass of Exception,
+            raise that Exception.
+        stacklevel: The stacklevel to use for the warning message.
+            Use 2 to point to where this function is called, 3+ to point
+            further up the stack.
+
+    Raises:
+        ValueError: If `error=True`.
+        Exception: Of type `error`, iff `error` is a sub-class of `Exception`.
+    """
+    msg = "`{}` has been deprecated.{}".format(
+        old, (" Use `{}` instead.".format(new) if new else f" {help}" if help else "")
+    )
+
+    if error:
+        if not isinstance(error, bool) and issubclass(error, Exception):
+            # error is an Exception
+            raise error(msg)
+        else:
+            # error is a boolean, construct ValueError ourselves
+            raise ValueError(msg)
+    else:
+        logger.warning(
+            "DeprecationWarning: " + msg + " This will raise an error in the future!",
+            stacklevel=stacklevel,
+        )
+
+
+def Deprecated(old=None, *, new=None, help=None, error):
+    """Decorator for documenting a deprecated class, method, or function.
+
+    Automatically adds a `deprecation.deprecation_warning(old=...,
+    error=False)` to not break existing code at this point to the decorated
+    class' constructor, method, or function.
+
+    In a next major release, this warning should then be made an error
+    (by setting error=True), which means at this point that the
+    class/method/function is no longer supported, but will still inform
+    the user about the deprecation event.
+
+    In a further major release, the class, method, function should be erased
+    entirely from the codebase.
+
+
+    .. testcode::
+        :skipif: True
+
+        from ray._common.deprecation import Deprecated
+        # Deprecated class: Patches the constructor to warn if the class is
+        # used.
+        @Deprecated(new="NewAndMuchCoolerClass", error=False)
+        class OldAndUncoolClass:
+            ...
+
+        # Deprecated class method: Patches the method to warn if called.
+        class StillCoolClass:
+            ...
+            @Deprecated(new="StillCoolClass.new_and_much_cooler_method()",
+                        error=False)
+            def old_and_uncool_method(self, uncool_arg):
+                ...
+
+        # Deprecated function: Patches the function to warn if called.
+        @Deprecated(new="new_and_much_cooler_function", error=False)
+        def old_and_uncool_function(*uncool_args):
+            ...
+    """
+
+    def _inner(obj):
+        # A deprecated class.
+        if inspect.isclass(obj):
+            # Patch the class' init method to raise the warning/error.
+            obj_init = obj.__init__
+
+            def patched_init(*args, **kwargs):
+                if log_once(old or obj.__name__):
+                    deprecation_warning(
+                        old=old or obj.__name__,
+                        new=new,
+                        help=help,
+                        error=error,
+                        stacklevel=3,
+                    )
+                return obj_init(*args, **kwargs)
+
+            obj.__init__ = patched_init
+            _mark_annotated(obj)
+            # Return the patched class (with the warning/error when
+            # instantiated).
+            return obj
+
+        # A deprecated class method or function.
+        # Patch with the warning/error at the beginning.
+        def _ctor(*args, **kwargs):
+            if log_once(old or obj.__name__):
+                deprecation_warning(
+                    old=old or obj.__name__,
+                    new=new,
+                    help=help,
+                    error=error,
+                    stacklevel=3,
+                )
+            # Call the deprecated method/function.
+            return obj(*args, **kwargs)
+
+        # Return the patched class method/function.
+        return _ctor
+
+    # Return the prepared decorator.
+    return _inner

lib/python3.12/site-packages/ray/_common/filters.py

@@ -0,0 +1,55 @@
+import logging
+from typing import Any, Dict
+
+import ray
+from ray._private.ray_logging.constants import LogKey
+
+
+class CoreContextFilter(logging.Filter):
+    TASK_LEVEL_LOG_KEYS = [
+        LogKey.TASK_ID.value,
+        LogKey.TASK_NAME.value,
+        LogKey.TASK_FUNCTION_NAME.value,
+    ]
+
+    @classmethod
+    def get_ray_core_logging_context(cls) -> Dict[str, Any]:
+        """
+        Get the ray core logging context as a dict.
+        Only use this function if you need include the attributes to the log record
+        yourself by bypassing the filter.
+        """
+        if not ray.is_initialized():
+            # There is no additional context if ray is not initialized
+            return {}
+
+        runtime_context = ray.get_runtime_context()
+        ray_core_logging_context = {
+            LogKey.JOB_ID.value: runtime_context.get_job_id(),
+            LogKey.WORKER_ID.value: runtime_context.get_worker_id(),
+            LogKey.NODE_ID.value: runtime_context.get_node_id(),
+        }
+        if runtime_context.worker.mode == ray.WORKER_MODE:
+            ray_core_logging_context[
+                LogKey.ACTOR_ID.value
+            ] = runtime_context.get_actor_id()
+            ray_core_logging_context[
+                LogKey.TASK_ID.value
+            ] = runtime_context.get_task_id()
+            ray_core_logging_context[
+                LogKey.TASK_NAME.value
+            ] = runtime_context.get_task_name()
+            ray_core_logging_context[
+                LogKey.TASK_FUNCTION_NAME.value
+            ] = runtime_context.get_task_function_name()
+            ray_core_logging_context[
+                LogKey.ACTOR_NAME.value
+            ] = runtime_context.get_actor_name()
+        return ray_core_logging_context
+
+    def filter(self, record):
+        context = self.get_ray_core_logging_context()
+        for key, value in context.items():
+            if value is not None:
+                setattr(record, key, value)
+        return True

lib/python3.12/site-packages/ray/_common/formatters.py

@@ -0,0 +1,120 @@
+import json
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List
+
+from ray._private.log import INTERNAL_TIMESTAMP_LOG_KEY
+from ray._private.ray_constants import LOGGER_FORMAT
+from ray._private.ray_logging.constants import (
+    LOGGER_FLATTEN_KEYS,
+    LOGRECORD_STANDARD_ATTRS,
+    LogKey,
+)
+
+
+def _append_flatten_attributes(formatted_attrs: Dict[str, Any], key: str, value: Any):
+    """Flatten the dictionary values for special keys and append the values in place.
+
+    If the key is in `LOGGER_FLATTEN_KEYS`, the value will be flattened and appended
+    to the `formatted_attrs` dictionary. Otherwise, the key-value pair will be appended
+    directly.
+    """
+    if key in LOGGER_FLATTEN_KEYS:
+        if not isinstance(value, dict):
+            raise ValueError(
+                f"Expected a dictionary passing into {key}, but got {type(value)}"
+            )
+        for k, v in value.items():
+            if k in formatted_attrs:
+                raise KeyError(f"Found duplicated key in the log record: {k}")
+            formatted_attrs[k] = v
+    else:
+        formatted_attrs[key] = value
+
+
+class AbstractFormatter(logging.Formatter, ABC):
+    def __init__(self, fmt=None, datefmt=None, style="%", validate=True) -> None:
+        super().__init__(fmt, datefmt, style, validate)
+        self._additional_log_standard_attrs = []
+
+    def set_additional_log_standard_attrs(
+        self, additional_log_standard_attrs: List[str]
+    ) -> None:
+        self._additional_log_standard_attrs = additional_log_standard_attrs
+
+    @property
+    def additional_log_standard_attrs(self) -> List[str]:
+        return self._additional_log_standard_attrs
+
+    def generate_record_format_attrs(
+        self,
+        record: logging.LogRecord,
+        exclude_default_standard_attrs,
+    ) -> dict:
+        record_format_attrs = {}
+
+        # If `exclude_default_standard_attrs` is False, include the standard attributes.
+        # Otherwise, include only Ray and user-provided context.
+        if not exclude_default_standard_attrs:
+            record_format_attrs.update(
+                {
+                    LogKey.ASCTIME.value: self.formatTime(record),
+                    LogKey.LEVELNAME.value: record.levelname,
+                    LogKey.MESSAGE.value: record.getMessage(),
+                    LogKey.FILENAME.value: record.filename,
+                    LogKey.LINENO.value: record.lineno,
+                    LogKey.PROCESS.value: record.process,
+                }
+            )
+            if record.exc_info:
+                if not record.exc_text:
+                    record.exc_text = self.formatException(record.exc_info)
+                record_format_attrs[LogKey.EXC_TEXT.value] = record.exc_text
+
+        # Add the user specified additional standard attributes.
+        for key in self._additional_log_standard_attrs:
+            _append_flatten_attributes(
+                record_format_attrs, key, getattr(record, key, None)
+            )
+
+        for key, value in record.__dict__.items():
+            # Both Ray and user-provided context are stored in `record_format`.
+            if key not in LOGRECORD_STANDARD_ATTRS:
+                _append_flatten_attributes(record_format_attrs, key, value)
+
+        # Format the internal timestamp to the standardized `timestamp_ns` key.
+        if INTERNAL_TIMESTAMP_LOG_KEY in record_format_attrs:
+            record_format_attrs[LogKey.TIMESTAMP_NS.value] = record_format_attrs.pop(
+                INTERNAL_TIMESTAMP_LOG_KEY
+            )
+
+        return record_format_attrs
+
+    @abstractmethod
+    def format(self, record: logging.LogRecord) -> str:
+        pass
+
+
+class JSONFormatter(AbstractFormatter):
+    def format(self, record: logging.LogRecord) -> str:
+        record_format_attrs = self.generate_record_format_attrs(
+            record, exclude_default_standard_attrs=False
+        )
+        return json.dumps(record_format_attrs)
+
+
+class TextFormatter(AbstractFormatter):
+    def __init__(self, fmt=None, datefmt=None, style="%", validate=True) -> None:
+        super().__init__(fmt, datefmt, style, validate)
+        self._inner_formatter = logging.Formatter(LOGGER_FORMAT)
+
+    def format(self, record: logging.LogRecord) -> str:
+        s = self._inner_formatter.format(record)
+        record_format_attrs = self.generate_record_format_attrs(
+            record, exclude_default_standard_attrs=True
+        )
+
+        additional_attrs = " ".join(
+            [f"{key}={value}" for key, value in record_format_attrs.items()]
+        )
+        return f"{s} {additional_attrs}"

lib/python3.12/site-packages/ray/_common/network_utils.py

@@ -0,0 +1,111 @@
+import socket
+from contextlib import closing
+from functools import lru_cache
+from typing import Optional, Tuple, Union
+
+from ray._raylet import (
+    build_address as _build_address,
+    is_ipv6 as _is_ipv6,
+    node_ip_address_from_perspective as _node_ip_address_from_perspective,
+    parse_address as _parse_address,
+)
+
+
+def parse_address(address: str) -> Optional[Tuple[str, str]]:
+    """Parse a network address string into host and port.
+
+    Args:
+        address: The address string to parse (e.g., "localhost:8000", "[::1]:8000").
+
+    Returns:
+        Tuple with (host, port) if port found, None if no colon separator.
+    """
+    return _parse_address(address)
+
+
+def build_address(host: str, port: Union[int, str]) -> str:
+    """Build a network address string from host and port.
+
+    Args:
+        host: The hostname or IP address.
+        port: The port number (int or string).
+
+    Returns:
+        Formatted address string (e.g., "localhost:8000" or "[::1]:8000").
+    """
+    return _build_address(host, port)
+
+
+def node_ip_address_from_perspective(address: Optional[str] = None) -> str:
+    """IP address by which the local node can be reached *from* the `address`.
+
+    If no address is given, defaults to public DNS servers for detection.
+
+    Args:
+        address: The IP address and port of any known live service on the
+            network you care about.
+
+    Returns:
+        The IP address by which the local node can be reached from the address.
+    """
+    return _node_ip_address_from_perspective(address)
+
+
+def is_ipv6(host: str) -> bool:
+    """Check if a host is resolved to IPv6.
+
+    Args:
+        host: The IP or domain name to check (must be without port).
+
+    Returns:
+        True if the host is resolved to IPv6, False if IPv4.
+    """
+    return _is_ipv6(host)
+
+
+@lru_cache(maxsize=1)
+def get_localhost_ip() -> str:
+    """Get localhost loopback ip with IPv4/IPv6 support.
+
+    Returns:
+        The localhost loopback IP.
+    """
+    # Try IPv4 first, then IPv6 localhost resolution
+    for family in [socket.AF_INET, socket.AF_INET6]:
+        try:
+            dns_result = socket.getaddrinfo(
+                "localhost", None, family, socket.SOCK_STREAM
+            )
+            return dns_result[0][4][0]
+        except Exception:
+            continue
+
+    # Final fallback to IPv4 loopback
+    return "127.0.0.1"
+
+
+def is_localhost(host: str) -> bool:
+    """Check if the given host string represents a localhost address.
+
+    Args:
+        host: The hostname or IP address to check.
+
+    Returns:
+        True if the host is a localhost address, False otherwise.
+    """
+    return host in ("localhost", "127.0.0.1", "::1")
+
+
+def find_free_port(family: socket.AddressFamily = socket.AF_INET) -> int:
+    """Find a free port on the local machine.
+
+    Args:
+        family: The socket address family (AF_INET for IPv4, AF_INET6 for IPv6).
+            Defaults to AF_INET.
+
+    Returns:
+        An available port number.
+    """
+    with closing(socket.socket(family, socket.SOCK_STREAM)) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]

lib/python3.12/site-packages/ray/_common/pydantic_compat.py

@@ -0,0 +1,108 @@
+# ruff: noqa
+import packaging.version
+
+# Pydantic is a dependency of `ray["default"]` but not the minimal installation,
+# so handle the case where it isn't installed.
+try:
+    import pydantic
+
+    PYDANTIC_INSTALLED = True
+except ImportError:
+    pydantic = None
+    PYDANTIC_INSTALLED = False
+
+
+if not PYDANTIC_INSTALLED:
+    IS_PYDANTIC_2 = False
+    BaseModel = None
+    Extra = None
+    Field = None
+    NonNegativeFloat = None
+    NonNegativeInt = None
+    PositiveFloat = None
+    PositiveInt = None
+    PrivateAttr = None
+    StrictInt = None
+    ValidationError = None
+    root_validator = None
+    validator = None
+    is_subclass_of_base_model = lambda obj: False
+# In pydantic <1.9.0, __version__ attribute is missing, issue ref:
+# https://github.com/pydantic/pydantic/issues/2572, so we need to check
+# the existence prior to comparison.
+elif not hasattr(pydantic, "__version__") or packaging.version.parse(
+    pydantic.__version__
+) < packaging.version.parse("2.0"):
+    IS_PYDANTIC_2 = False
+    from pydantic import (
+        BaseModel,
+        Extra,
+        Field,
+        NonNegativeFloat,
+        NonNegativeInt,
+        PositiveFloat,
+        PositiveInt,
+        PrivateAttr,
+        StrictInt,
+        ValidationError,
+        root_validator,
+        validator,
+    )
+
+    def is_subclass_of_base_model(obj):
+        return issubclass(obj, BaseModel)
+
+else:
+    IS_PYDANTIC_2 = True
+    from pydantic.v1 import (
+        BaseModel,
+        Extra,
+        Field,
+        NonNegativeFloat,
+        NonNegativeInt,
+        PositiveFloat,
+        PositiveInt,
+        PrivateAttr,
+        StrictInt,
+        ValidationError,
+        root_validator,
+        validator,
+    )
+
+    def is_subclass_of_base_model(obj):
+        from pydantic import BaseModel as BaseModelV2
+        from pydantic.v1 import BaseModel as BaseModelV1
+
+        return issubclass(obj, BaseModelV1) or issubclass(obj, BaseModelV2)
+
+
+def register_pydantic_serializers(serialization_context):
+    if not PYDANTIC_INSTALLED:
+        return
+
+    if IS_PYDANTIC_2:
+        # TODO(edoakes): compare against the version that has the fixes.
+        from pydantic.v1.fields import ModelField
+    else:
+        from pydantic.fields import ModelField
+
+    # Pydantic's Cython validators are not serializable.
+    # https://github.com/cloudpipe/cloudpickle/issues/408
+    serialization_context._register_cloudpickle_serializer(
+        ModelField,
+        custom_serializer=lambda o: {
+            "name": o.name,
+            # outer_type_ is the original type for ModelFields,
+            # while type_ can be updated later with the nested type
+            # like int for List[int].
+            "type_": o.outer_type_,
+            "class_validators": o.class_validators,
+            "model_config": o.model_config,
+            "default": o.default,
+            "default_factory": o.default_factory,
+            "required": o.required,
+            "alias": o.alias,
+            "field_info": o.field_info,
+        },
+        custom_deserializer=lambda kwargs: ModelField(**kwargs),
+    )

lib/python3.12/site-packages/ray/_common/ray_constants.py

@@ -0,0 +1,5 @@
+# Default max_concurrency option in @ray.remote for async actors.
+DEFAULT_MAX_CONCURRENCY_ASYNC = 1000
+
+LOGGING_ROTATE_BYTES = 512 * 1024 * 1024  # 512MB.
+LOGGING_ROTATE_BACKUP_COUNT = 5  # 5 Backup files at max.

lib/python3.12/site-packages/ray/_common/ray_option_utils.py

@@ -0,0 +1,384 @@
+"""Manage, parse and validate options for Ray tasks, actors and actor methods."""
+import warnings
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, Optional, Tuple, Union
+
+import ray
+from ray._private import ray_constants
+from ray._private.label_utils import (
+    validate_fallback_strategy,
+    validate_label_selector,
+)
+from ray._private.utils import get_ray_doc_version
+from ray.util.placement_group import PlacementGroup
+from ray.util.scheduling_strategies import (
+    NodeAffinitySchedulingStrategy,
+    NodeLabelSchedulingStrategy,
+    PlacementGroupSchedulingStrategy,
+)
+
+
+@dataclass
+class Option:
+    # Type constraint of an option.
+    type_constraint: Optional[Union[type, Tuple[type]]] = None
+    # Value constraint of an option.
+    # The callable should return None if there is no error.
+    # Otherwise, return the error message.
+    value_constraint: Optional[Callable[[Any], Optional[str]]] = None
+    # Default value.
+    default_value: Any = None
+
+    def validate(self, keyword: str, value: Any):
+        """Validate the option."""
+        if self.type_constraint is not None:
+            if not isinstance(value, self.type_constraint):
+                raise TypeError(
+                    f"The type of keyword '{keyword}' must be {self.type_constraint}, "
+                    f"but received type {type(value)}"
+                )
+        if self.value_constraint is not None:
+            possible_error_message = self.value_constraint(value)
+            if possible_error_message:
+                raise ValueError(possible_error_message)
+
+
+def _counting_option(name: str, infinite: bool = True, default_value: Any = None):
+    """This is used for positive and discrete options.
+
+    Args:
+        name: The name of the option keyword.
+        infinite: If True, user could use -1 to represent infinity.
+        default_value: The default value for this option.
+
+    Returns:
+        An Option object.
+    """
+    if infinite:
+        return Option(
+            (int, type(None)),
+            lambda x: None
+            if (x is None or x >= -1)
+            else f"The keyword '{name}' only accepts None, 0, -1"
+            " or a positive integer, where -1 represents infinity.",
+            default_value=default_value,
+        )
+    return Option(
+        (int, type(None)),
+        lambda x: None
+        if (x is None or x >= 0)
+        else f"The keyword '{name}' only accepts None, 0 or a positive integer.",
+        default_value=default_value,
+    )
+
+
+def _validate_resource_quantity(name, quantity):
+    if quantity < 0:
+        return f"The quantity of resource {name} cannot be negative"
+    if (
+        isinstance(quantity, float)
+        and quantity != 0.0
+        and int(quantity * ray._raylet.RESOURCE_UNIT_SCALING) == 0
+    ):
+        return (
+            f"The precision of the fractional quantity of resource {name}"
+            " cannot go beyond 0.0001"
+        )
+    resource_name = "GPU" if name == "num_gpus" else name
+    if resource_name in ray._private.accelerators.get_all_accelerator_resource_names():
+        (
+            valid,
+            error_message,
+        ) = ray._private.accelerators.get_accelerator_manager_for_resource(
+            resource_name
+        ).validate_resource_request_quantity(
+            quantity
+        )
+        if not valid:
+            return error_message
+    return None
+
+
+def _resource_option(name: str, default_value: Any = None):
+    """This is used for resource related options."""
+    return Option(
+        (float, int, type(None)),
+        lambda x: None if (x is None) else _validate_resource_quantity(name, x),
+        default_value=default_value,
+    )
+
+
+def _validate_resources(resources: Optional[Dict[str, float]]) -> Optional[str]:
+    if resources is None:
+        return None
+
+    if "CPU" in resources or "GPU" in resources:
+        return (
+            "Use the 'num_cpus' and 'num_gpus' keyword instead of 'CPU' and 'GPU' "
+            "in 'resources' keyword"
+        )
+
+    for name, quantity in resources.items():
+        possible_error_message = _validate_resource_quantity(name, quantity)
+        if possible_error_message:
+            return possible_error_message
+
+    return None
+
+
+_common_options = {
+    "label_selector": Option((dict, type(None)), lambda x: validate_label_selector(x)),
+    "fallback_strategy": Option(
+        (list, type(None)), lambda x: validate_fallback_strategy(x)
+    ),
+    "accelerator_type": Option((str, type(None))),
+    "memory": _resource_option("memory"),
+    "name": Option((str, type(None))),
+    "num_cpus": _resource_option("num_cpus"),
+    "num_gpus": _resource_option("num_gpus"),
+    "object_store_memory": _counting_option("object_store_memory", False),
+    # TODO(suquark): "placement_group", "placement_group_bundle_index"
+    # and "placement_group_capture_child_tasks" are deprecated,
+    # use "scheduling_strategy" instead.
+    "placement_group": Option(
+        (type(None), str, PlacementGroup), default_value="default"
+    ),
+    "placement_group_bundle_index": Option(int, default_value=-1),
+    "placement_group_capture_child_tasks": Option((bool, type(None))),
+    "resources": Option((dict, type(None)), lambda x: _validate_resources(x)),
+    "runtime_env": Option((dict, type(None))),
+    "scheduling_strategy": Option(
+        (
+            type(None),
+            str,
+            PlacementGroupSchedulingStrategy,
+            NodeAffinitySchedulingStrategy,
+            NodeLabelSchedulingStrategy,
+        )
+    ),
+    "enable_task_events": Option(bool, default_value=True),
+    "_labels": Option((dict, type(None))),
+}
+
+
+def issubclass_safe(obj: Any, cls_: type) -> bool:
+    try:
+        return issubclass(obj, cls_)
+    except TypeError:
+        return False
+
+
+_task_only_options = {
+    "max_calls": _counting_option("max_calls", False, default_value=0),
+    # Normal tasks may be retried on failure this many times.
+    # TODO(swang): Allow this to be set globally for an application.
+    "max_retries": _counting_option(
+        "max_retries", default_value=ray_constants.DEFAULT_TASK_MAX_RETRIES
+    ),
+    # override "_common_options"
+    "num_cpus": _resource_option("num_cpus", default_value=1),
+    "num_returns": Option(
+        (int, str, type(None)),
+        lambda x: None
+        if (x is None or x == "dynamic" or x == "streaming" or x >= 0)
+        else "Default None. When None is passed, "
+        "The default value is 1 for a task and actor task, and "
+        "'streaming' for generator tasks and generator actor tasks. "
+        "The keyword 'num_returns' only accepts None, "
+        "a non-negative integer, "
+        "'streaming' (for generators), or 'dynamic'. 'dynamic' flag "
+        "will be deprecated in the future, and it is recommended to use "
+        "'streaming' instead.",
+        default_value=None,
+    ),
+    "object_store_memory": Option(  # override "_common_options"
+        (int, type(None)),
+        lambda x: None
+        if (x is None)
+        else "Setting 'object_store_memory' is not implemented for tasks",
+    ),
+    "retry_exceptions": Option(
+        (bool, list, tuple),
+        lambda x: None
+        if (
+            isinstance(x, bool)
+            or (
+                isinstance(x, (list, tuple))
+                and all(issubclass_safe(x_, Exception) for x_ in x)
+            )
+        )
+        else "retry_exceptions must be either a boolean or a list of exceptions",
+        default_value=False,
+    ),
+    "_generator_backpressure_num_objects": Option(
+        (int, type(None)),
+        lambda x: None
+        if x != 0
+        else (
+            "_generator_backpressure_num_objects=0 is not allowed. "
+            "Use a value > 0. If the value is equal to 1, the behavior "
+            "is identical to Python generator (generator 1 object "
+            "whenever `next` is called). Use -1 to disable this feature. "
+        ),
+    ),
+}
+
+_actor_only_options = {
+    "concurrency_groups": Option((list, dict, type(None))),
+    "enable_tensor_transport": Option((bool, type(None)), default_value=None),
+    "lifetime": Option(
+        (str, type(None)),
+        lambda x: None
+        if x in (None, "detached", "non_detached")
+        else "actor `lifetime` argument must be one of 'detached', "
+        "'non_detached' and 'None'.",
+    ),
+    "max_concurrency": _counting_option("max_concurrency", False),
+    "max_restarts": _counting_option("max_restarts", default_value=0),
+    "max_task_retries": _counting_option("max_task_retries", default_value=0),
+    "max_pending_calls": _counting_option("max_pending_calls", default_value=-1),
+    "namespace": Option((str, type(None))),
+    "get_if_exists": Option(bool, default_value=False),
+    "allow_out_of_order_execution": Option((bool, type(None))),
+}
+
+# Priority is important here because during dictionary update, same key with higher
+# priority overrides the same key with lower priority. We make use of priority
+# to set the correct default value for tasks / actors.
+
+# priority: _common_options > _actor_only_options > _task_only_options
+valid_options: Dict[str, Option] = {
+    **_task_only_options,
+    **_actor_only_options,
+    **_common_options,
+}
+# priority: _task_only_options > _common_options
+task_options: Dict[str, Option] = {**_common_options, **_task_only_options}
+# priority: _actor_only_options > _common_options
+actor_options: Dict[str, Option] = {**_common_options, **_actor_only_options}
+
+remote_args_error_string = (
+    "The @ray.remote decorator must be applied either with no arguments and no "
+    "parentheses, for example '@ray.remote', or it must be applied using some of "
+    f"the arguments in the list {list(valid_options.keys())}, for example "
+    "'@ray.remote(num_returns=2, resources={\"CustomResource\": 1})'."
+)
+
+
+def _check_deprecate_placement_group(options: Dict[str, Any]):
+    """Check if deprecated placement group option exists."""
+    placement_group = options.get("placement_group", "default")
+    scheduling_strategy = options.get("scheduling_strategy")
+    # TODO(suquark): @ray.remote(placement_group=None) is used in
+    # "python/ray.data._internal/remote_fn.py" and many other places,
+    # while "ray.data.read_api.read_datasource" set "scheduling_strategy=SPREAD".
+    # This might be a bug, but it is also ok to allow them co-exist.
+    if (placement_group not in ("default", None)) and (scheduling_strategy is not None):
+        raise ValueError(
+            "Placement groups should be specified via the "
+            "scheduling_strategy option. "
+            "The placement_group option is deprecated."
+        )
+
+
+def _warn_if_using_deprecated_placement_group(
+    options: Dict[str, Any], caller_stacklevel: int
+):
+    placement_group = options["placement_group"]
+    placement_group_bundle_index = options["placement_group_bundle_index"]
+    placement_group_capture_child_tasks = options["placement_group_capture_child_tasks"]
+    if placement_group != "default":
+        warnings.warn(
+            "placement_group parameter is deprecated. Use "
+            "scheduling_strategy=PlacementGroupSchedulingStrategy(...) "
+            "instead, see the usage at "
+            f"https://docs.ray.io/en/{get_ray_doc_version()}/ray-core/package-ref.html#ray-remote.",  # noqa: E501
+            DeprecationWarning,
+            stacklevel=caller_stacklevel + 1,
+        )
+    if placement_group_bundle_index != -1:
+        warnings.warn(
+            "placement_group_bundle_index parameter is deprecated. Use "
+            "scheduling_strategy=PlacementGroupSchedulingStrategy(...) "
+            "instead, see the usage at "
+            f"https://docs.ray.io/en/{get_ray_doc_version()}/ray-core/package-ref.html#ray-remote.",  # noqa: E501
+            DeprecationWarning,
+            stacklevel=caller_stacklevel + 1,
+        )
+    if placement_group_capture_child_tasks:
+        warnings.warn(
+            "placement_group_capture_child_tasks parameter is deprecated. Use "
+            "scheduling_strategy=PlacementGroupSchedulingStrategy(...) "
+            "instead, see the usage at "
+            f"https://docs.ray.io/en/{get_ray_doc_version()}/ray-core/package-ref.html#ray-remote.",  # noqa: E501
+            DeprecationWarning,
+            stacklevel=caller_stacklevel + 1,
+        )
+
+
+def validate_task_options(options: Dict[str, Any], in_options: bool):
+    """Options check for Ray tasks.
+
+    Args:
+        options: Options for Ray tasks.
+        in_options: If True, we are checking the options under the context of
+            ".options()".
+    """
+    for k, v in options.items():
+        if k not in task_options:
+            raise ValueError(
+                f"Invalid option keyword {k} for remote functions. "
+                f"Valid ones are {list(task_options)}."
+            )
+        task_options[k].validate(k, v)
+    if in_options and "max_calls" in options:
+        raise ValueError("Setting 'max_calls' is not supported in '.options()'.")
+    _check_deprecate_placement_group(options)
+
+
+def validate_actor_options(options: Dict[str, Any], in_options: bool):
+    """Options check for Ray actors.
+
+    Args:
+        options: Options for Ray actors.
+        in_options: If True, we are checking the options under the context of
+            ".options()".
+    """
+    for k, v in options.items():
+        if k not in actor_options:
+            raise ValueError(
+                f"Invalid option keyword {k} for actors. "
+                f"Valid ones are {list(actor_options)}."
+            )
+        actor_options[k].validate(k, v)
+
+    if in_options and "concurrency_groups" in options:
+        raise ValueError(
+            "Setting 'concurrency_groups' is not supported in '.options()'."
+        )
+
+    if options.get("get_if_exists") and not options.get("name"):
+        raise ValueError("The actor name must be specified to use `get_if_exists`.")
+
+    if "object_store_memory" in options:
+        warnings.warn(
+            "Setting 'object_store_memory'"
+            " for actors is deprecated since it doesn't actually"
+            " reserve the required object store memory."
+            f" Use object spilling that's enabled by default (https://docs.ray.io/en/{get_ray_doc_version()}/ray-core/objects/object-spilling.html) "  # noqa: E501
+            "instead to bypass the object store memory size limitation.",
+            DeprecationWarning,
+            stacklevel=1,
+        )
+
+    _check_deprecate_placement_group(options)
+
+
+def update_options(
+    original_options: Dict[str, Any], new_options: Dict[str, Any]
+) -> Dict[str, Any]:
+    """Update original options with new options and return.
+    The returned updated options contain shallow copy of original options.
+    """
+
+    return {**original_options, **new_options}

lib/python3.12/site-packages/ray/_common/retry.py

@@ -0,0 +1,82 @@
+import functools
+import logging
+import random
+import time
+from typing import Any, Callable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def call_with_retry(
+    f: Callable,
+    description: str,
+    match: Optional[List[str]] = None,
+    max_attempts: int = 10,
+    max_backoff_s: int = 32,
+    *args,
+    **kwargs,
+) -> Any:
+    """Retry a function with exponential backoff.
+
+    Args:
+        f: The function to retry.
+        description: An imperative description of the function being retried. For
+            example, "open the file".
+        match: A list of strings to match in the exception message. If ``None``, any
+            error is retried.
+        max_attempts: The maximum number of attempts to retry.
+        max_backoff_s: The maximum number of seconds to backoff.
+        *args: Arguments to pass to the function.
+        **kwargs: Keyword arguments to pass to the function.
+
+    Returns:
+        The result of the function.
+    """
+    # TODO: consider inverse match and matching exception type
+    assert max_attempts >= 1, f"`max_attempts` must be positive. Got {max_attempts}."
+
+    for i in range(max_attempts):
+        try:
+            return f(*args, **kwargs)
+        except Exception as e:
+            exception_str = str(e)
+            is_retryable = match is None or any(
+                pattern in exception_str for pattern in match
+            )
+            if is_retryable and i + 1 < max_attempts:
+                # Retry with binary exponential backoff with 20% random jitter.
+                backoff = min(2**i, max_backoff_s) * (random.uniform(0.8, 1.2))
+                logger.debug(
+                    f"Retrying {i+1} attempts to {description} after {backoff} seconds."
+                )
+                time.sleep(backoff)
+            else:
+                if is_retryable:
+                    logger.debug(
+                        f"Failed to {description} after {max_attempts} attempts. Raising."
+                    )
+                else:
+                    logger.debug(
+                        f"Did not find a match for {exception_str}. Raising after {i+1} attempts."
+                    )
+                raise e from None
+
+
+def retry(
+    description: str,
+    match: Optional[List[str]] = None,
+    max_attempts: int = 10,
+    max_backoff_s: int = 32,
+) -> Callable:
+    """Decorator-based version of call_with_retry."""
+
+    def decorator(func: Callable) -> Callable:
+        @functools.wraps(func)
+        def inner(*args, **kwargs):
+            return call_with_retry(
+                func, description, match, max_attempts, max_backoff_s, *args, **kwargs
+            )
+
+        return inner
+
+    return decorator

lib/python3.12/site-packages/ray/_common/serialization.py

@@ -0,0 +1,34 @@
+import io
+import logging
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    pass
+
+
+import ray._private.utils
+import ray.cloudpickle as pickle
+import ray.exceptions
+from ray._private import ray_constants
+from ray.util import inspect_serializability
+
+logger = logging.getLogger(__name__)
+ALLOW_OUT_OF_BAND_OBJECT_REF_SERIALIZATION = ray_constants.env_bool(
+    "RAY_allow_out_of_band_object_ref_serialization", True
+)
+
+
+def pickle_dumps(obj: Any, error_msg: str):
+    """Wrap cloudpickle.dumps to provide better error message
+    when the object is not serializable.
+    """
+    try:
+        return pickle.dumps(obj)
+    except (TypeError, ray.exceptions.OufOfBandObjectRefSerializationException) as e:
+        sio = io.StringIO()
+        inspect_serializability(obj, print_file=sio)
+        msg = f"{error_msg}:\n{sio.getvalue()}"
+        if isinstance(e, TypeError):
+            raise TypeError(msg) from e
+        else:
+            raise ray.exceptions.OufOfBandObjectRefSerializationException(msg)

lib/python3.12/site-packages/ray/_common/signature.py

@@ -0,0 +1,163 @@
+import inspect
+import logging
+from inspect import Parameter
+from typing import Any, Dict, List, Tuple
+
+from ray._private.inspect_util import is_cython
+
+# Logger for this module. It should be configured at the entry point
+# into the program using Ray. Ray provides a default configuration at
+# entry/init points.
+logger = logging.getLogger(__name__)
+
+# This dummy type is also defined in ArgumentsBuilder.java. Please keep it
+# synced.
+DUMMY_TYPE = b"__RAY_DUMMY__"
+
+
+def get_signature(func: Any) -> inspect.Signature:
+    """Get signature parameters.
+
+    Support Cython functions by grabbing relevant attributes from the Cython
+    function and attaching to a no-op function. This is somewhat brittle, since
+    inspect may change, but given that inspect is written to a PEP, we hope
+    it is relatively stable. Future versions of Python may allow overloading
+    the inspect 'isfunction' and 'ismethod' functions / create ABC for Python
+    functions. Until then, it appears that Cython won't do anything about
+    compatability with the inspect module.
+
+    Args:
+        func: The function whose signature should be checked.
+
+    Returns:
+        A function signature object, which includes the names of the keyword
+            arguments as well as their default values.
+
+    Raises:
+        TypeError: A type error if the signature is not supported
+    """
+    # The first condition for Cython functions, the latter for Cython instance
+    # methods
+    if is_cython(func):
+        attrs = ["__code__", "__annotations__", "__defaults__", "__kwdefaults__"]
+
+        if all(hasattr(func, attr) for attr in attrs):
+            original_func = func
+
+            def func():
+                return
+
+            for attr in attrs:
+                setattr(func, attr, getattr(original_func, attr))
+        else:
+            raise TypeError(f"{func!r} is not a Python function we can process")
+
+    return inspect.signature(func)
+
+
+def extract_signature(func: Any, ignore_first: bool = False) -> List[Parameter]:
+    """Extract the function signature from the function.
+
+    Args:
+        func: The function whose signature should be extracted.
+        ignore_first: True if the first argument should be ignored. This should
+            be used when func is a method of a class.
+
+    Returns:
+        List of Parameter objects representing the function signature.
+    """
+    signature_parameters = list(get_signature(func).parameters.values())
+
+    if ignore_first:
+        if len(signature_parameters) == 0:
+            raise ValueError(
+                "Methods must take a 'self' argument, but the "
+                f"method '{func.__name__}' does not have one."
+            )
+        signature_parameters = signature_parameters[1:]
+
+    return signature_parameters
+
+
+def validate_args(
+    signature_parameters: List[Parameter], args: Tuple[Any, ...], kwargs: Dict[str, Any]
+) -> None:
+    """Validates the arguments against the signature.
+
+    Args:
+        signature_parameters: The list of Parameter objects
+            representing the function signature, obtained from
+            `extract_signature`.
+        args: The positional arguments passed into the function.
+        kwargs: The keyword arguments passed into the function.
+
+    Raises:
+        TypeError: Raised if arguments do not fit in the function signature.
+    """
+    reconstructed_signature = inspect.Signature(parameters=signature_parameters)
+    try:
+        reconstructed_signature.bind(*args, **kwargs)
+    except TypeError as exc:  # capture a friendlier stacktrace
+        raise TypeError(str(exc)) from None
+
+
+def flatten_args(
+    signature_parameters: List[Parameter], args: Tuple[Any, ...], kwargs: Dict[str, Any]
+) -> List[Any]:
+    """Validates the arguments against the signature and flattens them.
+
+    The flat list representation is a serializable format for arguments.
+    Since the flatbuffer representation of function arguments is a list, we
+    combine both keyword arguments and positional arguments. We represent
+    this with two entries per argument value - [DUMMY_TYPE, x] for positional
+    arguments and [KEY, VALUE] for keyword arguments. See the below example.
+    See `recover_args` for logic restoring the flat list back to args/kwargs.
+
+    Args:
+        signature_parameters: The list of Parameter objects
+            representing the function signature, obtained from
+            `extract_signature`.
+        args: The positional arguments passed into the function.
+        kwargs: The keyword arguments passed into the function.
+
+    Returns:
+        List of args and kwargs. Non-keyword arguments are prefixed
+            by internal enum DUMMY_TYPE.
+
+    Raises:
+        TypeError: Raised if arguments do not fit in the function signature.
+    """
+    validate_args(signature_parameters, args, kwargs)
+    list_args = []
+    for arg in args:
+        list_args += [DUMMY_TYPE, arg]
+
+    for keyword, arg in kwargs.items():
+        list_args += [keyword, arg]
+    return list_args
+
+
+def recover_args(flattened_args: List[Any]) -> Tuple[List[Any], Dict[str, Any]]:
+    """Recreates `args` and `kwargs` from the flattened arg list.
+
+    Args:
+        flattened_args: List of args and kwargs. This should be the output of
+            `flatten_args`.
+
+    Returns:
+        args: The non-keyword arguments passed into the function.
+        kwargs: The keyword arguments passed into the function.
+    """
+    assert (
+        len(flattened_args) % 2 == 0
+    ), "Flattened arguments need to be even-numbered. See `flatten_args`."
+    args = []
+    kwargs = {}
+    for name_index in range(0, len(flattened_args), 2):
+        name, arg = flattened_args[name_index], flattened_args[name_index + 1]
+        if name == DUMMY_TYPE:
+            args.append(arg)
+        else:
+            kwargs[name] = arg
+
+    return args, kwargs

lib/python3.12/site-packages/ray/_common/test_utils.py

@@ -0,0 +1,249 @@
+"""Test utilities for Ray.
+
+This module contains test utility classes that are distributed with the Ray package
+and can be used by external libraries and tests. These utilities must remain in
+_common/ (not in tests/) to be accessible in the Ray package distribution.
+"""
+
+import asyncio
+import inspect
+import os
+import time
+import traceback
+import uuid
+from collections.abc import Awaitable
+from contextlib import contextmanager
+from enum import Enum
+from typing import Any, Callable, Dict, Iterator, List, Optional, Set
+
+import ray
+import ray._common.usage.usage_lib as ray_usage_lib
+import ray._private.utils
+from ray._common.network_utils import build_address
+
+
+@ray.remote(num_cpus=0)
+class SignalActor:
+    """A Ray actor for coordinating test execution through signals.
+
+    Useful for testing async coordination, waiting for specific states,
+    and synchronizing multiple actors or tasks in tests.
+    """
+
+    def __init__(self):
+        self.ready_event = asyncio.Event()
+        self.num_waiters = 0
+
+    def send(self, clear: bool = False):
+        self.ready_event.set()
+        if clear:
+            self.ready_event.clear()
+
+    async def wait(self, should_wait: bool = True):
+        if should_wait:
+            self.num_waiters += 1
+            await self.ready_event.wait()
+            self.num_waiters -= 1
+
+    async def cur_num_waiters(self) -> int:
+        return self.num_waiters
+
+
+@ray.remote(num_cpus=0)
+class Semaphore:
+    """A Ray actor implementing a semaphore for test coordination.
+
+    Useful for testing resource limiting, concurrency control,
+    and coordination between multiple actors or tasks.
+    """
+
+    def __init__(self, value: int = 1):
+        self._sema = asyncio.Semaphore(value=value)
+
+    async def acquire(self):
+        await self._sema.acquire()
+
+    async def release(self):
+        self._sema.release()
+
+    async def locked(self) -> bool:
+        return self._sema.locked()
+
+
+__all__ = ["SignalActor", "Semaphore"]
+
+
+def wait_for_condition(
+    condition_predictor: Callable[..., bool],
+    timeout: float = 10,
+    retry_interval_ms: float = 100,
+    raise_exceptions: bool = False,
+    **kwargs: Any,
+):
+    """Wait until a condition is met or time out with an exception.
+
+    Args:
+        condition_predictor: A function that predicts the condition.
+        timeout: Maximum timeout in seconds.
+        retry_interval_ms: Retry interval in milliseconds.
+        raise_exceptions: If true, exceptions that occur while executing
+            condition_predictor won't be caught and instead will be raised.
+        **kwargs: Arguments to pass to the condition_predictor.
+
+    Returns:
+        None: Returns when the condition is met.
+
+    Raises:
+        RuntimeError: If the condition is not met before the timeout expires.
+    """
+    start = time.time()
+    last_ex = None
+    while time.time() - start <= timeout:
+        try:
+            if condition_predictor(**kwargs):
+                return
+        except Exception:
+            if raise_exceptions:
+                raise
+            last_ex = ray._private.utils.format_error_message(traceback.format_exc())
+        time.sleep(retry_interval_ms / 1000.0)
+    message = "The condition wasn't met before the timeout expired."
+    if last_ex is not None:
+        message += f" Last exception: {last_ex}"
+    raise RuntimeError(message)
+
+
+async def async_wait_for_condition(
+    condition_predictor: Callable[..., Awaitable[bool]],
+    timeout: float = 10,
+    retry_interval_ms: float = 100,
+    **kwargs: Any,
+):
+    """Wait until a condition is met or time out with an exception.
+
+    Args:
+        condition_predictor: A function that predicts the condition.
+        timeout: Maximum timeout in seconds.
+        retry_interval_ms: Retry interval in milliseconds.
+        **kwargs: Arguments to pass to the condition_predictor.
+
+    Returns:
+        None: Returns when the condition is met.
+
+    Raises:
+        RuntimeError: If the condition is not met before the timeout expires.
+    """
+    start = time.time()
+    last_ex = None
+    while time.time() - start <= timeout:
+        try:
+            if inspect.iscoroutinefunction(condition_predictor):
+                if await condition_predictor(**kwargs):
+                    return
+            else:
+                if condition_predictor(**kwargs):
+                    return
+        except Exception as ex:
+            last_ex = ex
+        await asyncio.sleep(retry_interval_ms / 1000.0)
+    message = "The condition wasn't met before the timeout expired."
+    if last_ex is not None:
+        message += f" Last exception: {last_ex}"
+    raise RuntimeError(message)
+
+
+@contextmanager
+def simulate_s3_bucket(
+    port: int = 5002,
+    region: str = "us-west-2",
+) -> Iterator[str]:
+    """Context manager that simulates an S3 bucket and yields the URI.
+
+    Args:
+        port: The port of the localhost endpoint where S3 is being served.
+        region: The S3 region.
+
+    Yields:
+        str: URI for the simulated S3 bucket.
+    """
+    from moto.server import ThreadedMotoServer
+
+    old_env = os.environ
+    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
+    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
+    os.environ["AWS_SECURITY_TOKEN"] = "testing"
+    os.environ["AWS_SESSION_TOKEN"] = "testing"
+
+    s3_server = f"http://{build_address('localhost', port)}"
+    server = ThreadedMotoServer(port=port)
+    server.start()
+    url = f"s3://{uuid.uuid4().hex}?region={region}&endpoint_override={s3_server}"
+    yield url
+    server.stop()
+    os.environ = old_env
+
+
+class TelemetryCallsite(Enum):
+    DRIVER = "driver"
+    ACTOR = "actor"
+    TASK = "task"
+
+
+def _get_library_usages() -> Set[str]:
+    return set(
+        ray_usage_lib.get_library_usages_to_report(
+            ray.experimental.internal_kv.internal_kv_get_gcs_client()
+        )
+    )
+
+
+def _get_extra_usage_tags() -> Dict[str, str]:
+    return ray_usage_lib.get_extra_usage_tags_to_report(
+        ray.experimental.internal_kv.internal_kv_get_gcs_client()
+    )
+
+
+def check_library_usage_telemetry(
+    use_lib_fn: Callable[[], None],
+    *,
+    callsite: TelemetryCallsite,
+    expected_library_usages: List[Set[str]],
+    expected_extra_usage_tags: Optional[Dict[str, str]] = None,
+):
+    """Helper for writing tests to validate library usage telemetry.
+
+    `use_lib_fn` is a callable that will be called from the provided callsite.
+    After calling it, the telemetry data to export will be validated against
+    expected_library_usages and expected_extra_usage_tags.
+    """
+    assert len(_get_library_usages()) == 0, _get_library_usages()
+
+    if callsite == TelemetryCallsite.DRIVER:
+        use_lib_fn()
+    elif callsite == TelemetryCallsite.ACTOR:
+
+        @ray.remote
+        class A:
+            def __init__(self):
+                use_lib_fn()
+
+        a = A.remote()
+        ray.get(a.__ray_ready__.remote())
+    elif callsite == TelemetryCallsite.TASK:
+
+        @ray.remote
+        def f():
+            use_lib_fn()
+
+        ray.get(f.remote())
+    else:
+        assert False, f"Unrecognized callsite: {callsite}"
+
+    library_usages = _get_library_usages()
+    extra_usage_tags = _get_extra_usage_tags()
+
+    assert library_usages in expected_library_usages, library_usages
+    if expected_extra_usage_tags:
+        assert all(
+            [extra_usage_tags[k] == v for k, v in expected_extra_usage_tags.items()]
+        ), extra_usage_tags

lib/python3.12/site-packages/ray/_common/usage/usage_constants.py

@@ -0,0 +1,63 @@
+SCHEMA_VERSION = "0.1"
+
+# The key to store / obtain cluster metadata.
+CLUSTER_METADATA_KEY = b"CLUSTER_METADATA"
+
+# The name of a json file where usage stats will be written.
+USAGE_STATS_FILE = "usage_stats.json"
+
+USAGE_STATS_ENABLED_ENV_VAR = "RAY_USAGE_STATS_ENABLED"
+
+USAGE_STATS_SOURCE_ENV_VAR = "RAY_USAGE_STATS_SOURCE"
+
+USAGE_STATS_SOURCE_OSS = "OSS"
+
+USAGE_STATS_ENABLED_FOR_CLI_MESSAGE = (
+    "Usage stats collection is enabled. To disable this, add `--disable-usage-stats` "
+    "to the command that starts the cluster, or run the following command:"
+    " `ray disable-usage-stats` before starting the cluster. "
+    "See https://docs.ray.io/en/master/cluster/usage-stats.html for more details."
+)
+
+USAGE_STATS_ENABLED_FOR_RAY_INIT_MESSAGE = (
+    "Usage stats collection is enabled. To disable this, run the following command:"
+    " `ray disable-usage-stats` before starting Ray. "
+    "See https://docs.ray.io/en/master/cluster/usage-stats.html for more details."
+)
+
+USAGE_STATS_DISABLED_MESSAGE = "Usage stats collection is disabled."
+
+USAGE_STATS_ENABLED_BY_DEFAULT_FOR_CLI_MESSAGE = (
+    "Usage stats collection is enabled by default without user confirmation "
+    "because this terminal is detected to be non-interactive. "
+    "To disable this, add `--disable-usage-stats` to the command that starts "
+    "the cluster, or run the following command:"
+    " `ray disable-usage-stats` before starting the cluster. "
+    "See https://docs.ray.io/en/master/cluster/usage-stats.html for more details."
+)
+
+USAGE_STATS_ENABLED_BY_DEFAULT_FOR_RAY_INIT_MESSAGE = (
+    "Usage stats collection is enabled by default for nightly wheels. "
+    "To disable this, run the following command:"
+    " `ray disable-usage-stats` before starting Ray. "
+    "See https://docs.ray.io/en/master/cluster/usage-stats.html for more details."
+)
+
+USAGE_STATS_CONFIRMATION_MESSAGE = (
+    "Enable usage stats collection? "
+    "This prompt will auto-proceed in 10 seconds to avoid blocking cluster startup."
+)
+
+LIBRARY_USAGE_SET_NAME = "library_usage_"
+
+HARDWARE_USAGE_SET_NAME = "hardware_usage_"
+
+# Keep in-sync with the same constants defined in usage_stats_client.h
+EXTRA_USAGE_TAG_PREFIX = "extra_usage_tag_"
+USAGE_STATS_NAMESPACE = "usage_stats"
+
+KUBERNETES_SERVICE_HOST_ENV = "KUBERNETES_SERVICE_HOST"
+KUBERAY_ENV = "RAY_USAGE_STATS_KUBERAY_IN_USE"
+
+PROVIDER_KUBERNETES_GENERIC = "kubernetes"
+PROVIDER_KUBERAY = "kuberay"

lib/python3.12/site-packages/ray/_common/usage/usage_lib.py

@@ -0,0 +1,1016 @@
+"""This is the module that is in charge of Ray usage report (telemetry) APIs.
+
+NOTE: Ray's usage report is currently "on by default".
+      One could opt-out, see details at https://docs.ray.io/en/master/cluster/usage-stats.html. # noqa
+
+Ray usage report follows the specification from
+https://docs.ray.io/en/master/cluster/usage-stats.html#usage-stats-collection  # noqa
+
+# Module
+
+The module consists of 2 parts.
+
+## Public API
+It contains public APIs to obtain usage report information.
+APIs will be added before the usage report becomes opt-in by default.
+
+## Internal APIs for usage processing/report
+The telemetry report consists of 5 components. This module is in charge of the top 2 layers.
+
+Report                -> usage_lib
+---------------------
+Usage data processing -> usage_lib
+---------------------
+Data storage          -> Ray API server
+---------------------
+Aggregation           -> Ray API server (currently a dashboard server)
+---------------------
+Usage data collection -> Various components (Ray agent, GCS, etc.) + usage_lib (cluster metadata).
+
+Usage report is currently "off by default". You can enable the report by setting an environment variable
+RAY_USAGE_STATS_ENABLED=1. For example, `RAY_USAGE_STATS_ENABLED=1 ray start --head`.
+Or `RAY_USAGE_STATS_ENABLED=1 python [drivers with ray.init()]`.
+
+"Ray API server (currently a dashboard server)" reports the usage data to https://usage-stats.ray.io/.
+
+Data is reported every hour by default.
+
+Note that it is also possible to configure the interval using the environment variable,
+`RAY_USAGE_STATS_REPORT_INTERVAL_S`.
+
+To see collected/reported data, see `usage_stats.json` inside a temp
+folder (e.g., /tmp/ray/session_[id]/*).
+"""
+import json
+import logging
+import os
+import platform
+import sys
+import threading
+import time
+from dataclasses import asdict, dataclass
+from enum import Enum, auto
+from pathlib import Path
+from typing import Dict, List, Optional, Set
+
+import requests
+import yaml
+
+import ray
+import ray._common.usage.usage_constants as usage_constant
+import ray._private.ray_constants as ray_constants
+from ray._raylet import GcsClient
+from ray.core.generated import gcs_pb2, usage_pb2
+from ray.experimental.internal_kv import (
+    _internal_kv_initialized,
+    _internal_kv_put,
+)
+
+logger = logging.getLogger(__name__)
+TagKey = usage_pb2.TagKey
+
+#################
+# Internal APIs #
+#################
+
+
+@dataclass(init=True)
+class ClusterConfigToReport:
+    cloud_provider: Optional[str] = None
+    min_workers: Optional[int] = None
+    max_workers: Optional[int] = None
+    head_node_instance_type: Optional[str] = None
+    worker_node_instance_types: Optional[List[str]] = None
+
+
+@dataclass(init=True)
+class ClusterStatusToReport:
+    total_num_cpus: Optional[int] = None
+    total_num_gpus: Optional[int] = None
+    total_memory_gb: Optional[float] = None
+    total_object_store_memory_gb: Optional[float] = None
+
+
+@dataclass(init=True)
+class UsageStatsToReport:
+    """Usage stats to report"""
+
+    #: The schema version of the report.
+    schema_version: str
+    #: The source of the data (i.e. OSS).
+    source: str
+    #: When the data is collected and reported.
+    collect_timestamp_ms: int
+    #: The total number of successful reports for the lifetime of the cluster.
+    total_success: Optional[int] = None
+    #: The total number of failed reports for the lifetime of the cluster.
+    total_failed: Optional[int] = None
+    #: The sequence number of the report.
+    seq_number: Optional[int] = None
+    #: The Ray version in use.
+    ray_version: Optional[str] = None
+    #: The Python version in use.
+    python_version: Optional[str] = None
+    #: A random id of the cluster session.
+    session_id: Optional[str] = None
+    #: The git commit hash of Ray (i.e. ray.__commit__).
+    git_commit: Optional[str] = None
+    #: The operating system in use.
+    os: Optional[str] = None
+    #: When the cluster is started.
+    session_start_timestamp_ms: Optional[int] = None
+    #: The cloud provider found in the cluster.yaml file (e.g., aws).
+    cloud_provider: Optional[str] = None
+    #: The min_workers found in the cluster.yaml file.
+    min_workers: Optional[int] = None
+    #: The max_workers found in the cluster.yaml file.
+    max_workers: Optional[int] = None
+    #: The head node instance type found in the cluster.yaml file (e.g., i3.8xlarge).
+    head_node_instance_type: Optional[str] = None
+    #: The worker node instance types found in the cluster.yaml file (e.g., i3.8xlarge).
+    worker_node_instance_types: Optional[List[str]] = None
+    #: The total num of cpus in the cluster.
+    total_num_cpus: Optional[int] = None
+    #: The total num of gpus in the cluster.
+    total_num_gpus: Optional[int] = None
+    #: The total size of memory in the cluster.
+    total_memory_gb: Optional[float] = None
+    #: The total size of object store memory in the cluster.
+    total_object_store_memory_gb: Optional[float] = None
+    #: The Ray libraries that are used (e.g., rllib).
+    library_usages: Optional[List[str]] = None
+    #: The extra tags to report when specified by an
+    #  environment variable RAY_USAGE_STATS_EXTRA_TAGS
+    extra_usage_tags: Optional[Dict[str, str]] = None
+    #: The number of alive nodes when the report is generated.
+    total_num_nodes: Optional[int] = None
+    #: The total number of running jobs excluding internal ones
+    #  when the report is generated.
+    total_num_running_jobs: Optional[int] = None
+    #: The libc version in the OS.
+    libc_version: Optional[str] = None
+    #: The hardwares that are used (e.g. Intel Xeon).
+    hardware_usages: Optional[List[str]] = None
+
+
+@dataclass(init=True)
+class UsageStatsToWrite:
+    """Usage stats to write to `USAGE_STATS_FILE`
+
+    We are writing extra metadata such as the status of report
+    to this file.
+    """
+
+    usage_stats: UsageStatsToReport
+    # Whether or not the last report succeeded.
+    success: bool
+    # The error message of the last report if it happens.
+    error: str
+
+
+class UsageStatsEnabledness(Enum):
+    ENABLED_EXPLICITLY = auto()
+    DISABLED_EXPLICITLY = auto()
+    ENABLED_BY_DEFAULT = auto()
+
+
+_recorded_library_usages = set()
+_recorded_library_usages_lock = threading.Lock()
+_recorded_extra_usage_tags = dict()
+_recorded_extra_usage_tags_lock = threading.Lock()
+
+
+def _add_to_usage_set(set_name: str, value: str):
+    assert _internal_kv_initialized()
+    try:
+        _internal_kv_put(
+            f"{set_name}{value}".encode(),
+            b"",
+            namespace=usage_constant.USAGE_STATS_NAMESPACE.encode(),
+        )
+    except Exception as e:
+        logger.debug(f"Failed to add {value} to usage set {set_name}, {e}")
+
+
+def _get_usage_set(gcs_client, set_name: str) -> Set[str]:
+    try:
+        result = set()
+        usages = gcs_client.internal_kv_keys(
+            set_name.encode(),
+            namespace=usage_constant.USAGE_STATS_NAMESPACE.encode(),
+        )
+        for usage in usages:
+            usage = usage.decode("utf-8")
+            result.add(usage[len(set_name) :])
+
+        return result
+    except Exception as e:
+        logger.debug(f"Failed to get usage set {set_name}, {e}")
+        return set()
+
+
+def _put_library_usage(library_usage: str):
+    _add_to_usage_set(usage_constant.LIBRARY_USAGE_SET_NAME, library_usage)
+
+
+def _put_hardware_usage(hardware_usage: str):
+    _add_to_usage_set(usage_constant.HARDWARE_USAGE_SET_NAME, hardware_usage)
+
+
+def record_extra_usage_tag(
+    key: TagKey, value: str, gcs_client: Optional[GcsClient] = None
+):
+    """Record extra kv usage tag.
+
+    If the key already exists, the value will be overwritten.
+
+    To record an extra tag, first add the key to the TagKey enum and
+    then call this function.
+    It will make a synchronous call to the internal kv store if the tag is updated.
+
+    Args:
+        key: The key of the tag.
+        value: The value of the tag.
+        gcs_client: The GCS client to perform KV operation PUT. Defaults to None.
+            When None, it will try to get the global client from the internal_kv.
+
+    Returns:
+        None
+    """
+    key = TagKey.Name(key).lower()
+    with _recorded_extra_usage_tags_lock:
+        if _recorded_extra_usage_tags.get(key) == value:
+            return
+        _recorded_extra_usage_tags[key] = value
+
+    if not _internal_kv_initialized() and gcs_client is None:
+        # This happens if the record is before ray.init and
+        # no GCS client is used for recording explicitly.
+        return
+
+    _put_extra_usage_tag(key, value, gcs_client)
+
+
+def _put_extra_usage_tag(key: str, value: str, gcs_client: Optional[GcsClient] = None):
+    try:
+        key = f"{usage_constant.EXTRA_USAGE_TAG_PREFIX}{key}".encode()
+        val = value.encode()
+        namespace = usage_constant.USAGE_STATS_NAMESPACE.encode()
+        if gcs_client is not None:
+            # Use the GCS client.
+            gcs_client.internal_kv_put(key, val, namespace=namespace)
+        else:
+            # Use internal kv.
+            assert _internal_kv_initialized()
+            _internal_kv_put(key, val, namespace=namespace)
+    except Exception as e:
+        logger.debug(f"Failed to put extra usage tag, {e}")
+
+
+def record_hardware_usage(hardware_usage: str):
+    """Record hardware usage (e.g. which CPU model is used)"""
+    assert _internal_kv_initialized()
+    _put_hardware_usage(hardware_usage)
+
+
+def record_library_usage(library_usage: str):
+    """Record library usage (e.g. which library is used)"""
+    with _recorded_library_usages_lock:
+        if library_usage in _recorded_library_usages:
+            return
+        _recorded_library_usages.add(library_usage)
+
+    if not _internal_kv_initialized():
+        # This happens if the library is imported before ray.init
+        return
+
+    # Only report lib usage for driver / ray client / workers. Otherwise,
+    # it can be reported if the library is imported from
+    # e.g., API server.
+    if (
+        ray._private.worker.global_worker.mode == ray.SCRIPT_MODE
+        or ray._private.worker.global_worker.mode == ray.WORKER_MODE
+        or ray.util.client.ray.is_connected()
+    ):
+        _put_library_usage(library_usage)
+
+
+def _put_pre_init_library_usages():
+    assert _internal_kv_initialized()
+    # NOTE: When the lib is imported from a worker, ray should
+    # always be initialized, so there's no need to register the
+    # pre init hook.
+    if not (
+        ray._private.worker.global_worker.mode == ray.SCRIPT_MODE
+        or ray.util.client.ray.is_connected()
+    ):
+        return
+
+    for library_usage in _recorded_library_usages:
+        _put_library_usage(library_usage)
+
+
+def _put_pre_init_extra_usage_tags():
+    assert _internal_kv_initialized()
+    for k, v in _recorded_extra_usage_tags.items():
+        _put_extra_usage_tag(k, v)
+
+
+def put_pre_init_usage_stats():
+    _put_pre_init_library_usages()
+    _put_pre_init_extra_usage_tags()
+
+
+def reset_global_state():
+    global _recorded_library_usages, _recorded_extra_usage_tags
+
+    with _recorded_library_usages_lock:
+        _recorded_library_usages = set()
+    with _recorded_extra_usage_tags_lock:
+        _recorded_extra_usage_tags = dict()
+
+
+ray._private.worker._post_init_hooks.append(put_pre_init_usage_stats)
+
+
+def _usage_stats_report_url():
+    # The usage collection server URL.
+    # The environment variable is testing-purpose only.
+    return os.getenv("RAY_USAGE_STATS_REPORT_URL", "https://usage-stats.ray.io/")
+
+
+def _usage_stats_report_interval_s():
+    return int(os.getenv("RAY_USAGE_STATS_REPORT_INTERVAL_S", 3600))
+
+
+def _usage_stats_config_path():
+    return os.getenv(
+        "RAY_USAGE_STATS_CONFIG_PATH", os.path.expanduser("~/.ray/config.json")
+    )
+
+
+def _usage_stats_enabledness() -> UsageStatsEnabledness:
+    # Env var has higher priority than config file.
+    usage_stats_enabled_env_var = os.getenv(usage_constant.USAGE_STATS_ENABLED_ENV_VAR)
+    if usage_stats_enabled_env_var == "0":
+        return UsageStatsEnabledness.DISABLED_EXPLICITLY
+    elif usage_stats_enabled_env_var == "1":
+        return UsageStatsEnabledness.ENABLED_EXPLICITLY
+    elif usage_stats_enabled_env_var is not None:
+        raise ValueError(
+            f"Valid value for {usage_constant.USAGE_STATS_ENABLED_ENV_VAR} "
+            f"env var is 0 or 1, but got {usage_stats_enabled_env_var}"
+        )
+
+    usage_stats_enabled_config_var = None
+    try:
+        with open(_usage_stats_config_path()) as f:
+            config = json.load(f)
+            usage_stats_enabled_config_var = config.get("usage_stats")
+    except FileNotFoundError:
+        pass
+    except Exception as e:
+        logger.debug(f"Failed to load usage stats config {e}")
+
+    if usage_stats_enabled_config_var is False:
+        return UsageStatsEnabledness.DISABLED_EXPLICITLY
+    elif usage_stats_enabled_config_var is True:
+        return UsageStatsEnabledness.ENABLED_EXPLICITLY
+    elif usage_stats_enabled_config_var is not None:
+        raise ValueError(
+            f"Valid value for 'usage_stats' in {_usage_stats_config_path()}"
+            f" is true or false, but got {usage_stats_enabled_config_var}"
+        )
+
+    # Usage stats is enabled by default.
+    return UsageStatsEnabledness.ENABLED_BY_DEFAULT
+
+
+def is_nightly_wheel() -> bool:
+    return ray.__commit__ != "{{RAY_COMMIT_SHA}}" and "dev" in ray.__version__
+
+
+def usage_stats_enabled() -> bool:
+    return _usage_stats_enabledness() is not UsageStatsEnabledness.DISABLED_EXPLICITLY
+
+
+def usage_stats_prompt_enabled():
+    return int(os.getenv("RAY_USAGE_STATS_PROMPT_ENABLED", "1")) == 1
+
+
+def _generate_cluster_metadata(*, ray_init_cluster: bool):
+    """Return a dictionary of cluster metadata.
+
+    Params:
+        ray_init_cluster: Whether the cluster is started by ray.init()
+
+    Returns:
+        A dictionary of cluster metadata.
+    """
+    ray_version, python_version = ray._private.utils.compute_version_info()
+    # These two metadata is necessary although usage report is not enabled
+    # to check version compatibility.
+    metadata = {
+        "ray_version": ray_version,
+        "python_version": python_version,
+        "ray_init_cluster": ray_init_cluster,
+    }
+    # Additional metadata is recorded only when usage stats are enabled.
+    if usage_stats_enabled():
+        metadata.update(
+            {
+                "git_commit": ray.__commit__,
+                "os": sys.platform,
+                "session_start_timestamp_ms": int(time.time() * 1000),
+            }
+        )
+        if sys.platform == "linux":
+            # Record llibc version
+            (lib, ver) = platform.libc_ver()
+            if not lib:
+                metadata.update({"libc_version": "NA"})
+            else:
+                metadata.update({"libc_version": f"{lib}:{ver}"})
+    return metadata
+
+
+def show_usage_stats_prompt(cli: bool) -> None:
+    if not usage_stats_prompt_enabled():
+        return
+
+    from ray.autoscaler._private.cli_logger import cli_logger
+
+    prompt_print = cli_logger.print if cli else print
+
+    usage_stats_enabledness = _usage_stats_enabledness()
+    if usage_stats_enabledness is UsageStatsEnabledness.DISABLED_EXPLICITLY:
+        prompt_print(usage_constant.USAGE_STATS_DISABLED_MESSAGE)
+    elif usage_stats_enabledness is UsageStatsEnabledness.ENABLED_BY_DEFAULT:
+        if not cli:
+            prompt_print(
+                usage_constant.USAGE_STATS_ENABLED_BY_DEFAULT_FOR_RAY_INIT_MESSAGE
+            )
+        elif cli_logger.interactive:
+            enabled = cli_logger.confirm(
+                False,
+                usage_constant.USAGE_STATS_CONFIRMATION_MESSAGE,
+                _default=True,
+                _timeout_s=10,
+            )
+            set_usage_stats_enabled_via_env_var(enabled)
+            # Remember user's choice.
+            try:
+                set_usage_stats_enabled_via_config(enabled)
+            except Exception as e:
+                logger.debug(
+                    f"Failed to persist usage stats choice for future clusters: {e}"
+                )
+            if enabled:
+                prompt_print(usage_constant.USAGE_STATS_ENABLED_FOR_CLI_MESSAGE)
+            else:
+                prompt_print(usage_constant.USAGE_STATS_DISABLED_MESSAGE)
+        else:
+            prompt_print(
+                usage_constant.USAGE_STATS_ENABLED_BY_DEFAULT_FOR_CLI_MESSAGE,
+            )
+    else:
+        assert usage_stats_enabledness is UsageStatsEnabledness.ENABLED_EXPLICITLY
+        prompt_print(
+            usage_constant.USAGE_STATS_ENABLED_FOR_CLI_MESSAGE
+            if cli
+            else usage_constant.USAGE_STATS_ENABLED_FOR_RAY_INIT_MESSAGE
+        )
+
+
+def set_usage_stats_enabled_via_config(enabled) -> None:
+    config = {}
+    try:
+        with open(_usage_stats_config_path()) as f:
+            config = json.load(f)
+        if not isinstance(config, dict):
+            logger.debug(
+                f"Invalid ray config file, should be a json dict but got {type(config)}"
+            )
+            config = {}
+    except FileNotFoundError:
+        pass
+    except Exception as e:
+        logger.debug(f"Failed to load ray config file {e}")
+
+    config["usage_stats"] = enabled
+
+    try:
+        os.makedirs(os.path.dirname(_usage_stats_config_path()), exist_ok=True)
+        with open(_usage_stats_config_path(), "w") as f:
+            json.dump(config, f)
+    except Exception as e:
+        raise Exception(
+            "Failed to "
+            f'{"enable" if enabled else "disable"}'
+            ' usage stats by writing {"usage_stats": '
+            f'{"true" if enabled else "false"}'
+            "} to "
+            f"{_usage_stats_config_path()}"
+        ) from e
+
+
+def set_usage_stats_enabled_via_env_var(enabled) -> None:
+    os.environ[usage_constant.USAGE_STATS_ENABLED_ENV_VAR] = "1" if enabled else "0"
+
+
+def put_cluster_metadata(gcs_client: GcsClient, *, ray_init_cluster: bool) -> dict:
+    """Generate the cluster metadata and store it to GCS.
+
+    It is a blocking API.
+
+    Params:
+        gcs_client: The GCS client to perform KV operation PUT.
+        ray_init_cluster: Whether the cluster is started by ray.init()
+
+    Raises:
+        gRPC exceptions: If PUT fails.
+
+    Returns:
+        The cluster metadata.
+    """
+    metadata = _generate_cluster_metadata(ray_init_cluster=ray_init_cluster)
+    gcs_client.internal_kv_put(
+        usage_constant.CLUSTER_METADATA_KEY,
+        json.dumps(metadata).encode(),
+        overwrite=True,
+        namespace=ray_constants.KV_NAMESPACE_CLUSTER,
+    )
+    return metadata
+
+
+def get_total_num_running_jobs_to_report(gcs_client) -> Optional[int]:
+    """Return the total number of running jobs in the cluster excluding internal ones"""
+    try:
+        result = gcs_client.get_all_job_info(
+            skip_submission_job_info_field=True, skip_is_running_tasks_field=True
+        )
+        total_num_running_jobs = 0
+        for job_info in result.values():
+            if not job_info.is_dead and not job_info.config.ray_namespace.startswith(
+                "_ray_internal"
+            ):
+                total_num_running_jobs += 1
+        return total_num_running_jobs
+    except Exception as e:
+        logger.info(f"Failed to query number of running jobs in the cluster: {e}")
+        return None
+
+
+def get_total_num_alive_nodes_to_report(gcs_client, timeout=None) -> Optional[int]:
+    """Return the total number of alive nodes in the cluster"""
+    try:
+        result = gcs_client.get_all_node_info(
+            timeout=timeout, state_filter=gcs_pb2.GcsNodeInfo.GcsNodeState.ALIVE
+        )
+        return len(result.items())
+    except Exception as e:
+        logger.info(f"Failed to query number of nodes in the cluster: {e}")
+        return None
+
+
+def get_library_usages_to_report(gcs_client) -> List[str]:
+    return list(_get_usage_set(gcs_client, usage_constant.LIBRARY_USAGE_SET_NAME))
+
+
+def get_hardware_usages_to_report(gcs_client) -> List[str]:
+    return list(_get_usage_set(gcs_client, usage_constant.HARDWARE_USAGE_SET_NAME))
+
+
+def get_extra_usage_tags_to_report(gcs_client: GcsClient) -> Dict[str, str]:
+    """Get the extra usage tags from env var and gcs kv store.
+
+    The env var should be given this way; key=value;key=value.
+    If parsing is failed, it will return the empty data.
+
+    Params:
+        gcs_client: The GCS client.
+
+    Returns:
+        Extra usage tags as kv pairs.
+    """
+    extra_usage_tags = dict()
+
+    extra_usage_tags_env_var = os.getenv("RAY_USAGE_STATS_EXTRA_TAGS", None)
+    if extra_usage_tags_env_var:
+        try:
+            kvs = extra_usage_tags_env_var.strip(";").split(";")
+            for kv in kvs:
+                k, v = kv.split("=")
+                extra_usage_tags[k] = v
+        except Exception as e:
+            logger.info(f"Failed to parse extra usage tags env var: {e}")
+
+    valid_tag_keys = [tag_key.lower() for tag_key in TagKey.keys()]
+    try:
+        keys = gcs_client.internal_kv_keys(
+            usage_constant.EXTRA_USAGE_TAG_PREFIX.encode(),
+            namespace=usage_constant.USAGE_STATS_NAMESPACE.encode(),
+        )
+        kv = gcs_client.internal_kv_multi_get(
+            keys, namespace=usage_constant.USAGE_STATS_NAMESPACE.encode()
+        )
+        for key, value in kv.items():
+            key = key.decode("utf-8")
+            key = key[len(usage_constant.EXTRA_USAGE_TAG_PREFIX) :]
+            assert key in valid_tag_keys
+            extra_usage_tags[key] = value.decode("utf-8")
+    except Exception as e:
+        logger.info(f"Failed to get extra usage tags from kv store: {e}")
+    return extra_usage_tags
+
+
+def _get_cluster_status_to_report_v2(gcs_client: GcsClient) -> ClusterStatusToReport:
+    """
+    Get the current status of this cluster. A temporary proxy for the
+    autoscaler v2 API.
+
+    It is a blocking API.
+
+    Params:
+        gcs_client: The GCS client.
+
+    Returns:
+        The current cluster status or empty ClusterStatusToReport
+        if it fails to get that information.
+    """
+    from ray.autoscaler.v2.sdk import get_cluster_status
+
+    result = ClusterStatusToReport()
+    try:
+        cluster_status = get_cluster_status(gcs_client.address)
+        total_resources = cluster_status.total_resources()
+        result.total_num_cpus = int(total_resources.get("CPU", 0))
+        result.total_num_gpus = int(total_resources.get("GPU", 0))
+
+        to_GiB = 1 / 2**30
+        result.total_memory_gb = total_resources.get("memory", 0) * to_GiB
+        result.total_object_store_memory_gb = (
+            total_resources.get("object_store_memory", 0) * to_GiB
+        )
+    except Exception as e:
+        logger.info(f"Failed to get cluster status to report {e}")
+    finally:
+        return result
+
+
+def get_cluster_status_to_report(gcs_client: GcsClient) -> ClusterStatusToReport:
+    """Get the current status of this cluster.
+
+    It is a blocking API.
+
+    Params:
+        gcs_client: The GCS client to perform KV operation GET.
+
+    Returns:
+        The current cluster status or empty if it fails to get that information.
+    """
+    try:
+
+        from ray.autoscaler.v2.utils import is_autoscaler_v2
+
+        if is_autoscaler_v2():
+            return _get_cluster_status_to_report_v2(gcs_client)
+
+        cluster_status = gcs_client.internal_kv_get(
+            ray._private.ray_constants.DEBUG_AUTOSCALING_STATUS.encode(),
+            namespace=None,
+        )
+        if not cluster_status:
+            return ClusterStatusToReport()
+
+        result = ClusterStatusToReport()
+        to_GiB = 1 / 2**30
+        cluster_status = json.loads(cluster_status.decode("utf-8"))
+        if (
+            "load_metrics_report" not in cluster_status
+            or "usage" not in cluster_status["load_metrics_report"]
+        ):
+            return ClusterStatusToReport()
+
+        usage = cluster_status["load_metrics_report"]["usage"]
+        # usage is a map from resource to (used, total) pair
+        if "CPU" in usage:
+            result.total_num_cpus = int(usage["CPU"][1])
+        if "GPU" in usage:
+            result.total_num_gpus = int(usage["GPU"][1])
+        if "memory" in usage:
+            result.total_memory_gb = usage["memory"][1] * to_GiB
+        if "object_store_memory" in usage:
+            result.total_object_store_memory_gb = (
+                usage["object_store_memory"][1] * to_GiB
+            )
+        return result
+    except Exception as e:
+        logger.info(f"Failed to get cluster status to report {e}")
+        return ClusterStatusToReport()
+
+
+def get_cloud_from_metadata_requests() -> str:
+    def cloud_metadata_request(url: str, headers: Optional[Dict[str, str]]) -> bool:
+        try:
+            res = requests.get(url, headers=headers, timeout=1)
+            # Only accept successful responses (200 OK) to avoid false positives like 400 - Bad Request
+            # when multiple cloud providers use the same IP (169.254.169.254)
+            if res.status_code == 200:
+                return True
+        # ConnectionError is a superclass of ConnectTimeout
+        except requests.exceptions.ConnectionError:
+            pass
+        except Exception as e:
+            logger.info(
+                f"Unexpected exception when making cloud provider metadata request: {e}"
+            )
+        return False
+
+    AZURE_METADATA_URL = (
+        "http://169.254.169.254/metadata/instance?api-version=2021-12-13"
+    )
+    AZURE_METADATA_HEADERS = {"Metadata": "true"}
+    GCP_METADATA_URL = "http://metadata.google.internal/computeMetadata/v1"
+    GCP_METADATA_HEADERS = {"Metadata-Flavor": "Google"}
+    AWS_METADATA_URL = "http://169.254.169.254/latest/meta-data/"
+    AWS_METADATA_HEADERS = None
+
+    if cloud_metadata_request(AZURE_METADATA_URL, AZURE_METADATA_HEADERS):
+        return "azure"
+    elif cloud_metadata_request(GCP_METADATA_URL, GCP_METADATA_HEADERS):
+        return "gcp"
+    elif cloud_metadata_request(AWS_METADATA_URL, AWS_METADATA_HEADERS):
+        return "aws"
+    else:
+        return "unknown"
+
+
+def get_cluster_config_to_report(
+    cluster_config_file_path: str,
+) -> ClusterConfigToReport:
+    """Get the static cluster (autoscaler) config used to launch this cluster.
+
+    Params:
+        cluster_config_file_path: The file path to the cluster config file.
+
+    Returns:
+        The cluster (autoscaler) config or empty if it fails to get that information.
+    """
+
+    def get_instance_type(node_config):
+        if not node_config:
+            return None
+        if "InstanceType" in node_config:
+            # aws
+            return node_config["InstanceType"]
+        if "machineType" in node_config:
+            # gcp
+            return node_config["machineType"]
+        if (
+            "azure_arm_parameters" in node_config
+            and "vmSize" in node_config["azure_arm_parameters"]
+        ):
+            return node_config["azure_arm_parameters"]["vmSize"]
+        return None
+
+    try:
+        with open(cluster_config_file_path) as f:
+            config = yaml.safe_load(f)
+            result = ClusterConfigToReport()
+            if "min_workers" in config:
+                result.min_workers = config["min_workers"]
+            if "max_workers" in config:
+                result.max_workers = config["max_workers"]
+
+            if "provider" in config and "type" in config["provider"]:
+                result.cloud_provider = config["provider"]["type"]
+
+            if "head_node_type" not in config:
+                return result
+            if "available_node_types" not in config:
+                return result
+            head_node_type = config["head_node_type"]
+            available_node_types = config["available_node_types"]
+            for available_node_type in available_node_types:
+                if available_node_type == head_node_type:
+                    head_node_instance_type = get_instance_type(
+                        available_node_types[available_node_type].get("node_config")
+                    )
+                    if head_node_instance_type:
+                        result.head_node_instance_type = head_node_instance_type
+                else:
+                    worker_node_instance_type = get_instance_type(
+                        available_node_types[available_node_type].get("node_config")
+                    )
+                    if worker_node_instance_type:
+                        result.worker_node_instance_types = (
+                            result.worker_node_instance_types or set()
+                        )
+                        result.worker_node_instance_types.add(worker_node_instance_type)
+            if result.worker_node_instance_types:
+                result.worker_node_instance_types = list(
+                    result.worker_node_instance_types
+                )
+            return result
+    except FileNotFoundError:
+        # It's a manually started cluster or k8s cluster
+        result = ClusterConfigToReport()
+
+        # Check if we're on Kubernetes
+        if usage_constant.KUBERNETES_SERVICE_HOST_ENV in os.environ:
+            # Check if we're using KubeRay >= 0.4.0.
+            if usage_constant.KUBERAY_ENV in os.environ:
+                result.cloud_provider = usage_constant.PROVIDER_KUBERAY
+            # Else, we're on Kubernetes but not in either of the above categories.
+            else:
+                result.cloud_provider = usage_constant.PROVIDER_KUBERNETES_GENERIC
+
+        # if kubernetes was not set as cloud_provider vs. was set before
+        if result.cloud_provider is None:
+            result.cloud_provider = get_cloud_from_metadata_requests()
+        else:
+            result.cloud_provider += f"_${get_cloud_from_metadata_requests()}"
+
+        return result
+    except Exception as e:
+        logger.info(f"Failed to get cluster config to report {e}")
+        return ClusterConfigToReport()
+
+
+def get_cluster_metadata(gcs_client: GcsClient) -> dict:
+    """Get the cluster metadata from GCS.
+
+    It is a blocking API.
+
+    This will return None if `put_cluster_metadata` was never called.
+
+    Params:
+        gcs_client: The GCS client to perform KV operation GET.
+
+    Returns:
+        The cluster metadata in a dictionary.
+
+    Raises:
+        RuntimeError: If it fails to obtain cluster metadata from GCS.
+    """
+    return json.loads(
+        gcs_client.internal_kv_get(
+            usage_constant.CLUSTER_METADATA_KEY,
+            namespace=ray_constants.KV_NAMESPACE_CLUSTER,
+        ).decode("utf-8")
+    )
+
+
+def is_ray_init_cluster(gcs_client: ray._raylet.GcsClient) -> bool:
+    """Return whether the cluster is started by ray.init()"""
+    cluster_metadata = get_cluster_metadata(gcs_client)
+    return cluster_metadata["ray_init_cluster"]
+
+
+def generate_disabled_report_data() -> UsageStatsToReport:
+    """Generate the report data indicating usage stats is disabled"""
+    data = UsageStatsToReport(
+        schema_version=usage_constant.SCHEMA_VERSION,
+        source=os.getenv(
+            usage_constant.USAGE_STATS_SOURCE_ENV_VAR,
+            usage_constant.USAGE_STATS_SOURCE_OSS,
+        ),
+        collect_timestamp_ms=int(time.time() * 1000),
+    )
+    return data
+
+
+def generate_report_data(
+    cluster_config_to_report: ClusterConfigToReport,
+    total_success: int,
+    total_failed: int,
+    seq_number: int,
+    gcs_address: str,
+    cluster_id: str,
+) -> UsageStatsToReport:
+    """Generate the report data.
+
+    Params:
+        cluster_config_to_report: The cluster (autoscaler)
+            config generated by `get_cluster_config_to_report`.
+        total_success: The total number of successful report
+            for the lifetime of the cluster.
+        total_failed: The total number of failed report
+            for the lifetime of the cluster.
+        seq_number: The sequence number that's incremented whenever
+            a new report is sent.
+        gcs_address: the address of gcs to get data to report.
+        cluster_id: hex id of the cluster.
+
+    Returns:
+        UsageStats
+    """
+    assert cluster_id
+
+    gcs_client = ray._raylet.GcsClient(address=gcs_address, cluster_id=cluster_id)
+
+    cluster_metadata = get_cluster_metadata(gcs_client)
+    cluster_status_to_report = get_cluster_status_to_report(gcs_client)
+
+    data = UsageStatsToReport(
+        schema_version=usage_constant.SCHEMA_VERSION,
+        source=os.getenv(
+            usage_constant.USAGE_STATS_SOURCE_ENV_VAR,
+            usage_constant.USAGE_STATS_SOURCE_OSS,
+        ),
+        collect_timestamp_ms=int(time.time() * 1000),
+        total_success=total_success,
+        total_failed=total_failed,
+        seq_number=seq_number,
+        ray_version=cluster_metadata["ray_version"],
+        python_version=cluster_metadata["python_version"],
+        session_id=cluster_id,
+        git_commit=cluster_metadata["git_commit"],
+        os=cluster_metadata["os"],
+        session_start_timestamp_ms=cluster_metadata["session_start_timestamp_ms"],
+        cloud_provider=cluster_config_to_report.cloud_provider,
+        min_workers=cluster_config_to_report.min_workers,
+        max_workers=cluster_config_to_report.max_workers,
+        head_node_instance_type=cluster_config_to_report.head_node_instance_type,
+        worker_node_instance_types=cluster_config_to_report.worker_node_instance_types,
+        total_num_cpus=cluster_status_to_report.total_num_cpus,
+        total_num_gpus=cluster_status_to_report.total_num_gpus,
+        total_memory_gb=cluster_status_to_report.total_memory_gb,
+        total_object_store_memory_gb=cluster_status_to_report.total_object_store_memory_gb,  # noqa: E501
+        library_usages=get_library_usages_to_report(gcs_client),
+        extra_usage_tags=get_extra_usage_tags_to_report(gcs_client),
+        total_num_nodes=get_total_num_alive_nodes_to_report(gcs_client),
+        total_num_running_jobs=get_total_num_running_jobs_to_report(gcs_client),
+        libc_version=cluster_metadata.get("libc_version"),
+        hardware_usages=get_hardware_usages_to_report(gcs_client),
+    )
+    return data
+
+
+def generate_write_data(
+    usage_stats: UsageStatsToReport,
+    error: str,
+) -> UsageStatsToWrite:
+    """Generate the report data.
+
+    Params:
+        usage_stats: The usage stats that were reported.
+        error: The error message of failed reports.
+
+    Returns:
+        UsageStatsToWrite
+    """
+    data = UsageStatsToWrite(
+        usage_stats=usage_stats,
+        success=error is None,
+        error=error,
+    )
+    return data
+
+
+class UsageReportClient:
+    """The client implementation for usage report.
+
+    It is in charge of writing usage stats to the directory
+    and report usage stats.
+    """
+
+    def write_usage_data(self, data: UsageStatsToWrite, dir_path: str) -> None:
+        """Write the usage data to the directory.
+
+        Params:
+            data: Data to report
+            dir_path: The path to the directory to write usage data.
+        """
+        # Atomically update the file.
+        dir_path = Path(dir_path)
+        destination = dir_path / usage_constant.USAGE_STATS_FILE
+        temp = dir_path / f"{usage_constant.USAGE_STATS_FILE}.tmp"
+        with temp.open(mode="w") as json_file:
+            json_file.write(json.dumps(asdict(data)))
+        if sys.platform == "win32":
+            # Windows 32 doesn't support atomic renaming, so we should delete
+            # the file first.
+            destination.unlink(missing_ok=True)
+        temp.rename(destination)
+
+    def report_usage_data(self, url: str, data: UsageStatsToReport) -> None:
+        """Report the usage data to the usage server.
+
+        Params:
+            url: The URL to update resource usage.
+            data: Data to report.
+
+        Raises:
+            requests.HTTPError: If requests fails.
+        """
+        r = requests.request(
+            "POST",
+            url,
+            headers={"Content-Type": "application/json"},
+            json=asdict(data),
+            timeout=10,
+        )
+        r.raise_for_status()
+        return r

lib/python3.12/site-packages/ray/_common/utils.py

@@ -0,0 +1,368 @@
+import asyncio
+import binascii
+import errno
+import importlib
+import inspect
+import os
+import random
+import string
+import sys
+import tempfile
+from inspect import signature
+from types import ModuleType
+from typing import Any, Coroutine, Dict, Optional, Tuple
+
+import psutil
+
+
+def import_module_and_attr(
+    full_path: str, *, reload_module: bool = False
+) -> Tuple[ModuleType, Any]:
+    """Given a full import path to a module attr, return the imported module and attr.
+
+    If `reload_module` is set, the module will be reloaded using `importlib.reload`.
+
+    Args:
+        full_path: The full import path to the module and attr.
+        reload_module: Whether to reload the module.
+
+    Returns:
+        A tuple of the imported module and attr.
+    """
+    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)
+    if reload_module:
+        importlib.reload(module)
+    return module, getattr(module, attr_name)
+
+
+def import_attr(full_path: str, *, reload_module: bool = False) -> Any:
+    """Given a full import path to a module attr, return the imported attr.
+
+    If `reload_module` is set, the module will be reloaded using `importlib.reload`.
+
+    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
+    """
+    return import_module_and_attr(full_path, reload_module=reload_module)[1]
+
+
+def get_or_create_event_loop() -> asyncio.AbstractEventLoop:
+    """Get a running async event loop if one exists, otherwise create one.
+
+    This function serves as a proxy for the deprecating get_event_loop().
+    It tries to get the running loop first, and if no running loop
+    could be retrieved:
+    - For python version <3.10: it falls back to the get_event_loop
+        call.
+    - For python version >= 3.10: it uses the same python implementation
+        of _get_event_loop() at asyncio/events.py.
+
+    Ideally, one should use high level APIs like asyncio.run() with python
+    version >= 3.7, if not possible, one should create and manage the event
+    loops explicitly.
+    """
+    vers_info = sys.version_info
+    if vers_info.major >= 3 and vers_info.minor >= 10:
+        # This follows the implementation of the deprecating `get_event_loop`
+        # in python3.10's asyncio. See python3.10/asyncio/events.py
+        # _get_event_loop()
+        try:
+            loop = asyncio.get_running_loop()
+            assert loop is not None
+            return loop
+        except RuntimeError as e:
+            # No running loop, relying on the error message as for now to
+            # differentiate runtime errors.
+            assert "no running event loop" in str(e)
+            return asyncio.get_event_loop_policy().get_event_loop()
+
+    return asyncio.get_event_loop()
+
+
+_BACKGROUND_TASKS = set()
+
+
+def run_background_task(coroutine: Coroutine) -> asyncio.Task:
+    """Schedule a task reliably to the event loop.
+
+    This API is used when you don't want to cache the reference of `asyncio.Task`.
+    For example,
+
+    ```
+    get_event_loop().create_task(coroutine(*args))
+    ```
+
+    The above code doesn't guarantee to schedule the coroutine to the event loops
+
+    When using create_task in a  "fire and forget" way, we should keep the references
+    alive for the reliable execution. This API is used to fire and forget
+    asynchronous execution.
+
+    https://docs.python.org/3/library/asyncio-task.html#creating-tasks
+    """
+    task = get_or_create_event_loop().create_task(coroutine)
+    # Add task to the set. This creates a strong reference.
+    _BACKGROUND_TASKS.add(task)
+
+    # To prevent keeping references to finished tasks forever,
+    # make each task remove its own reference from the set after
+    # completion:
+    task.add_done_callback(_BACKGROUND_TASKS.discard)
+    return task
+
+
+# Used in gpu detection
+RESOURCE_CONSTRAINT_PREFIX = "accelerator_type:"
+PLACEMENT_GROUP_BUNDLE_RESOURCE_NAME = "bundle"
+
+
+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'"
+        )
+    elif PLACEMENT_GROUP_BUNDLE_RESOURCE_NAME in resources:
+        raise ValueError(
+            "The resource should not include `bundle` which "
+            f"is reserved for Ray. resources: {resources}"
+        )
+
+    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"] = int(memory)
+    if object_store_memory is not None:
+        resources["object_store_memory"] = object_store_memory
+    if accelerator_type is not None:
+        resources[f"{RESOURCE_CONSTRAINT_PREFIX}{accelerator_type}"] = 0.001
+
+    return resources
+
+
+# Match the standard alphabet used for UUIDs.
+RANDOM_STRING_ALPHABET = string.ascii_lowercase + string.digits
+
+
+def get_random_alphanumeric_string(length: int):
+    """Generates random string of length consisting exclusively of
+    - Lower-case ASCII chars
+    - Digits
+    """
+    return "".join(random.choices(RANDOM_STRING_ALPHABET, k=length))
+
+
+_PRINTED_WARNING = set()
+
+
+def get_call_location(back: int = 1):
+    """
+    Get the location (filename and line number) of a function caller, `back`
+    frames up the stack.
+
+    Args:
+        back: The number of frames to go up the stack, not including this
+            function.
+
+    Returns:
+        A string with the filename and line number of the caller.
+        For example, "myfile.py:123".
+    """
+    stack = inspect.stack()
+    try:
+        frame = stack[back + 1]
+        return f"{frame.filename}:{frame.lineno}"
+    except IndexError:
+        return "UNKNOWN"
+
+
+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 get_ray_address_file(temp_dir: Optional[str]):
+    if temp_dir is None:
+        temp_dir = get_ray_temp_dir()
+    return os.path.join(temp_dir, "ray_current_cluster")
+
+
+def reset_ray_address(temp_dir: Optional[str] = None):
+    address_file = get_ray_address_file(temp_dir)
+    if os.path.exists(address_file):
+        try:
+            os.remove(address_file)
+        except OSError:
+            pass
+
+
+def load_class(path):
+    """Load a class at runtime given a full path.
+
+    Example of the path: mypkg.mysubpkg.myclass
+    """
+    class_data = path.split(".")
+    if len(class_data) < 2:
+        raise ValueError("You need to pass a valid path like mymodule.provider_class")
+    module_path = ".".join(class_data[:-1])
+    class_str = class_data[-1]
+    module = importlib.import_module(module_path)
+    return getattr(module, class_str)
+
+
+def get_system_memory(
+    # For cgroups v1:
+    memory_limit_filename: str = "/sys/fs/cgroup/memory/memory.limit_in_bytes",
+    # For cgroups v2:
+    memory_limit_filename_v2: str = "/sys/fs/cgroup/memory.max",
+):
+    """Return the total amount of system memory in bytes.
+
+    Args:
+        memory_limit_filename: The path to the file that contains the memory
+            limit for the Docker container. Defaults to
+            /sys/fs/cgroup/memory/memory.limit_in_bytes.
+        memory_limit_filename_v2: The path to the file that contains the memory
+            limit for the Docker container in cgroups v2. Defaults to
+            /sys/fs/cgroup/memory.max.
+
+    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().strip())
+    elif os.path.exists(memory_limit_filename_v2):
+        with open(memory_limit_filename_v2, "r") as f:
+            # Don't forget to strip() the newline:
+            max_file = f.read().strip()
+            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 binary_to_hex(identifier):
+    hex_identifier = binascii.hexlify(identifier)
+    hex_identifier = hex_identifier.decode()
+    return hex_identifier
+
+
+def hex_to_binary(hex_identifier):
+    return binascii.unhexlify(hex_identifier)
+
+
+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.
+    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 get_function_args(callable):
+    all_parameters = frozenset(signature(callable).parameters)
+    return list(all_parameters)
+
+
+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.
+        encode_type: The encoding type to use for decoding. Defaults to "utf-8".
+
+    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.")
+    return byte_str.decode(encode_type)