diff --git a/doc/source/core/fault-tolerance/actors.md b/doc/source/core/fault-tolerance/actors.md index 1d53a7c2a3bc..cb0d8b0251d5 100644 --- a/doc/source/core/fault-tolerance/actors.md +++ b/doc/source/core/fault-tolerance/actors.md @@ -127,3 +127,30 @@ Retry behavior depends on the value of `retry_exceptions`: 1. The default value, which is `0`. For example, if a method sets `max_task_retries=5` and `retry_exceptions=True`, and the actor sets `max_restarts=2`, Ray executes the method up to six times: once for the initial invocation, and five additional retries. The six invocations might include two actor crashes. After the sixth invocation, a `ray.get` call on the result's object ref raises the exception from the last invocation, or `ray.exceptions.RayActorError` if the actor crashed in the last invocation. + +(actor-restart-on-exceptions)= + +### Restarting an actor on specific exceptions + +Some faults leave the actor process alive while making it unusable, for example a CUDA soft error that poisons the GPU context, or a subprocess that the actor manages dying. Retrying the method in place doesn't help, because the actor keeps the same broken process and state. + +For these faults, set `restart_on_exceptions` to a list of case-insensitive substring patterns. If an actor method, the actor's `__init__`, or one of its (streaming or dynamic) generator methods raises an exception whose formatted text (`f"{type(e).__name__}: {e}"`, including the `__cause__` and `__context__` chain) contains any pattern, the actor worker exits with a system error. Ray then restarts the actor on a fresh process, consuming one of its `max_restarts`, and reruns its constructor. + +```python +import ray + +@ray.remote(max_restarts=3, max_task_retries=2, restart_on_exceptions=["CUDA error"]) +class ModelWorker: + def __init__(self): + self._load_model() + + def predict(self, batch): + ... +``` + +Keep the following in mind: + +- **This option restarts the actor, not the call.** The exception isn't surfaced to the caller as the application exception; the call fails with a `RayActorError` instead. To have Ray rerun the call on the restarted actor, also set `max_task_retries` (or `retry_exceptions` with `max_task_retries` for application-level retries on a healthy actor). +- **In-flight calls fail and actor state is lost.** Other tasks running on the actor fail too, and the restarted actor starts from its constructor, so checkpoint any state you need to recover. Because the failed call may or may not have taken effect, only use this option with actors whose methods are idempotent or externally checkpointed. +- **A persistent failure becomes a restart loop.** If the constructor or the method raises a matching exception every time, Ray restarts the actor until `max_restarts` runs out, then fails the job. Set a finite `max_restarts` so a logic error surfaces instead of looping. +- **Patterns match text, not types.** A broad pattern such as `"error"` can match unrelated exceptions. Prefer specific patterns such as `"CUDA error: device-side assert"`, and remember that the `__cause__`/`__context__` chain is searched too, so a wrapped exception can match on the original fault. diff --git a/python/ray/_common/ray_option_utils.py b/python/ray/_common/ray_option_utils.py index 39edaeea581c..125f64033681 100644 --- a/python/ray/_common/ray_option_utils.py +++ b/python/ray/_common/ray_option_utils.py @@ -133,6 +133,27 @@ def _validate_resources(resources: Optional[Dict[str, float]]) -> Optional[str]: return None +def _validate_restart_on_exceptions(restart_on_exceptions: Any) -> Optional[str]: + if restart_on_exceptions is None: + return None + + # The container type (list / None) is enforced by the Option's + # type_constraint, so here we only validate the elements. + for pattern in restart_on_exceptions: + if not isinstance(pattern, str): + return ( + "restart_on_exceptions must be a list of non-empty strings, but " + f"received an element of type {type(pattern)}." + ) + if not pattern.strip(): + return ( + "restart_on_exceptions must not contain empty or whitespace-only " + "strings." + ) + + return None + + _common_options = { "label_selector": Option((dict, type(None)), lambda x: validate_label_selector(x)), "fallback_strategy": Option( @@ -254,6 +275,11 @@ def issubclass_safe(obj: Any, cls_: type) -> bool: "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), + "restart_on_exceptions": Option( + (list, type(None)), + _validate_restart_on_exceptions, + default_value=None, + ), "namespace": Option((str, type(None))), "get_if_exists": Option(bool, default_value=False), "allow_out_of_order_execution": Option((bool, type(None))), diff --git a/python/ray/_private/worker.py b/python/ray/_private/worker.py index d339a9ad29cf..3e3e400d67b0 100644 --- a/python/ray/_private/worker.py +++ b/python/ray/_private/worker.py @@ -3957,6 +3957,21 @@ def method(self): means restart indefinitely. - ``max_task_retries``: *actors only*. How many times to retry an actor task if it fails due to a system error. + - ``restart_on_exceptions``: *actors only*. List of case-insensitive + exception substring patterns. When an actor method, its + ``__init__``, or one of its (streaming or dynamic) generator + methods raises an exception whose formatted text + (``f"{type(e).__name__}: {e}"``, walking the + ``__cause__``/``__context__`` chain) contains any pattern, the + actor worker exits with a system error so the actor is restarted + on a fresh process (consuming one of its ``max_restarts``) + instead of surfacing the exception to the caller. Intended for + transient process-level faults (e.g. CUDA/GPU soft errors). + This restarts the actor, not the failed call: unless + ``max_task_retries`` also retries the task, the original call + (and other in-flight calls) fail with a ``RayActorError`` and + all actor state is lost. Defaults to ``None`` (disabled); + requires ``max_restarts`` to be non-zero. - ``max_retries``: *remote functions only*. Maximum number of times to rerun the function when the worker crashes unexpectedly. Default is 3; ``-1`` means infinite retries. diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx index e5dcc793f052..e9b5ef80982e 100644 --- a/python/ray/_raylet.pyx +++ b/python/ray/_raylet.pyx @@ -712,6 +712,83 @@ def raise_sys_exit_with_custom_error_message( raise e +class _ActorRestartRequested(Exception): + """Internal signal that `restart_on_exceptions` matched an exception. + + Streaming generators and async methods must carry this signal to the task + thread. Raising SystemExit on the shared event loop would stop it without + resolving the waiting future, hanging the task instead of exiting the worker. + """ + + def __init__(self, message: str) -> None: + super().__init__(message) + self.restart_message = message + + +def _format_restart_exception(exc: BaseException) -> str: + """Format a fault without letting a broken __str__ mask the original error.""" + try: + message = str(exc) + except BaseException: + message = "" + return f"{type(exc).__name__}: {message}" + + +def _exit_for_actor_restart(request: _ActorRestartRequested) -> None: + """Convert a restart request only at the outermost task-thread boundary.""" + try: + raise_sys_exit_with_custom_error_message(request.restart_message, 1) + except SystemExit as error: + raise error from request + + +def _match_restart_exception(exc: BaseException, patterns): + """Return the first `restart_on_exceptions` pattern matched by `exc`. + + Walks the exception and its ``__cause__`` / ``__context__`` chain. For each + exception, performs a case-insensitive substring match of every pattern + against ``f"{type(e).__name__}: {e}"`` (equivalent to the last line of + ``traceback.format_exception_only``). Returns the first (original-cased) + pattern that matches, or ``None`` if nothing matches. + """ + if not patterns: + return None + folded_patterns = [(p, p.casefold()) for p in patterns] + seen = set() + cur = exc + while cur is not None and id(cur) not in seen: + seen.add(id(cur)) + text = _format_restart_exception(cur).casefold() + for original, folded in folded_patterns: + if folded in text: + return original + cur = cur.__cause__ if cur.__cause__ is not None else cur.__context__ + return None + + +def _raise_actor_restart_if_needed(exc: BaseException, function_name: str) -> None: + """Raise a restart request if an actor pattern matches, or propagate one.""" + if isinstance(exc, _ActorRestartRequested): + raise exc + + restart_patterns = [ + pattern.decode("utf-8") for pattern in + CCoreWorkerProcess.GetCoreWorker().GetWorkerContext() + .CurrentActorRestartOnExceptionPatterns() + ] + matched_pattern = _match_restart_exception(exc, restart_patterns) + if matched_pattern is not None: + logger.warning( + "restart_on_exceptions matched pattern %r for an " + "exception raised by %s; exiting the actor worker to " + "trigger a restart.", + matched_pattern, function_name) + raise _ActorRestartRequested( + "Actor worker exiting because restart_on_exceptions " + f"pattern {matched_pattern!r} matched an exception from " + f"{function_name}: {_format_restart_exception(exc)}") from exc + + cdef prepare_args_and_increment_put_refs( Language language, args, c_vector[unique_ptr[CTaskArg]] *args_vector, function_descriptor, @@ -1323,6 +1400,9 @@ cdef report_streaming_generator_exception( generator or raised as an exception. generator_index: The ObjectRef stream index for this exception. """ + # Check both streaming paths before reporting an error to the caller. + _raise_actor_restart_if_needed(e, context.function_name) + worker = ray._private.worker.global_worker cdef: @@ -1724,9 +1804,19 @@ async def execute_streaming_generator_async( completed_normally = True break + except _ActorRestartRequested: + # Interrupt queued work before this context is freed on worker exit. + interrupt_signal_event.set() + raise except Exception as e: # Report the exception to the owner of the task. - report_streaming_generator_exception(context, e, cur_generator_index, None) + try: + report_streaming_generator_exception( + context, e, cur_generator_index, None) + except _ActorRestartRequested: + # Apply the same queued-work cleanup for a newly matched fault. + interrupt_signal_event.set() + raise except BaseException as be: # NOTE: PLEASE READ CAREFULLY BEFORE CHANGING @@ -1972,6 +2062,8 @@ cdef execute_dynamic_generator_and_store_task_outputs( dynamic_returns, generator_id.Binary()) except Exception as error: + # Dynamic generators are consumed in store_task_outputs, not the method. + _raise_actor_restart_if_needed(error, function_name) is_retryable_error[0] = determine_if_retryable( should_retry_exceptions, error, @@ -2282,6 +2374,7 @@ cdef void execute_task( # Special casing these two because Ray can raise them raise except BaseException as e: + _raise_actor_restart_if_needed(e, function_name) is_retryable_error[0] = determine_if_retryable( should_retry_exceptions, e, @@ -2407,6 +2500,11 @@ cdef void execute_task( # Special casing these two because Ray can raise them raise except BaseException as e: + # Convert matched faults and propagated requests at the task boundary. + try: + _raise_actor_restart_if_needed(e, function_name) + except _ActorRestartRequested as request: + _exit_for_actor_restart(request) num_errors_stored = store_task_errors( worker, e, task_exception, actor, actor_id, function_name, task_type, title, caller_address, returns, application_error) @@ -4073,6 +4171,7 @@ cdef class CoreWorker: c_bool enable_tensor_transport, fallback_strategy, int64_t actor_generator_backpressure_num_objects=-1, + restart_on_exceptions=None, ): cdef: CRayFunction ray_function @@ -4088,6 +4187,7 @@ cdef class CoreWorker: unordered_map[c_string, c_string] c_labels CLabelSelector c_label_selector c_vector[CFallbackOption] c_fallback_strategy + c_vector[c_string] c_restart_on_exception_patterns c_string call_site self.python_scheduling_strategy_to_c( @@ -4115,6 +4215,14 @@ cdef class CoreWorker: is_detached_optional = make_optional[c_bool]( True if is_detached else False) + # The user-facing option is `restart_on_exceptions`; internally + # (proto/C++/WorkerContext) it is `restart_on_exception_patterns`, + # mirroring the existing `retry_exceptions` -> + # `retry_exception_allowlist` naming convention. + for pattern in (restart_on_exceptions or []): + c_restart_on_exception_patterns.push_back( + pattern.encode("utf-8")) + with nogil: status = CCoreWorkerProcess.GetCoreWorker().CreateActor( ray_function, args_vector, @@ -4134,7 +4242,8 @@ cdef class CoreWorker: c_labels, c_label_selector, c_fallback_strategy, - actor_generator_backpressure_num_objects), + actor_generator_backpressure_num_objects, + c_restart_on_exception_patterns), extension_data, call_site, &c_actor_id, diff --git a/python/ray/actor.py b/python/ray/actor.py index c4a8d5e04649..37956dd7526f 100644 --- a/python/ray/actor.py +++ b/python/ray/actor.py @@ -1793,6 +1793,23 @@ def method(self): PendingCallsLimitExceeded will be raised for further tasks. Note that this limit is counted per handle. -1 means that the number of pending calls is unlimited. + - ``restart_on_exceptions``: A list of exception substring + patterns. When an actor method or the actor's ``__init__`` + raises an exception whose formatted text + (``f"{type(e).__name__}: {e}"``, including the + ``__cause__``/``__context__`` chain) contains any of these + patterns (case-insensitive substring match), the actor + worker exits with a system error so the actor is restarted + on a fresh process, consuming one of its ``max_restarts``, + instead of surfacing the exception to the caller. Intended + for transient process-level faults (for example CUDA/GPU + soft errors) that require a full worker restart to recover. + Note that this restarts the actor, not the failed call: + unless ``max_task_retries`` also retries the task, the + original call (and any other in-flight calls on the actor) + fails with a ``RayActorError``, and all actor state is lost. + Defaults to ``None``, which disables the feature. Requires + ``max_restarts`` to be non-zero for the restart to happen. - ``max_concurrency``: The max number of concurrent calls to allow for this actor. This only works with direct actor calls. The max concurrency defaults to 1 for threaded @@ -1968,6 +1985,7 @@ def _remote( max_restarts = actor_options["max_restarts"] max_task_retries = actor_options["max_task_retries"] max_pending_calls = actor_options["max_pending_calls"] + restart_on_exceptions = actor_options.get("restart_on_exceptions") # Override enable_task_events to default for actor if not specified (i.e. None) enable_task_events = actor_options.get("enable_task_events") @@ -2218,6 +2236,7 @@ def _remote( actor_generator_backpressure_num_objects=( actor_generator_backpressure_num_objects ), + restart_on_exceptions=restart_on_exceptions, ) if _actor_launch_hook: diff --git a/python/ray/includes/common.pxd b/python/ray/includes/common.pxd index b07f78a59bd7..7f6e691dfb8b 100644 --- a/python/ray/includes/common.pxd +++ b/python/ray/includes/common.pxd @@ -425,7 +425,8 @@ cdef extern from "ray/core_worker/common.h" nogil: const unordered_map[c_string, c_string] &labels, CLabelSelector label_selector, c_vector[CFallbackOption] fallback_strategy, - int64_t actor_generator_backpressure_num_objects) + int64_t actor_generator_backpressure_num_objects, + const c_vector[c_string] &restart_on_exception_patterns) cdef cppclass CPlacementGroupCreationOptions \ "ray::core::PlacementGroupCreationOptions": diff --git a/python/ray/includes/libcoreworker.pxd b/python/ray/includes/libcoreworker.pxd index 1d625f2a20c5..a8707c2fe9a8 100644 --- a/python/ray/includes/libcoreworker.pxd +++ b/python/ray/includes/libcoreworker.pxd @@ -99,6 +99,7 @@ cdef extern from "ray/core_worker/context.h" nogil: c_bool GetCurrentActorShouldExit() const c_string &GetCurrentSerializedRuntimeEnv() int CurrentActorMaxConcurrency() + c_vector[c_string] CurrentActorRestartOnExceptionPatterns() CActorID GetRootDetachedActorID() cdef extern from "ray/core_worker/generator_waiter.h" nogil: diff --git a/python/ray/tests/BUILD.bazel b/python/ray/tests/BUILD.bazel index ad994daa39e4..7ee159c8d3a5 100644 --- a/python/ray/tests/BUILD.bazel +++ b/python/ray/tests/BUILD.bazel @@ -39,6 +39,7 @@ py_test_module_list( "test_actor_lifetime.py", "test_actor_pool.py", "test_actor_resources.py", + "test_actor_restart_on_exceptions.py", "test_advanced.py", "test_advanced_2.py", "test_advanced_3.py", diff --git a/python/ray/tests/test_actor_restart_on_exceptions.py b/python/ray/tests/test_actor_restart_on_exceptions.py new file mode 100644 index 000000000000..d31a5b0a3a9b --- /dev/null +++ b/python/ray/tests/test_actor_restart_on_exceptions.py @@ -0,0 +1,384 @@ +import os +import sys +import time + +import pytest + +import ray + + +class TransientGPUError(Exception): + """Stands in for a CUDA/GPU soft fault that poisons the process.""" + + +@ray.remote(num_cpus=0) +class FailOnceGate: + """Allows one failure across worker restarts; an in-worker gate would reset.""" + + def __init__(self): + self._used = False + + def should_fail(self) -> bool: + if not self._used: + self._used = True + return True + return False + + def calls(self) -> int: + return 1 if self._used else 0 + + +@ray.remote(num_cpus=0) +class AttemptCounter: + """Records attempts outside the actor process being restarted.""" + + def __init__(self): + self.pids = [] + + def record(self, pid): + self.pids.append(pid) + return len(self.pids) + + def get_pids(self): + return self.pids + + +def test_method_matching_exception_triggers_restart(ray_start_regular): + """A matching method exception restarts the actor (does not surface).""" + gate = FailOnceGate.remote() + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class Worker: + def __init__(self, gate) -> None: + self._gate = gate + + def maybe_fail(self) -> str: + if ray.get(self._gate.should_fail.remote()): + raise TransientGPUError("CUDA error: device-side assert") + return "ok" + + def pid(self) -> int: + return os.getpid() + + worker = Worker.remote(gate) + first_pid = ray.get(worker.pid.remote(), timeout=60) + + # ActorUnavailableError or ActorDiedError can surface, depending on timing. + with pytest.raises(ray.exceptions.RayActorError): + ray.get(worker.maybe_fail.remote(), timeout=60) + + second_pid = ray.get(worker.pid.remote(), timeout=60) + assert second_pid != first_pid + assert ray.get(worker.maybe_fail.remote(), timeout=60) == "ok" + assert ray.get(gate.calls.remote(), timeout=60) == 1 + + +def test_non_matching_exception_surfaces_normally(ray_start_regular): + """A non-matching exception is surfaced to the caller, no restart.""" + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class Worker: + def pid(self) -> int: + return os.getpid() + + def raise_value_error(self) -> None: + raise ValueError("ordinary bug") + + worker = Worker.remote() + first_pid = ray.get(worker.pid.remote(), timeout=60) + + with pytest.raises(ValueError): + ray.get(worker.raise_value_error.remote(), timeout=60) + + assert ray.get(worker.pid.remote(), timeout=60) == first_pid + + +def test_init_matching_exception_recovers(ray_start_regular): + """A matching __init__ exception restarts the actor, which then succeeds.""" + gate = FailOnceGate.remote() + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class Worker: + def __init__(self, gate) -> None: + if ray.get(gate.should_fail.remote()): + raise TransientGPUError("CUDA error during init") + + def ping(self) -> str: + return "pong" + + worker = Worker.remote(gate) + # The handle can be temporarily unavailable during constructor recovery. + deadline = time.time() + 60 + while True: + try: + assert ray.get(worker.ping.remote(), timeout=60) == "pong" + break + except ray.exceptions.RayActorError: + assert time.time() < deadline, "actor did not come back after restart" + time.sleep(0.5) + assert ray.get(gate.calls.remote(), timeout=60) == 1 + + +def test_matching_exception_exhausts_max_restarts(ray_start_regular): + """A matching __init__ exception consumes max_restarts and eventually fails.""" + + from ray._private.test_utils import wait_for_condition + + counter = AttemptCounter.remote() + + @ray.remote(max_restarts=1, restart_on_exceptions=["TransientGPUError"]) + class AlwaysFailsInit: + def __init__(self, counter) -> None: + ray.get(counter.record.remote(os.getpid()), timeout=30) + raise TransientGPUError("CUDA error during init") + + def ping(self) -> str: + return "pong" + + actor = AlwaysFailsInit.remote(counter) + actor_id = actor._actor_id.hex() + + def permanently_dead(): + info = ray._private.state.actors(actor_id) + return info is not None and info["State"] == "DEAD" + + wait_for_condition(permanently_dead, timeout=60) + pids = ray.get(counter.get_pids.remote(), timeout=30) + assert len(pids) == 2 + assert pids[0] != pids[1] + for _ in range(2): + with pytest.raises(ray.exceptions.ActorDiedError): + ray.get(actor.ping.remote(), timeout=30) + assert permanently_dead() + assert ray.get(counter.get_pids.remote(), timeout=30) == pids + + +def test_max_task_retries_reruns_failed_call_on_restarted_actor(ray_start_regular): + """Matching restart takes precedence over retry_exceptions in the same process.""" + counter = AttemptCounter.remote() + + @ray.remote( + max_restarts=1, + max_task_retries=1, + restart_on_exceptions=["TransientGPUError"], + ) + class Worker: + def __init__(self, counter) -> None: + self._counter = counter + + def pid(self): + return os.getpid() + + @ray.method(retry_exceptions=True) + def maybe_fail(self): + if ray.get(self._counter.record.remote(os.getpid()), timeout=30) == 1: + raise TransientGPUError("CUDA error: device-side assert") + return os.getpid() + + worker = Worker.remote(counter) + first_pid = ray.get(worker.pid.remote(), timeout=60) + second_pid = ray.get(worker.maybe_fail.remote(), timeout=60) + assert second_pid != first_pid + assert ray.get(counter.get_pids.remote(), timeout=30) == [first_pid, second_pid] + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +def test_partial_stream_retries_same_call(ray_start_regular, is_async): + counter = AttemptCounter.remote() + + def sync_stream(self): + attempt = ray.get(self.counter.record.remote(os.getpid()), timeout=30) + yield 0 + yield 1 + if attempt == 1: + raise TransientGPUError("CUDA fault after partial output") + yield 2 + yield 3 + + async def async_stream(self): + attempt = await self.counter.record.remote(os.getpid()) + yield 0 + yield 1 + if attempt == 1: + raise TransientGPUError("CUDA fault after partial output") + yield 2 + yield 3 + + @ray.remote( + max_restarts=1, + max_task_retries=1, + restart_on_exceptions=["TransientGPUError"], + ) + class Worker: + def __init__(self, counter): + self.counter = counter + + def pid(self): + return os.getpid() + + stream = ray.method(num_returns="streaming")( + async_stream if is_async else sync_stream + ) + + worker = Worker.remote(counter) + first_pid = ray.get(worker.pid.remote(), timeout=60) + # Consume one invocation, not a manually resubmitted stream after failure. + stream = worker.stream.remote() + values = [] + while True: + try: + ref = stream._next_sync(timeout_s=60) + except StopIteration: + break + assert not ref.is_nil(), "timed out waiting for retried stream" + values.append(ray.get(ref, timeout=30)) + assert len(values) <= 4, "duplicate output during replay" + assert values == [0, 1, 2, 3] + second_pid = ray.get(worker.pid.remote(), timeout=60) + assert second_pid != first_pid + assert ray.get(counter.get_pids.remote(), timeout=30) == [first_pid, second_pid] + + +@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"]) +def test_streaming_generator_matching_exception_triggers_restart( + ray_start_regular, is_async +): + """Without task retries, the failed stream errors; a new call recovers.""" + gate = FailOnceGate.remote() + + def sync_stream(self): + if ray.get(self._gate.should_fail.remote()): + raise TransientGPUError("CUDA error: device-side assert in stream") + yield 0 + yield 1 + + async def async_stream(self): + if await self._gate.should_fail.remote(): + raise TransientGPUError("CUDA error: device-side assert in stream") + yield 0 + yield 1 + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class StreamingWorker: + def __init__(self, gate) -> None: + self._gate = gate + + def pid(self) -> int: + return os.getpid() + + stream = ray.method(num_returns="streaming")( + async_stream if is_async else sync_stream + ) + + worker = StreamingWorker.remote(gate) + first_pid = ray.get(worker.pid.remote(), timeout=60) + + with pytest.raises(ray.exceptions.RayActorError): + ray.get(next(worker.stream.remote()), timeout=60) + + second_pid = ray.get(worker.pid.remote(), timeout=60) + assert second_pid != first_pid + gen = worker.stream.remote() + assert ray.get(next(gen), timeout=60) == 0 + assert ray.get(next(gen), timeout=60) == 1 + + +def test_streaming_generator_non_matching_exception_reports_error(ray_start_regular): + """A non-matching generator exception is reported, the actor survives.""" + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class StreamingWorker: + def pid(self) -> int: + return os.getpid() + + @ray.method(num_returns="streaming") + def stream(self, fail: bool): + if fail: + raise ValueError("ordinary bug") + yield 0 + + worker = StreamingWorker.remote() + first_pid = ray.get(worker.pid.remote(), timeout=60) + + with pytest.raises(ValueError): + ray.get(next(worker.stream.remote(True)), timeout=60) + + assert ray.get(worker.pid.remote(), timeout=60) == first_pid + gen = worker.stream.remote(False) + assert ray.get(next(gen), timeout=60) == 0 + + +def test_static_generator_matching_exception_triggers_restart(ray_start_regular): + """A matching exception from a `num_returns=2` generator restarts the actor.""" + gate = FailOnceGate.remote() + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class Worker: + def __init__(self, gate) -> None: + self._gate = gate + + def pid(self) -> int: + return os.getpid() + + @ray.method(num_returns=2) + def two_outputs(self): + if ray.get(self._gate.should_fail.remote()): + raise TransientGPUError("CUDA error: device-side assert") + yield 0 + yield 1 + + worker = Worker.remote(gate) + first_pid = ray.get(worker.pid.remote(), timeout=60) + + refs = worker.two_outputs.remote() + with pytest.raises(ray.exceptions.RayActorError): + ray.get(refs[0], timeout=60) + + second_pid = ray.get(worker.pid.remote(), timeout=60) + assert second_pid != first_pid + + refs = worker.two_outputs.remote() + assert ray.get(refs[0], timeout=60) == 0 + assert ray.get(refs[1], timeout=60) == 1 + + +def test_dynamic_generator_matching_exception_triggers_restart(ray_start_regular): + """A matching exception from a `num_returns="dynamic"` generator restarts it.""" + gate = FailOnceGate.remote() + + @ray.remote(max_restarts=2, restart_on_exceptions=["TransientGPUError"]) + class Worker: + def __init__(self, gate) -> None: + self._gate = gate + + def pid(self) -> int: + return os.getpid() + + @ray.method(num_returns="dynamic") + def dynamic_outputs(self, n: int): + if ray.get(self._gate.should_fail.remote()): + raise TransientGPUError("CUDA error: device-side assert") + for i in range(n): + yield i + + worker = Worker.remote(gate) + first_pid = ray.get(worker.pid.remote(), timeout=60) + + # `num_returns="dynamic"` first returns an ObjectRef to a generator of + # ObjectRefs; the worker exit surfaces while fetching that generator. + with pytest.raises(ray.exceptions.RayActorError): + gen = ray.get( + worker.dynamic_outputs.remote(2), + timeout=60, + ) + + second_pid = ray.get(worker.pid.remote(), timeout=60) + assert second_pid != first_pid + + gen = ray.get(worker.dynamic_outputs.remote(2), timeout=60) + values = [ray.get(ref, timeout=60) for ref in gen] + assert values == [0, 1] + + +if __name__ == "__main__": + sys.exit(pytest.main(["-sv", __file__])) diff --git a/python/ray/tests/unit/test_actor_restart_exception_matching.py b/python/ray/tests/unit/test_actor_restart_exception_matching.py new file mode 100644 index 000000000000..eaa66f802ef7 --- /dev/null +++ b/python/ray/tests/unit/test_actor_restart_exception_matching.py @@ -0,0 +1,130 @@ +import sys + +import pytest + +from ray._common.ray_option_utils import ( + validate_actor_options, + validate_task_options, +) + + +@pytest.fixture +def match(): + from ray._raylet import _match_restart_exception + + return _match_restart_exception + + +def test_validate_restart_on_exceptions_accepts_valid_values(): + validate_actor_options({"restart_on_exceptions": None}, in_options=True) + validate_actor_options( + {"restart_on_exceptions": ["CUDA error", "out of memory"]}, + in_options=True, + ) + + +def test_validate_restart_on_exceptions_rejects_invalid_values(): + with pytest.raises(TypeError): + validate_actor_options( + {"restart_on_exceptions": ("CUDA error",)}, in_options=True + ) + with pytest.raises(ValueError): + validate_actor_options({"restart_on_exceptions": ["ok", 123]}, in_options=True) + with pytest.raises(ValueError): + validate_actor_options({"restart_on_exceptions": [" "]}, in_options=True) + + +def test_restart_on_exceptions_is_rejected_for_tasks(): + with pytest.raises(ValueError, match="restart_on_exceptions"): + validate_task_options( + {"restart_on_exceptions": ["CUDA error"]}, in_options=True + ) + + +def test_matcher_matches_case_insensitive_substring(match): + exc = RuntimeError("CUDA Error: Out Of Memory") + assert match(exc, []) is None + assert match(exc, ["disk full"]) is None + assert match(exc, ["nope", "out of memory"]) == "out of memory" + + +def test_matcher_walks_cause_chain(match): + try: + try: + raise ValueError("inner cuda failure") + except ValueError as inner: + raise RuntimeError("outer wrapper") from inner + except RuntimeError as outer: + exc = outer + assert match(exc, ["inner cuda failure"]) == "inner cuda failure" + + +@pytest.mark.parametrize("raises_on_bool", [False, True]) +def test_matcher_does_not_evaluate_cause_truthiness(match, raises_on_bool): + class CauseError(Exception): + def __bool__(self): + if raises_on_bool: + raise RuntimeError("exception truthiness must not be evaluated") + return False + + exc = RuntimeError("outer wrapper") + exc.__cause__ = CauseError("CUDA fault") + exc.__context__ = ValueError("unrelated context") + assert match(exc, ["CUDA fault"]) == "CUDA fault" + assert match(exc, ["unrelated context"]) is None + + +def test_matcher_walks_suppressed_context_chain(match): + try: + try: + raise ValueError("cuda failure") + except ValueError: + # `from None` suppresses the context for display, but the matcher + # still walks it so wrapped faults are caught. + raise RuntimeError("outer wrapper") from None + except RuntimeError as outer: + exc = outer + assert match(exc, ["cuda failure"]) == "cuda failure" + + +def test_matcher_terminates_on_exception_cycles(match): + exc = RuntimeError("cycle") + exc.__cause__ = exc + assert match(exc, ["cycle"]) == "cycle" + + +@pytest.mark.parametrize("str_error", [RuntimeError, SystemExit]) +def test_safe_restart_formatter(match, str_error): + from ray._raylet import _format_restart_exception + + class BrokenStringError(Exception): + def __str__(self): + raise str_error("formatting failed") + + exc = BrokenStringError() + assert _format_restart_exception(exc) == ( + "BrokenStringError: " + ) + assert match(exc, ["brokenstringerror"]) == "brokenstringerror" + exc.__cause__ = ValueError("inner CUDA fault") + assert match(exc, ["inner cuda"]) == "inner cuda" + assert match(exc, ["unrelated"]) is None + assert _format_restart_exception(ValueError("ordinary")) == "ValueError: ordinary" + + +def test_restart_exit_preserves_cause(): + from ray._raylet import _ActorRestartRequested, _exit_for_actor_restart + + original = ValueError("original fault") + request = _ActorRestartRequested("restart reason") + request.__cause__ = original + with pytest.raises(SystemExit) as caught: + _exit_for_actor_restart(request) + assert caught.value.code == 1 + assert caught.value.ray_terminate_msg == "restart reason" + assert caught.value.__cause__ is request + assert caught.value.__cause__.__cause__ is original + + +if __name__ == "__main__": + sys.exit(pytest.main(["-sv", __file__])) diff --git a/src/ray/common/task/task_spec.cc b/src/ray/common/task/task_spec.cc index 72dc232a57b6..a0e16a2423dc 100644 --- a/src/ray/common/task/task_spec.cc +++ b/src/ray/common/task/task_spec.cc @@ -471,6 +471,14 @@ std::vector TaskSpecification::DynamicWorkerOptions() const { message_->actor_creation_task_spec().dynamic_worker_options()); } +std::vector TaskSpecification::RestartOnExceptionPatterns() const { + if (!IsActorCreationTask()) { + return {}; + } + return VectorFromProtobuf( + message_->actor_creation_task_spec().restart_on_exception_patterns()); +} + absl::flat_hash_map TaskSpecification::GetLabels() const { return MapFromProtobuf(message_->labels()); } diff --git a/src/ray/common/task/task_spec.h b/src/ray/common/task/task_spec.h index 9c9e7c3544d2..6de6267dc2a5 100644 --- a/src/ray/common/task/task_spec.h +++ b/src/ray/common/task/task_spec.h @@ -318,6 +318,13 @@ class TaskSpecification : public MessageWrapper { std::vector DynamicWorkerOptionsOrEmpty() const; + /** + * @brief Get exception substring patterns that trigger actor restarts. + * @return The configured patterns, or an empty vector if none are configured or + * this is not an actor creation task. + */ + std::vector RestartOnExceptionPatterns() const; + absl::flat_hash_map GetLabels() const; // Methods specific to actor tasks. diff --git a/src/ray/common/task/task_util.h b/src/ray/common/task/task_util.h index d39535a20a27..3f7e97b2fe79 100644 --- a/src/ray/common/task/task_util.h +++ b/src/ray/common/task/task_util.h @@ -267,7 +267,8 @@ class TaskSpecBuilder { const std::string &extension_data = "", bool allow_out_of_order_execution = false, ActorID root_detached_actor_id = ActorID::Nil(), - int64_t actor_generator_backpressure_num_objects = -1) { + int64_t actor_generator_backpressure_num_objects = -1, + const std::vector &restart_on_exception_patterns = {}) { message_->set_type(TaskType::ACTOR_CREATION_TASK); auto actor_creation_spec = message_->mutable_actor_creation_task_spec(); actor_creation_spec->set_actor_id(actor_id.Binary()); @@ -296,6 +297,9 @@ class TaskSpecBuilder { actor_creation_spec->set_allow_out_of_order_execution(allow_out_of_order_execution); actor_creation_spec->set_actor_generator_backpressure_num_objects( actor_generator_backpressure_num_objects); + for (const auto &pattern : restart_on_exception_patterns) { + actor_creation_spec->add_restart_on_exception_patterns(pattern); + } message_->mutable_scheduling_strategy()->CopyFrom(scheduling_strategy); if (!root_detached_actor_id.IsNil()) { message_->set_root_detached_actor_id(root_detached_actor_id.Binary()); diff --git a/src/ray/core_worker/common.h b/src/ray/core_worker/common.h index ee1e68623908..96f2d32ad8c6 100644 --- a/src/ray/core_worker/common.h +++ b/src/ray/core_worker/common.h @@ -144,7 +144,8 @@ struct ActorCreationOptions { std::unordered_map labels_p = {}, LabelSelector label_selector_p = {}, std::vector fallback_strategy_p = {}, - int64_t actor_generator_backpressure_num_objects_p = -1) + int64_t actor_generator_backpressure_num_objects_p = -1, + std::vector restart_on_exception_patterns_p = {}) : max_restarts(max_restarts_p), max_task_retries(max_task_retries_p), max_concurrency(max_concurrency_p), @@ -167,7 +168,8 @@ struct ActorCreationOptions { label_selector(std::move(label_selector_p)), fallback_strategy(std::move(fallback_strategy_p)), actor_generator_backpressure_num_objects( - actor_generator_backpressure_num_objects_p) { + actor_generator_backpressure_num_objects_p), + restart_on_exception_patterns(std::move(restart_on_exception_patterns_p)) { // Check that resources is a subset of placement resources. for (auto &resource : resources) { auto it = this->placement_resources.find(resource.first); @@ -232,6 +234,10 @@ struct ActorCreationOptions { // on this actor. -1 disables the cap. See proto field // ActorCreationTaskSpec.actor_generator_backpressure_num_objects. const int64_t actor_generator_backpressure_num_objects = -1; + /// Exception patterns that, when matched (case-insensitive substring) against an + /// exception raised by an actor method or __init__, cause the worker to exit with a + /// system error so the actor is restarted. Empty disables the feature. + const std::vector restart_on_exception_patterns; }; using PlacementStrategy = rpc::PlacementStrategy; diff --git a/src/ray/core_worker/context.cc b/src/ray/core_worker/context.cc index b70f33d17d51..65fa0ce6649e 100644 --- a/src/ray/core_worker/context.cc +++ b/src/ray/core_worker/context.cc @@ -18,6 +18,7 @@ #include #include +#include #include "absl/strings/str_format.h" #include "ray/common/runtime_env_common.h" @@ -321,6 +322,7 @@ void WorkerContext::SetCurrentTask(const TaskSpecification &task_spec) { current_actor_is_direct_call_ = true; current_actor_max_concurrency_ = task_spec.MaxActorConcurrency(); current_actor_is_asyncio_ = task_spec.IsAsyncioActor(); + current_actor_restart_on_exception_patterns_ = task_spec.RestartOnExceptionPatterns(); is_detached_actor_ = task_spec.IsDetachedActor(); current_actor_placement_group_id_ = task_spec.PlacementGroupBundleId().first; placement_group_capture_child_tasks_ = task_spec.PlacementGroupCaptureChildTasks(); @@ -433,6 +435,11 @@ bool WorkerContext::CurrentActorIsAsync() const { return current_actor_is_asyncio_; } +std::vector WorkerContext::CurrentActorRestartOnExceptionPatterns() const { + absl::ReaderMutexLock lock(&mutex_); + return current_actor_restart_on_exception_patterns_; +} + void WorkerContext::SetCurrentActorShouldExit() { absl::WriterMutexLock lock(&mutex_); RAY_CHECK(!current_actor_id_.IsNil()) diff --git a/src/ray/core_worker/context.h b/src/ray/core_worker/context.h index 8957bb211d1b..c00de68fd7c5 100644 --- a/src/ray/core_worker/context.h +++ b/src/ray/core_worker/context.h @@ -17,6 +17,7 @@ #include #include #include +#include #include "absl/base/thread_annotations.h" #include "absl/synchronization/mutex.h" @@ -167,6 +168,13 @@ class WorkerContext { bool CurrentActorIsAsync() const ABSL_LOCKS_EXCLUDED(mutex_); + /** + * @brief Get exception substring patterns that trigger restarts for the current actor. + * @return The configured patterns, or an empty vector if no actor patterns are set. + */ + std::vector CurrentActorRestartOnExceptionPatterns() const + ABSL_LOCKS_EXCLUDED(mutex_); + /** * @brief Set a flag to indicate that the current actor should exit, it'll be checked * periodically and the actor will exit if the flag is set. @@ -208,6 +216,8 @@ class WorkerContext { ActorID current_actor_id_ ABSL_GUARDED_BY(mutex_); int current_actor_max_concurrency_ ABSL_GUARDED_BY(mutex_) = 1; bool current_actor_is_asyncio_ ABSL_GUARDED_BY(mutex_) = false; + std::vector current_actor_restart_on_exception_patterns_ + ABSL_GUARDED_BY(mutex_); bool current_actor_should_exit_ ABSL_GUARDED_BY(mutex_) = false; bool is_detached_actor_ ABSL_GUARDED_BY(mutex_) = false; // The placement group id that the current actor belongs to. diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc index 13c9f0e9d2d8..d370a5b4f76e 100644 --- a/src/ray/core_worker/core_worker.cc +++ b/src/ray/core_worker/core_worker.cc @@ -2415,7 +2415,8 @@ Status CoreWorker::CreateActor(const RayFunction &function, extension_data, actor_creation_options.allow_out_of_order_execution, root_detached_actor_id, - actor_creation_options.actor_generator_backpressure_num_objects); + actor_creation_options.actor_generator_backpressure_num_objects, + actor_creation_options.restart_on_exception_patterns); // Add the actor handle before we submit the actor creation task, since the // actor handle must be in scope by the time the GCS sends the // WaitForActorRefDeletedRequest. diff --git a/src/ray/protobuf/common.proto b/src/ray/protobuf/common.proto index c4b8d304cd49..65cb48ffaa8d 100644 --- a/src/ray/protobuf/common.proto +++ b/src/ray/protobuf/common.proto @@ -849,6 +849,12 @@ message ActorCreationTaskSpec { // backpressure remains purely per-task via // TaskSpec.generator_backpressure_num_objects. int64 actor_generator_backpressure_num_objects = 15; + // Exception patterns that, when matched against an exception raised by an actor + // method or the actor's __init__, cause the actor worker to exit with a system + // error so the actor is restarted (consuming max_actor_restarts). Matching is a + // case-insensitive substring check against the formatted exception text. An empty + // list disables the feature. + repeated string restart_on_exception_patterns = 16; } // Task spec of an actor task.