[core] handle unserializable user exception

doc/source/ray-core/doc_code/task_exceptions.py

@@ -35,6 +35,32 @@ def g(x):
     # Exception: the real error
 
 # __task_exceptions_end__
+# __unserializable_exceptions_begin__
+
+import threading
+
+class UnserializableException(Exception):
+    def __init__(self):
+        self.lock = threading.Lock()
+
+@ray.remote
+def raise_unserializable_error():
+    raise UnserializableException
+
+try:
+    ray.get(raise_unserializable_error.remote())
+except ray.exceptions.RayTaskError as e:
+    print(e)
+    # ray::raise_unserializable_error() (pid=222813, ip=172.31.5.154)
+    #   File "/home/ubuntu/ray/tmp~/main.py", line 25, in raise_unserializable_error
+    #     raise UnserializableException
+    # UnserializableException
+    print(type(e.cause))
+    # <class 'ray.exceptions.RayError'>
+    print(e.cause)
+    # The original cause of RayTaskError (<class '__main__.UnserializableException'>) is not serializable: cannot pickle '_thread.lock' object. Overwriting the cause to RayError.
+
+# __unserializable_exceptions_end__
 # __catch_user_exceptions_begin__
 
 class MyException(Exception):

doc/source/ray-core/fault_tolerance/tasks.rst

@@ -27,6 +27,13 @@ field of the ``RayTaskError``.
   :start-after: __task_exceptions_begin__
   :end-before: __task_exceptions_end__
 
+If the user's exception cannot be serialized, it would be converted to a ``RayError``:
+
+.. literalinclude:: ../doc_code/task_exceptions.py
+  :language: python
+  :start-after: __unserializable_exceptions_begin__
+  :end-before: __unserializable_exceptions_end__
+
 Example code of catching the user exception type when the exception type can be subclassed:
 
 .. literalinclude:: ../doc_code/task_exceptions.py

python/ray/exceptions.py

@@ -130,8 +130,20 @@ def __init__(
         self.traceback_str = traceback_str
         self.actor_repr = actor_repr
         self._actor_id = actor_id
-        # TODO(edoakes): should we handle non-serializable exception objects?
         self.cause = cause
+
+        try:
+            pickle.dumps(cause)
+        except (pickle.PicklingError, TypeError) as e:
+            err_msg = (
+                "The original cause of RayTaskError"
+                f" ({self.cause.__class__}) is not serializable: {e}."
+                " Overwriting the cause to RayError."
+            )
+            logger.warning(err_msg)
+            self.cause = RayError(err_msg)
+            self.args = (function_name, traceback_str, self.cause, proctitle, pid, ip)
+
         assert traceback_str is not None
 
     def make_dual_exception_type(self) -> Type:

python/ray/tests/test_failure.py

@@ -3,6 +3,7 @@
 import sys
 import time
 import logging
+import threading
 
 import numpy as np
 import pytest
@@ -600,6 +601,25 @@ def check_actor_restart():
     ray.get(obj2)
 
 
+# Previously when there is threading.Lock in the exception, it will cause
+# the serialization to fail. This test case is to cover this scenario.
+def test_unserializable_exception(ray_start_regular, propagate_logs):
+    class UnserializableException(Exception):
+        def __init__(self):
+            self.lock = threading.Lock()
+
+    @ray.remote
+    def func():
+        raise UnserializableException
+
+    with pytest.raises(ray.exceptions.RayTaskError) as exc_info:
+        ray.get(func.remote())
+
+    assert isinstance(exc_info.value, ray.exceptions.RayTaskError)
+    assert isinstance(exc_info.value.cause, ray.exceptions.RayError)
+    assert "is not serializable" in str(exc_info.value.cause)
+
+
 def test_final_user_exception(ray_start_regular, propagate_logs, caplog):
     class MyFinalException(Exception):
         def __init_subclass__(cls, /, *args, **kwargs):