[coe] Remove gil when submit actor and tasks to avoid deadlock for some cases.

python/ray/_raylet.pyx

@@ -1532,6 +1532,7 @@ cdef class CoreWorker:
         cdef:
             unordered_map[c_string, double] c_resources
             CRayFunction ray_function
+            CTaskOptions task_options
             c_vector[unique_ptr[CTaskArg]] args_vector
             c_vector[CObjectReference] return_refs
             CSchedulingStrategy c_scheduling_strategy
@@ -1548,17 +1549,17 @@ cdef class CoreWorker:
                 self, language, args, &args_vector, function_descriptor,
                 &incremented_put_arg_ids)
 
-            # NOTE(edoakes): releasing the GIL while calling this method causes
-            # segfaults. See relevant issue for details:
-            # https://github.com/ray-project/ray/pull/12803
-            return_refs = CCoreWorkerProcess.GetCoreWorker().SubmitTask(
-                ray_function, args_vector, CTaskOptions(
-                    name, num_returns, c_resources,
-                    b"",
-                    serialized_runtime_env_info),
-                max_retries, retry_exceptions,
-                c_scheduling_strategy,
-                debugger_breakpoint)
+            task_options = CTaskOptions(
+                name, num_returns, c_resources,
+                b"",
+                serialized_runtime_env_info)
+            with nogil:
+                return_refs = CCoreWorkerProcess.GetCoreWorker().SubmitTask(
+                    ray_function, args_vector, task_options,
+                    max_retries, retry_exceptions,
+                    c_scheduling_strategy,
+                    debugger_breakpoint,
+                )
 
             # These arguments were serialized and put into the local object
             # store during task submission. The backend increments their local
@@ -1742,15 +1743,13 @@ cdef class CoreWorker:
                 self, language, args, &args_vector, function_descriptor,
                 &incremented_put_arg_ids)
 
-            # NOTE(edoakes): releasing the GIL while calling this method causes
-            # segfaults. See relevant issue for details:
-            # https://github.com/ray-project/ray/pull/12803
-            return_refs = CCoreWorkerProcess.GetCoreWorker().SubmitActorTask(
-                c_actor_id,
-                ray_function,
-                args_vector,
-                CTaskOptions(
-                    name, num_returns, c_resources, concurrency_group_name))
+            with nogil:
+                return_refs = CCoreWorkerProcess.GetCoreWorker().SubmitActorTask(
+                    c_actor_id,
+                    ray_function,
+                    args_vector,
+                    CTaskOptions(
+                        name, num_returns, c_resources, concurrency_group_name))
             # These arguments were serialized and put into the local object
             # store during task submission. The backend increments their local
             # ref count initially to ensure that they remain in scope until we

python/ray/tests/test_failure_3.py

@@ -1,9 +1,9 @@
 import os
 import sys
 import signal
+import threading
 
 import ray
-
 import numpy as np
 import pytest
 import time
@@ -141,6 +141,36 @@ async def get(self, x, wait=False):
     assert ray.get(ref_3) == 3
 
 
+def test_actor_failure_async(ray_start_regular):
+    @ray.remote
+    class A:
+        def echo(self):
+            pass
+
+        def pid(self):
+            return os.getpid()
+
+    a = A.remote()
+    rs = []
+
+    def submit():
+        for i in range(100000):
+            r = a.echo.remote()
+            r._on_completed(lambda x: 1)
+            rs.append(r)
+
+    t = threading.Thread(target=submit)
+    pid = ray.get(a.pid.remote())
+
+    t.start()
+    from time import sleep
+
+    sleep(0.1)
+    os.kill(pid, signal.SIGKILL)
+
+    t.join()
+
+
 if __name__ == "__main__":
     import pytest