[core] Acquire GIL before calling PyGILState_Release

python/ray/_raylet.pyx

@@ -72,6 +72,7 @@ from cython.operator import dereference, postincrement
 from cpython.pystate cimport (
     PyGILState_Ensure,
     PyGILState_Release,
+    PyGILState_STATE,
 )
 
 from ray.includes.common cimport (
@@ -2256,6 +2257,10 @@ cdef shared_ptr[LocalMemoryBuffer] ray_error_to_memory_buf(ray_error):
     return make_shared[LocalMemoryBuffer](
         <uint8_t*>py_bytes, len(py_bytes), True)
 
+cdef void pygilstate_release(PyGILState_STATE gstate) nogil:
+    with gil:
+        PyGILState_Release(gstate)
+
 cdef function[void()] initialize_pygilstate_for_thread() nogil:
     """
     This function initializes a C++ thread to make it be considered as a
@@ -2272,7 +2277,7 @@ cdef function[void()] initialize_pygilstate_for_thread() nogil:
     cdef function[void()] callback
     with gil:
         gstate = PyGILState_Ensure()
-        callback = bind(PyGILState_Release, ref(gstate))
+        callback = bind(pygilstate_release, ref(gstate))
     return callback
 
 cdef CRayStatus task_execution_handler(

python/ray/tests/test_actor.py

@@ -1046,6 +1046,7 @@ def get_value(self):
     )
 
 
+@pytest.mark.parametrize("enable_concurrency_group", [True, False])
 @pytest.mark.parametrize(
     "exit_condition",
     [
@@ -1055,13 +1056,17 @@ def get_value(self):
         "ray.kill",
     ],
 )
-def test_atexit_handler(ray_start_regular_shared, exit_condition):
-    @ray.remote
+def test_atexit_handler(
+    ray_start_regular_shared, exit_condition, enable_concurrency_group
+):
+    concurrency_groups = {"io": 1} if enable_concurrency_group else None
+
+    @ray.remote(concurrency_groups=concurrency_groups)
     class A:
         def __init__(self, tmpfile, data):
             import atexit
 
-            def f(*args, **kwargs):
+            def f():
                 with open(tmpfile, "w") as f:
                     f.write(data)
                     f.flush()