Fix call_soon_threadsafe() racing with close() on free-threaded Python

tests/test_base.py

@@ -773,6 +773,59 @@ def scheduler():
         thread.join()
         self.assertEqual(counter[0], ITERATIONS)
 
+    def test_call_soon_threadsafe_while_closing(self):
+        # Other threads post to loops that close at the same time.  On
+        # free-threaded Python this used to crash in call_soon_threadsafe()
+        # (handler_async was already None) or make close() fail.
+        OWNERS = 2
+        POSTERS = 4
+        stop = threading.Event()
+        loops = [None] * OWNERS
+        errors = []
+
+        def owner(i):
+            try:
+                while not stop.is_set():
+                    loop = self.new_loop()
+                    loops[i] = loop
+                    try:
+                        loop.call_later(0.0005, loop.stop)
+                        loop.run_forever()
+                    finally:
+                        loop.close()
+            except BaseException as ex:
+                errors.append(ex)
+
+        def poster(i):
+            n = 0
+            try:
+                while not stop.is_set():
+                    loop = loops[n % OWNERS]
+                    n += 1
+                    if loop is None:
+                        continue
+                    try:
+                        loop.call_soon_threadsafe(lambda: None)
+                    except RuntimeError as ex:
+                        if str(ex) != 'Event loop is closed':
+                            raise
+            except BaseException as ex:
+                errors.append(ex)
+
+        threads = [threading.Thread(target=owner, args=(i,), daemon=True)
+                   for i in range(OWNERS)]
+        threads += [threading.Thread(target=poster, args=(i,), daemon=True)
+                    for i in range(POSTERS)]
+        for t in threads:
+            t.start()
+        time.sleep(1)
+        stop.set()
+        for t in threads:
+            t.join(10)
+        self.assertFalse([t for t in threads if t.is_alive()],
+                         'threads hang, likely a deadlock')
+        self.assertEqual(errors, [])
+
     def test_freethreading(self):
         if not hasattr(sys, "_is_gil_enabled"):
             raise unittest.SkipTest("No sys._is_gil_enabled()")

uvloop/loop.pxd

@@ -1,6 +1,8 @@
 # cython: language_level=3
 
 
+cimport cython
+
 from .includes cimport uv
 from .includes cimport system
 
@@ -39,6 +41,9 @@ cdef class Loop:
         public slow_callback_duration
 
         readonly bint _closed
+        # Held by call_soon_threadsafe() and by _close() while it sets
+        # _closed, so another thread never posts into a closing loop.
+        cython.pymutex _close_lock
         bint _debug
         bint _running
         bint _stopping

uvloop/loop.pyx

@@ -555,10 +555,15 @@ cdef class Loop:
         if self._running == 1:
             raise RuntimeError("Cannot close a running event loop")
 
-        if self._closed == 1:
-            return
-
-        self._closed = 1
+        # call_soon_threadsafe() can run in another thread.  It holds
+        # _close_lock while it checks _closed, appends to _ready and wakes
+        # the loop.  Once we set _closed under the lock, no other thread
+        # touches _ready or handler_async again.  A critical section is not
+        # enough: the deque append can block, which suspends the section.
+        with self._close_lock:
+            if self._closed == 1:
+                return
+            self._closed = 1
 
         for cb_handle in self._ready:
             cb_handle.cancel()
@@ -1278,11 +1283,23 @@ cdef class Loop:
         if not args:
             args = None
         cdef Handle handle = new_Handle(self, callback, args, context)
-        self._append_ready_handle(handle)  # deque append is atomic
-        # libuv async handler is thread-safe while the idle handler is not -
-        # we only set the async handler here, which will start the idle handler
-        # in _on_wake() from the loop and eventually call the callback.
-        self.handler_async.send()
+        cdef bint closed
+        # Another thread can close the loop at the same time.  _close() takes
+        # the same lock, so it cannot close handler_async or touch _ready
+        # between the _closed check and send().  The lock is not reentrant:
+        # run no code under it that can allocate Python objects and so start
+        # a GC, whose finalizers may call us again.
+        with self._close_lock:
+            closed = self._closed
+            if not closed:
+                self._ready.append(handle)  # deque append is atomic
+                # libuv async handler is thread-safe while the idle handler
+                # is not - we only set the async handler here, which will
+                # start the idle handler in _on_wake() from the loop and
+                # eventually call the callback.
+                self.handler_async.send()
+        if closed:
+            self._check_closed()  # raises: _closed never goes back to 0
         return handle
 
     def call_later(self, delay, callback, *args, context=None):