[Core] Support restart-based retries for actor tasks

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.

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))),

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.

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 = "<exception str() failed>"
+    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,

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:

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":

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:

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",