wehub-resource-sync
1126 lines
37 KiB
Python
1126 lines
37 KiB
Python
import logging
|
|
import os
|
|
import sys
|
|
from traceback import format_exception
|
|
from typing import Optional, Union
|
|
|
|
import colorama
|
|
|
|
import ray._private.ray_constants as ray_constants
|
|
import ray.cloudpickle as pickle
|
|
from ray._raylet import ActorID, TaskID, WorkerID
|
|
from ray.core.generated.common_pb2 import (
|
|
PYTHON,
|
|
ActorDiedErrorContext,
|
|
Address,
|
|
ErrorType,
|
|
Language,
|
|
NodeDeathInfo,
|
|
RayException,
|
|
)
|
|
from ray.util.annotations import DeveloperAPI, PublicAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@PublicAPI
|
|
class RayError(Exception):
|
|
"""Super class of all ray exception types."""
|
|
|
|
def to_bytes(self):
|
|
# Extract exc_info from exception object.
|
|
exc_info = (type(self), self, self.__traceback__)
|
|
formatted_exception_string = "\n".join(format_exception(*exc_info))
|
|
return RayException(
|
|
language=PYTHON,
|
|
serialized_exception=pickle.dumps(self),
|
|
formatted_exception_string=formatted_exception_string,
|
|
).SerializeToString()
|
|
|
|
@staticmethod
|
|
def from_bytes(b):
|
|
ray_exception = RayException()
|
|
ray_exception.ParseFromString(b)
|
|
return RayError.from_ray_exception(ray_exception)
|
|
|
|
@staticmethod
|
|
def from_ray_exception(ray_exception):
|
|
if ray_exception.language == PYTHON:
|
|
try:
|
|
return pickle.loads(ray_exception.serialized_exception)
|
|
except Exception:
|
|
# formatted_exception_string is set in to_bytes() above by calling
|
|
# traceback.format_exception() on the original exception. It contains
|
|
# the string representation and stack trace of the original error.
|
|
original_stacktrace = getattr(
|
|
ray_exception,
|
|
"formatted_exception_string",
|
|
"No formatted exception string available.",
|
|
)
|
|
return UnserializableException(original_stacktrace)
|
|
else:
|
|
return CrossLanguageError(ray_exception)
|
|
|
|
|
|
@PublicAPI
|
|
class CrossLanguageError(RayError):
|
|
"""Raised from another language."""
|
|
|
|
def __init__(self, ray_exception):
|
|
super().__init__(
|
|
"An exception raised from {}:\n{}".format(
|
|
Language.Name(ray_exception.language),
|
|
ray_exception.formatted_exception_string,
|
|
)
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class TaskCancelledError(RayError):
|
|
"""Raised when this task is cancelled."""
|
|
|
|
def __init__(
|
|
self, task_id: Optional[TaskID] = None, error_message: Optional[str] = None
|
|
):
|
|
"""Initialize a ``TaskCancelledError``.
|
|
|
|
Args:
|
|
task_id: The TaskID of the function that was directly
|
|
cancelled.
|
|
error_message: Optional additional error message describing why
|
|
the task was cancelled.
|
|
"""
|
|
self.task_id = task_id
|
|
self.error_message = error_message
|
|
|
|
def __str__(self):
|
|
msg = ""
|
|
if self.task_id:
|
|
msg = "Task: " + str(self.task_id) + " was cancelled. "
|
|
if self.error_message:
|
|
msg += self.error_message
|
|
return msg
|
|
|
|
|
|
@PublicAPI
|
|
class RayTaskError(RayError):
|
|
"""Indicates that a task threw an exception during execution.
|
|
|
|
If a task throws an exception during execution, a RayTaskError is stored in
|
|
the object store for each of the task's outputs. When an object is
|
|
retrieved from the object store, the Python method that retrieved it checks
|
|
to see if the object is a RayTaskError and if it is then an exception is
|
|
thrown propagating the error message.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
function_name,
|
|
traceback_str,
|
|
cause,
|
|
proctitle=None,
|
|
pid=None,
|
|
ip=None,
|
|
actor_repr=None,
|
|
actor_id=None,
|
|
):
|
|
"""Initialize a RayTaskError."""
|
|
import ray
|
|
|
|
if proctitle:
|
|
self.proctitle = proctitle
|
|
else:
|
|
self.proctitle = ray._raylet.getproctitle()
|
|
self.pid = pid or os.getpid()
|
|
self.ip = ip or ray.util.get_node_ip_address()
|
|
self.function_name = function_name
|
|
self.traceback_str = traceback_str
|
|
self.actor_repr = actor_repr
|
|
self._actor_id = actor_id
|
|
self.cause = cause
|
|
|
|
try:
|
|
pickle.dumps(cause)
|
|
except (pickle.PicklingError, TypeError) as e:
|
|
err_type = f"{cause.__class__.__module__}.{cause.__class__.__name__}"
|
|
|
|
err_msg = (
|
|
f"Exception {err_type} isn't serializable: {e}.\n"
|
|
f"Original exception details:\n{traceback_str}"
|
|
)
|
|
|
|
logger.exception(
|
|
f"The original cause of the RayTaskError ({err_type}) isn't serializable."
|
|
)
|
|
self.cause = RayError(err_msg)
|
|
|
|
# BaseException implements a __reduce__ method that returns
|
|
# a tuple with the type and the value of self.args.
|
|
# https://stackoverflow.com/a/49715949/2213289
|
|
self.args = (function_name, traceback_str, self.cause, proctitle, pid, ip)
|
|
|
|
assert traceback_str is not None
|
|
|
|
def make_dual_exception_instance(self) -> "RayTaskError":
|
|
"""Makes a object instance that inherits from both RayTaskError and the type of
|
|
`self.cause`. Raises TypeError if the cause class can't be subclassed"""
|
|
# For normal user Exceptions, we subclass from both
|
|
# RayTaskError and the user exception. For ExceptionGroup,
|
|
# we special handle it because it has a different __new__()
|
|
# signature from Exception.
|
|
# Ref: https://docs.python.org/3/library/exceptions.html#exception-groups
|
|
if sys.version_info >= (3, 11) and isinstance(
|
|
self.cause, ExceptionGroup # noqa: F821
|
|
):
|
|
return self._make_exceptiongroup_dual_exception_instance()
|
|
return self._make_normal_dual_exception_instance()
|
|
|
|
def _make_normal_dual_exception_instance(self) -> "RayTaskError":
|
|
cause_cls = self.cause.__class__
|
|
error_msg = str(self)
|
|
|
|
class cls(RayTaskError, cause_cls):
|
|
def __init__(self, cause):
|
|
self.cause = cause
|
|
# Store args separately to avoid writing to user-defined
|
|
# read-only or property-based `args`.
|
|
self._ray_task_error_args = (cause,)
|
|
|
|
@property
|
|
def args(self):
|
|
return self._ray_task_error_args
|
|
|
|
@args.setter
|
|
def args(self, value):
|
|
self._ray_task_error_args = value
|
|
|
|
def __reduce__(self):
|
|
return (cls, self._ray_task_error_args)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self.cause, name)
|
|
|
|
def __str__(self):
|
|
return error_msg
|
|
|
|
name = f"RayTaskError({cause_cls.__name__})"
|
|
cls.__name__ = name
|
|
cls.__qualname__ = name
|
|
|
|
return cls(self.cause)
|
|
|
|
def _make_exceptiongroup_dual_exception_instance(self) -> "RayTaskError":
|
|
cause_cls = self.cause.__class__
|
|
error_msg = str(self)
|
|
|
|
class cls(RayTaskError, cause_cls):
|
|
def __new__(cls, cause):
|
|
self = super().__new__(cls, cause.message, cause.exceptions)
|
|
return self
|
|
|
|
def __init__(self, cause):
|
|
self.cause = cause
|
|
self._ray_task_error_args = (cause,)
|
|
|
|
@property
|
|
def args(self):
|
|
return self._ray_task_error_args
|
|
|
|
@args.setter
|
|
def args(self, value):
|
|
self._ray_task_error_args = value
|
|
|
|
def __reduce__(self):
|
|
return (cls, self._ray_task_error_args)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self.cause, name)
|
|
|
|
def __str__(self):
|
|
return error_msg
|
|
|
|
name = f"RayTaskError({cause_cls.__name__})"
|
|
cls.__name__ = name
|
|
cls.__qualname__ = name
|
|
|
|
return cls(self.cause)
|
|
|
|
def as_instanceof_cause(self):
|
|
"""Returns an exception that's an instance of the cause's class.
|
|
|
|
The returned exception inherits from both RayTaskError and the
|
|
cause class and contains all of the attributes of the cause
|
|
exception.
|
|
|
|
If the cause class can't be subclassed, issues a warning and returns `self`.
|
|
"""
|
|
cause_cls = self.cause.__class__
|
|
if issubclass(RayTaskError, cause_cls):
|
|
return self # already satisfied
|
|
|
|
try:
|
|
return self.make_dual_exception_instance()
|
|
except TypeError as e:
|
|
logger.warning(
|
|
f"User exception type {type(self.cause)} in RayTaskError can't"
|
|
" be subclassed! This exception is raised as"
|
|
" RayTaskError only. You can use `ray_task_error.cause` to"
|
|
f" access the user exception. Failure in subclassing: {e}"
|
|
)
|
|
return self
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Failed to combine RayTaskError with user exception type "
|
|
f"{type(self.cause)}; raising RayTaskError only. This can "
|
|
"happen when the user exception overrides attributes like "
|
|
"`args` or otherwise blocks subclass construction. "
|
|
f"Failure in subclassing: {e}"
|
|
)
|
|
return self
|
|
|
|
def __str__(self):
|
|
"""Format a RayTaskError as a string."""
|
|
lines = self.traceback_str.strip().split("\n")
|
|
out = []
|
|
code_from_internal_file = False
|
|
|
|
# Format tracebacks.
|
|
# Python stacktrace consists of
|
|
# Traceback...: Indicate the next line will be a traceback.
|
|
# File [file_name + line number]
|
|
# code
|
|
# XError: [message]
|
|
# NOTE: For _raylet.pyx (Cython), the code is not always included.
|
|
for i, line in enumerate(lines):
|
|
# Convert traceback to the readable information.
|
|
if line.startswith("Traceback "):
|
|
traceback_line = (
|
|
f"{colorama.Fore.CYAN}"
|
|
f"{self.proctitle}()"
|
|
f"{colorama.Fore.RESET} "
|
|
f"(pid={self.pid}, ip={self.ip}"
|
|
)
|
|
if self.actor_repr:
|
|
traceback_line += (
|
|
f", actor_id={self._actor_id}, repr={self.actor_repr})"
|
|
)
|
|
else:
|
|
traceback_line += ")"
|
|
code_from_internal_file = False
|
|
out.append(traceback_line)
|
|
elif line.startswith(" File ") and (
|
|
"ray/worker.py" in line
|
|
or "ray/_private/" in line
|
|
or "ray/util/tracing/" in line
|
|
or "ray/_raylet.pyx" in line
|
|
):
|
|
# TODO(windows)
|
|
# Process the internal file line.
|
|
# The file line always starts with 2 space and File.
|
|
# https://github.com/python/cpython/blob/0a0a135bae2692d069b18d2d590397fbe0a0d39a/Lib/traceback.py#L421 # noqa
|
|
if "ray._raylet.raise_if_dependency_failed" in line:
|
|
# It means the current task is failed
|
|
# due to the dependency failure.
|
|
# Print out an user-friendly
|
|
# message to explain that..
|
|
out.append(
|
|
" At least one of the input arguments for "
|
|
"this task could not be computed:"
|
|
)
|
|
if i + 1 < len(lines) and lines[i + 1].startswith(" "):
|
|
# If the next line is indented with 2 space,
|
|
# that means it contains internal code information.
|
|
# For example,
|
|
# File [file_name] [line]
|
|
# [code] # if the next line is indented, it is code.
|
|
# Note there there are 4 spaces in the code line.
|
|
code_from_internal_file = True
|
|
elif code_from_internal_file:
|
|
# If the current line is internal file's code,
|
|
# the next line is not code anymore.
|
|
code_from_internal_file = False
|
|
else:
|
|
out.append(line)
|
|
return "\n".join(out)
|
|
|
|
|
|
@PublicAPI
|
|
class LocalRayletDiedError(RayError):
|
|
"""Indicates that the task's local raylet died."""
|
|
|
|
def __str__(self):
|
|
return "The task's local raylet died. Check raylet.out for more information."
|
|
|
|
|
|
@PublicAPI
|
|
class WorkerCrashedError(RayError):
|
|
"""Indicates that the worker died unexpectedly while executing a task."""
|
|
|
|
def __str__(self):
|
|
return (
|
|
"The worker died unexpectedly while executing this task. "
|
|
"Check python-core-worker-*.log files for more information."
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class RayActorError(RayError):
|
|
"""Indicates that the actor has outages unexpectedly before finishing a task.
|
|
|
|
This exception could happen because the actor process is dead, or is unavailable for
|
|
the moment. Ray raises subclasses `ActorDiedError` and `ActorUnavailableError`
|
|
respectively.
|
|
"""
|
|
|
|
BASE_ERROR_MSG = "The actor experienced an error before finishing this task."
|
|
|
|
def __init__(
|
|
self,
|
|
actor_id: str = None,
|
|
error_msg: str = BASE_ERROR_MSG,
|
|
actor_init_failed: bool = False,
|
|
preempted: bool = False,
|
|
):
|
|
#: The actor ID in hex string.
|
|
self.actor_id = actor_id
|
|
#: Whether the actor failed in the middle of __init__.
|
|
self.error_msg = error_msg
|
|
#: The full error message.
|
|
self._actor_init_failed = actor_init_failed
|
|
#: Whether the actor died because the node was preempted.
|
|
self._preempted = preempted
|
|
|
|
def __str__(self) -> str:
|
|
return self.error_msg
|
|
|
|
@property
|
|
def preempted(self) -> bool:
|
|
return self._preempted
|
|
|
|
@property
|
|
def actor_init_failed(self) -> bool:
|
|
return self._actor_init_failed
|
|
|
|
|
|
@DeveloperAPI
|
|
class ActorDiedError(RayActorError):
|
|
"""Indicates that the actor died unexpectedly before finishing a task.
|
|
|
|
This exception could happen either because the actor process dies while
|
|
executing a task, or because a task is submitted to a dead actor.
|
|
"""
|
|
|
|
BASE_ERROR_MSG = "The actor died unexpectedly before finishing this task."
|
|
|
|
def __init__(
|
|
self, cause: Optional[Union[RayTaskError, ActorDiedErrorContext]] = None
|
|
):
|
|
"""Construct a RayActorError by building the arguments.
|
|
|
|
Args:
|
|
cause: The cause of the actor error. ``RayTaskError`` type means
|
|
the actor has died because of an exception within ``__init__``.
|
|
``ActorDiedErrorContext`` means the actor has died because of
|
|
an unexpected system error. None means the cause isn't known.
|
|
Theoretically, this shouldn't happen, but it's there as a
|
|
safety check.
|
|
"""
|
|
|
|
actor_id = None
|
|
error_msg = ActorDiedError.BASE_ERROR_MSG
|
|
actor_init_failed = False
|
|
preempted = False
|
|
|
|
if not cause:
|
|
# Use the defaults above.
|
|
pass
|
|
elif isinstance(cause, RayTaskError):
|
|
actor_init_failed = True
|
|
actor_id = cause._actor_id
|
|
error_msg = (
|
|
"The actor died because of an error"
|
|
" raised in its creation task, "
|
|
f"{cause.__str__()}"
|
|
)
|
|
else:
|
|
# Indicating system-level actor failures.
|
|
assert isinstance(cause, ActorDiedErrorContext)
|
|
error_msg_lines = [ActorDiedError.BASE_ERROR_MSG]
|
|
error_msg_lines.append(f"\tclass_name: {cause.class_name}")
|
|
error_msg_lines.append(f"\tactor_id: {ActorID(cause.actor_id).hex()}")
|
|
# Below items are optional fields.
|
|
if cause.pid != 0:
|
|
error_msg_lines.append(f"\tpid: {cause.pid}")
|
|
if cause.name != "":
|
|
error_msg_lines.append(f"\tname: {cause.name}")
|
|
if cause.ray_namespace != "":
|
|
error_msg_lines.append(f"\tnamespace: {cause.ray_namespace}")
|
|
if cause.node_ip_address != "":
|
|
error_msg_lines.append(f"\tip: {cause.node_ip_address}")
|
|
error_msg_lines.append(cause.error_message)
|
|
if cause.never_started:
|
|
error_msg_lines.append(
|
|
"The actor never ran - it was cancelled before it started running."
|
|
)
|
|
if (
|
|
cause.node_death_info
|
|
and cause.node_death_info.reason
|
|
== NodeDeathInfo.AUTOSCALER_DRAIN_PREEMPTED
|
|
):
|
|
preempted = True
|
|
error_msg = "\n".join(error_msg_lines)
|
|
actor_id = ActorID(cause.actor_id).hex()
|
|
super().__init__(actor_id, error_msg, actor_init_failed, preempted)
|
|
|
|
@staticmethod
|
|
def from_task_error(task_error: RayTaskError):
|
|
return ActorDiedError(task_error)
|
|
|
|
|
|
@DeveloperAPI
|
|
class ActorUnavailableError(RayActorError):
|
|
"""Raised when the actor is temporarily unavailable but may be available later."""
|
|
|
|
def __init__(self, error_message: str, actor_id: Optional[bytes]):
|
|
actor_id = ActorID(actor_id).hex() if actor_id is not None else None
|
|
error_msg = (
|
|
f"The actor {actor_id} is unavailable: {error_message}. The task may or "
|
|
"may not have been executed on the actor."
|
|
)
|
|
actor_init_failed = False
|
|
preempted = False
|
|
|
|
super().__init__(actor_id, error_msg, actor_init_failed, preempted)
|
|
|
|
|
|
@PublicAPI
|
|
class RaySystemError(RayError):
|
|
"""Indicates that Ray encountered a system error.
|
|
|
|
This exception can be thrown when the raylet is killed.
|
|
"""
|
|
|
|
def __init__(self, client_exc, traceback_str=None):
|
|
self.client_exc = client_exc
|
|
self.traceback_str = traceback_str
|
|
|
|
def __str__(self):
|
|
error_msg = f"System error: {self.client_exc}"
|
|
if self.traceback_str:
|
|
error_msg += f"\ntraceback: {self.traceback_str}"
|
|
return error_msg
|
|
|
|
|
|
@PublicAPI
|
|
class AuthenticationError(RayError):
|
|
"""Indicates that an authentication error occurred.
|
|
|
|
Most commonly, this is caused by a missing or mismatching token set on the client
|
|
(e.g., a Ray CLI command interacting with a remote cluster).
|
|
|
|
Only applicable when `RAY_AUTH_MODE` is not set to `disabled`.
|
|
"""
|
|
|
|
def __init__(self, message: str):
|
|
self.message = message
|
|
|
|
# Always hide traceback for cleaner output
|
|
self.__suppress_context__ = True
|
|
super().__init__(message)
|
|
|
|
def __str__(self) -> str:
|
|
# Check if RAY_AUTH_MODE is set to token and add a heads-up if not
|
|
auth_mode_note = ""
|
|
|
|
from ray._private.authentication.authentication_utils import (
|
|
get_authentication_mode_name,
|
|
)
|
|
from ray._raylet import AuthenticationMode, get_authentication_mode
|
|
|
|
current_mode = get_authentication_mode()
|
|
if current_mode != AuthenticationMode.TOKEN:
|
|
mode_name = get_authentication_mode_name(current_mode)
|
|
auth_mode_note = (
|
|
f" Note: RAY_AUTH_MODE is currently '{mode_name}' (not 'token')."
|
|
)
|
|
|
|
help_text = (
|
|
" Ensure that the token for the cluster is available in a local file (e.g., ~/.ray/auth_token or via "
|
|
"RAY_AUTH_TOKEN_PATH) or as the `RAY_AUTH_TOKEN` environment variable. "
|
|
"To generate a token for local development, use `ray get-auth-token --generate` "
|
|
"For remote clusters, ensure that the token is propagated to all nodes of the cluster when token authentication is enabled. "
|
|
"For more information, see: https://docs.ray.io/en/latest/ray-security/token-auth.html"
|
|
)
|
|
return self.message + "." + auth_mode_note + help_text
|
|
|
|
|
|
@DeveloperAPI
|
|
class UserCodeException(RayError):
|
|
"""Indicates that an exception occurred while executing user code.
|
|
For example, this exception can be used to wrap user code exceptions
|
|
from a remote task or actor. The `retry_exceptions` parameter will
|
|
still respect the underlying cause of this exception."""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI
|
|
class ObjectStoreFullError(RayError):
|
|
"""Indicates that the object store is full.
|
|
|
|
This is raised if the attempt to store the object fails
|
|
because the object store is full even after multiple retries.
|
|
"""
|
|
|
|
def __str__(self):
|
|
return super(ObjectStoreFullError, self).__str__() + (
|
|
"\n"
|
|
"The local object store is full of objects that are still in "
|
|
"scope and cannot be evicted. Tip: Use the `ray memory` command "
|
|
"to list active objects in the cluster."
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class OutOfDiskError(RayError):
|
|
"""Indicates that the local disk is full.
|
|
|
|
This is raised if the attempt to store the object fails
|
|
because both the object store and disk are full.
|
|
"""
|
|
|
|
def __str__(self):
|
|
# TODO(scv119): expose more disk usage information and link to a doc.
|
|
return super(OutOfDiskError, self).__str__() + (
|
|
"\n"
|
|
"The object cannot be created because the local object store"
|
|
" is full and the local disk's utilization is over capacity"
|
|
" (95% by default)."
|
|
"Tip: Use `df` on this node to check disk usage and "
|
|
"`ray memory` to check object store memory usage."
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class OutOfMemoryError(RayError):
|
|
"""Indicates that the node is running out of memory and is close to full.
|
|
|
|
This is raised if the node is low on memory and tasks or actors are being
|
|
evicted to free up memory.
|
|
"""
|
|
|
|
# TODO: (clarng) expose the error message string here and format it with proto
|
|
def __init__(self, message):
|
|
self.message = message
|
|
|
|
def __str__(self):
|
|
return self.message
|
|
|
|
|
|
@PublicAPI
|
|
class NodeDiedError(RayError):
|
|
"""Indicates that the node is either dead or unreachable."""
|
|
|
|
# TODO: (clarng) expose the error message string here and format it with proto
|
|
def __init__(self, message):
|
|
self.message = message
|
|
|
|
def __str__(self):
|
|
return self.message
|
|
|
|
|
|
@PublicAPI
|
|
class ObjectLostError(RayError):
|
|
"""Indicates that the object is lost from distributed memory, due to
|
|
node failure or system error.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
object_ref_hex: str,
|
|
owner_address: Optional[bytes],
|
|
call_site: str,
|
|
):
|
|
"""Initialize an ``ObjectLostError``.
|
|
|
|
Args:
|
|
object_ref_hex: Hex ID of the object.
|
|
owner_address: Address of the worker that owns the object, if
|
|
known.
|
|
call_site: Stringified Python call site at which the ``ObjectRef``
|
|
was originally created. Newlines are normalized for display.
|
|
"""
|
|
self.object_ref_hex = object_ref_hex
|
|
self.owner_address = owner_address
|
|
self.call_site = call_site.replace(
|
|
ray_constants.CALL_STACK_LINE_DELIMITER, "\n "
|
|
)
|
|
|
|
def _base_str(self):
|
|
msg = f"Failed to retrieve object {self.object_ref_hex}. "
|
|
if self.call_site:
|
|
msg += f"The ObjectRef was created at: {self.call_site}"
|
|
else:
|
|
msg += (
|
|
"To see information about where this ObjectRef was created "
|
|
"in Python, set the environment variable "
|
|
"RAY_record_ref_creation_sites=1 during `ray start` and "
|
|
"`ray.init()`."
|
|
)
|
|
return msg
|
|
|
|
def __str__(self):
|
|
return (
|
|
self._base_str()
|
|
+ "\n\n"
|
|
+ (
|
|
f"All copies of {self.object_ref_hex} have been lost due to node "
|
|
"failure. Check cluster logs (`/tmp/ray/session_latest/logs`) for "
|
|
"more information about the failure."
|
|
)
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class ObjectFetchTimedOutError(ObjectLostError):
|
|
"""Indicates that an object fetch timed out.
|
|
|
|
Args:
|
|
object_ref_hex: Hex ID of the object.
|
|
"""
|
|
|
|
def __str__(self):
|
|
return (
|
|
self._base_str()
|
|
+ "\n\n"
|
|
+ (
|
|
f"Fetch for object {self.object_ref_hex} timed out because no "
|
|
"locations were found for the object. This may indicate a "
|
|
"system-level bug."
|
|
)
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
class RpcError(RayError):
|
|
"""Indicates an error in the underlying RPC system."""
|
|
|
|
def __init__(self, message, rpc_code=None):
|
|
self.message = message
|
|
self.rpc_code = rpc_code
|
|
|
|
def __str__(self):
|
|
return self.message
|
|
|
|
|
|
@DeveloperAPI
|
|
class ReferenceCountingAssertionError(ObjectLostError, AssertionError):
|
|
"""Indicates that an object has been deleted while there was still a
|
|
reference to it.
|
|
|
|
Args:
|
|
object_ref_hex: Hex ID of the object.
|
|
"""
|
|
|
|
def __str__(self):
|
|
return (
|
|
self._base_str()
|
|
+ "\n\n"
|
|
+ (
|
|
"The object has already been deleted by the reference counting "
|
|
"protocol. This should not happen."
|
|
)
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
class ObjectFreedError(ObjectLostError):
|
|
"""Indicates that an object was manually freed by the application.
|
|
|
|
Attributes:
|
|
object_ref_hex: Hex ID of the object.
|
|
"""
|
|
|
|
def __str__(self):
|
|
return (
|
|
self._base_str()
|
|
+ "\n\n"
|
|
+ (
|
|
"The object was manually freed using the internal `free` call. "
|
|
"Please ensure that `free` is only called once the object is no "
|
|
"longer needed."
|
|
)
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class OwnerDiedError(ObjectLostError):
|
|
"""Indicates that the owner of the object has died while there is still a
|
|
reference to the object.
|
|
|
|
Args:
|
|
object_ref_hex: Hex ID of the object.
|
|
"""
|
|
|
|
def __str__(self):
|
|
log_loc = "`/tmp/ray/session_latest/logs`"
|
|
owner_location = None
|
|
if self.owner_address:
|
|
try:
|
|
addr = Address()
|
|
addr.ParseFromString(self.owner_address)
|
|
ip_addr = addr.ip_address
|
|
worker_id = WorkerID(addr.worker_id)
|
|
node_id = addr.node_id.hex() if addr.node_id else "unknown"
|
|
owner_location = (
|
|
f"Owner worker ID: {worker_id.hex()}, "
|
|
f"owner node ID: {node_id}, "
|
|
f"owner address: {ip_addr}:{addr.port}."
|
|
)
|
|
log_loc = (
|
|
f"`/tmp/ray/session_latest/logs/*{worker_id.hex()}*`"
|
|
f" at IP address {ip_addr}"
|
|
)
|
|
except Exception:
|
|
# Catch all to make sure we always at least print the default
|
|
# message.
|
|
pass
|
|
|
|
return (
|
|
self._base_str()
|
|
+ "\n\n"
|
|
+ (
|
|
"The object's owner has exited. This is the Python "
|
|
"worker that first created the ObjectRef via `.remote()` or "
|
|
"`ray.put()`. "
|
|
f"Check cluster logs ({log_loc}) for more "
|
|
"information about the Python worker failure."
|
|
+ (f"\n\n{owner_location}" if owner_location else "")
|
|
)
|
|
)
|
|
|
|
|
|
@PublicAPI
|
|
class ObjectReconstructionFailedError(ObjectLostError):
|
|
"""Indicates that the object cannot be reconstructed."""
|
|
|
|
REASON_MESSAGES = {
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_MAX_ATTEMPTS_EXCEEDED: (
|
|
"The object cannot be reconstructed because the maximum number of "
|
|
"task retries has been exceeded. "
|
|
"Consider increasing the number of retries using `@ray.remote(max_retries=N)`."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_LINEAGE_EVICTED: (
|
|
"The object cannot be reconstructed because its lineage has been "
|
|
"evicted to reduce memory pressure. "
|
|
"To prevent this error, set the environment variable "
|
|
"RAY_max_lineage_bytes=<bytes> (default 1GB) during `ray start`."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_PUT: (
|
|
"The object cannot be reconstructed because it was created by "
|
|
"ray.put(), which has no task lineage. "
|
|
"To prevent this error, return the value from a task instead."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_RETRIES_DISABLED: (
|
|
"The object cannot be reconstructed because the task was created "
|
|
"with max_retries=0. "
|
|
"Consider enabling retries using `@ray.remote(max_retries=N)`."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_BORROWED: (
|
|
"The object cannot be reconstructed because it crossed an ownership "
|
|
"boundary. Only the owner of an object can trigger reconstruction, "
|
|
"but this worker borrowed the object from another worker."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_REF_NOT_FOUND: (
|
|
"The object cannot be reconstructed because its reference was "
|
|
"not found in the reference counter. "
|
|
"Please file an issue at https://github.com/ray-project/ray/issues."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_TASK_CANCELLED: (
|
|
"The object cannot be reconstructed because the task that would "
|
|
"produce it was cancelled."
|
|
),
|
|
ErrorType.OBJECT_UNRECONSTRUCTABLE_LINEAGE_DISABLED: (
|
|
"The object cannot be reconstructed because lineage reconstruction "
|
|
"is disabled system-wide (object_reconstruction_enabled=False)."
|
|
),
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
object_ref_hex: str,
|
|
reason: "ErrorType" = None,
|
|
reason_message: str = None,
|
|
owner_address: Optional[Address] = None,
|
|
call_site: str = "",
|
|
):
|
|
"""Initialize ObjectReconstructionFailedError.
|
|
|
|
Args:
|
|
object_ref_hex: Hex string of the object reference.
|
|
reason: ErrorType enum value indicating why reconstruction failed.
|
|
reason_message: Human-readable explanation of the failure.
|
|
owner_address: Address of the object's owner.
|
|
call_site: Call site where the object was created.
|
|
"""
|
|
super().__init__(object_ref_hex, owner_address, call_site)
|
|
self.reason = reason
|
|
self.reason_message = reason_message or self.REASON_MESSAGES.get(
|
|
self.reason,
|
|
"Unknown error reason. This should not happen, please file an issue "
|
|
"at https://github.com/ray-project/ray/issues.",
|
|
)
|
|
|
|
def __str__(self):
|
|
base = self._base_str()
|
|
if self.reason_message:
|
|
reason_name = ErrorType.Name(self.reason) if self.reason else "UNKNOWN"
|
|
return base + f"\n\n[{reason_name}] {self.reason_message}"
|
|
return base
|
|
|
|
|
|
@PublicAPI
|
|
class GetTimeoutError(RayError, TimeoutError):
|
|
"""Indicates that a call to the worker timed out."""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI
|
|
class PlasmaObjectNotAvailable(RayError):
|
|
"""Called when an object was not available within the given timeout."""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI
|
|
class AsyncioActorExit(RayError):
|
|
"""Raised when an asyncio actor intentionally exits via exit_actor()."""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI
|
|
class RuntimeEnvSetupError(RayError):
|
|
"""Raised when a runtime environment fails to be set up.
|
|
|
|
Args:
|
|
error_message: The error message that explains
|
|
why runtime env setup has failed.
|
|
"""
|
|
|
|
def __init__(self, error_message: str = None):
|
|
self.error_message = error_message
|
|
|
|
def __str__(self):
|
|
msgs = ["Failed to set up runtime environment."]
|
|
if self.error_message:
|
|
msgs.append(self.error_message)
|
|
return "\n".join(msgs)
|
|
|
|
|
|
@PublicAPI
|
|
class TaskPlacementGroupRemoved(RayError):
|
|
"""Raised when the corresponding placement group was removed."""
|
|
|
|
def __str__(self):
|
|
return "The placement group corresponding to this task has been removed."
|
|
|
|
|
|
@PublicAPI
|
|
class ActorPlacementGroupRemoved(RayError):
|
|
"""Raised when the corresponding placement group was removed."""
|
|
|
|
def __str__(self):
|
|
return "The placement group corresponding to this Actor has been removed."
|
|
|
|
|
|
@PublicAPI
|
|
class PendingCallsLimitExceeded(RayError):
|
|
"""Raised when the pending actor calls exceeds `max_pending_calls` option.
|
|
|
|
This exception could happen probably because the caller calls the callee
|
|
too frequently.
|
|
"""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI
|
|
class TaskUnschedulableError(RayError):
|
|
"""Raised when the task cannot be scheduled.
|
|
|
|
One example is that the node specified through
|
|
NodeAffinitySchedulingStrategy is dead.
|
|
"""
|
|
|
|
def __init__(self, error_message: str):
|
|
self.error_message = error_message
|
|
|
|
def __str__(self):
|
|
return f"The task is not schedulable: {self.error_message}"
|
|
|
|
|
|
@PublicAPI
|
|
class ActorUnschedulableError(RayError):
|
|
"""Raised when the actor cannot be scheduled.
|
|
|
|
One example is that the node specified through
|
|
NodeAffinitySchedulingStrategy is dead.
|
|
"""
|
|
|
|
def __init__(self, error_message: str):
|
|
self.error_message = error_message
|
|
|
|
def __str__(self):
|
|
return f"The actor is not schedulable: {self.error_message}"
|
|
|
|
|
|
@DeveloperAPI
|
|
class ObjectRefStreamEndOfStreamError(RayError):
|
|
"""Raised by streaming generator tasks when there are no more ObjectRefs to
|
|
read.
|
|
"""
|
|
|
|
pass
|
|
|
|
|
|
@DeveloperAPI
|
|
class OufOfBandObjectRefSerializationException(RayError):
|
|
"""Raised when an `ray.ObjectRef` is out of band serialized by
|
|
`ray.cloudpickle`. It is an anti pattern.
|
|
"""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI(stability="alpha")
|
|
class RayChannelError(RaySystemError):
|
|
"""Indicates that Ray encountered a system error related
|
|
to ray.experimental.channel.
|
|
"""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI(stability="alpha")
|
|
class RayChannelTimeoutError(RayChannelError, TimeoutError):
|
|
"""Raised when the Compiled Graph channel operation times out."""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI(stability="alpha")
|
|
class RayCgraphCapacityExceeded(RaySystemError):
|
|
"""Raised when the Compiled Graph channel's buffer is at max capacity"""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI(stability="alpha")
|
|
class RayDirectTransportError(RaySystemError):
|
|
"""Raised when there is an error during a Ray direct transport transfer."""
|
|
|
|
pass
|
|
|
|
|
|
@PublicAPI(stability="alpha")
|
|
class UnserializableException(RayError):
|
|
"""Raised when there is an error deserializing a serialized exception.
|
|
|
|
This occurs when deserializing (unpickling) a previously serialized exception
|
|
fails. In this case, we fall back to raising the string representation of
|
|
the original exception along with its stack trace that was captured at the
|
|
time of serialization.
|
|
|
|
For more details and how to handle this with custom serializers, :ref:`configuring custom exception serializers <custom-exception-serializer>`
|
|
|
|
Args:
|
|
original_stack_trace: The string representation and stack trace of the
|
|
original exception that was captured during serialization.
|
|
"""
|
|
|
|
def __init__(self, original_stack_trace: str):
|
|
self._original_stack_trace = original_stack_trace
|
|
|
|
def __str__(self):
|
|
return (
|
|
"Failed to deserialize exception. Refer to https://docs.ray.io/en/latest/ray-core/objects/serialization.html#custom-serializers-for-exceptions for more information.\n"
|
|
"Original exception:\n"
|
|
f"{self._original_stack_trace}"
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
class ActorAlreadyExistsError(ValueError, RayError):
|
|
"""Raised when a named actor already exists.
|
|
|
|
Note that this error is not only a subclass of RayError, but also a subclass of ValueError, to maintain backward compatibility.
|
|
|
|
Args:
|
|
error_message: The error message that contains information about the actor name and namespace.
|
|
"""
|
|
|
|
def __init__(self, error_message: str):
|
|
super().__init__(error_message)
|
|
self.error_message = error_message
|
|
|
|
def __str__(self):
|
|
return self.error_message
|
|
|
|
|
|
@DeveloperAPI
|
|
class ActorHandleNotFoundError(ValueError, RayError):
|
|
"""Raised when trying to kill an actor handle that doesn't exist.
|
|
|
|
This typically happens when using an actor handle from a previous Ray session
|
|
after calling ray.shutdown() and ray.init().
|
|
|
|
Note that this error is not only a subclass of RayError, but also a subclass of ValueError,
|
|
to maintain backward compatibility.
|
|
|
|
Args:
|
|
error_message: The error message that contains information about the actor handle.
|
|
"""
|
|
|
|
def __init__(self, error_message: str):
|
|
super().__init__(error_message)
|
|
self.error_message = error_message
|
|
|
|
def __str__(self):
|
|
return self.error_message
|
|
|
|
|
|
RAY_EXCEPTION_TYPES = [
|
|
PlasmaObjectNotAvailable,
|
|
RayError,
|
|
RayTaskError,
|
|
WorkerCrashedError,
|
|
RayActorError,
|
|
ObjectStoreFullError,
|
|
ObjectLostError,
|
|
ObjectFetchTimedOutError,
|
|
ReferenceCountingAssertionError,
|
|
ObjectReconstructionFailedError,
|
|
OwnerDiedError,
|
|
GetTimeoutError,
|
|
AsyncioActorExit,
|
|
RuntimeEnvSetupError,
|
|
TaskPlacementGroupRemoved,
|
|
ActorPlacementGroupRemoved,
|
|
PendingCallsLimitExceeded,
|
|
LocalRayletDiedError,
|
|
TaskUnschedulableError,
|
|
ActorDiedError,
|
|
ActorUnschedulableError,
|
|
ActorUnavailableError,
|
|
RayChannelError,
|
|
RayChannelTimeoutError,
|
|
OufOfBandObjectRefSerializationException,
|
|
RayCgraphCapacityExceeded,
|
|
UnserializableException,
|
|
ActorAlreadyExistsError,
|
|
ActorHandleNotFoundError,
|
|
AuthenticationError,
|
|
]
|