From 886b562be8dc87ced4e83e4a240e72bcf626c412 Mon Sep 17 00:00:00 2001 From: Maarten Breddels Date: Sun, 8 Nov 2026 12:00:04 +0100 Subject: [PATCH] Fix call_soon_threadsafe() racing with close() on free-threaded Python call_soon_threadsafe() checks _closed, appends to _ready and then calls handler_async.send(). On a free-threaded build another thread can run close() in between: it closes handler_async and sets it to None, so the poster segfaults in send(). The same race also makes close() fail with "deque mutated during iteration" or "new callbacks were queued during loop closing". With the GIL the method runs as one step, so this did not happen. A critical section on the loop is not enough: the deque append can block on the deque's own lock, and that suspends the outer critical section. A PyMutex stays held while the thread blocks. The mutex is not reentrant, so nothing that can start a GC runs under it: a finalizer could post to the same loop and deadlock. That is why the closed-loop error is raised after the lock is released. Co-Authored-By: Claude Opus 5.5 --- tests/test_base.py | 53 ++++++++++++++++++++++++++++++++++++++++++++++ uvloop/loop.pxd | 5 +++++ uvloop/loop.pyx | 35 ++++++++++++++++++++++-------- 3 files changed, 84 insertions(+), 9 deletions(-) diff --git a/tests/test_base.py b/tests/test_base.py index b226a84f..b7d453a1 100644 --- a/tests/test_base.py +++ b/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()") diff --git a/uvloop/loop.pxd b/uvloop/loop.pxd index 89566ea3..5b4945bc 100644 --- a/uvloop/loop.pxd +++ b/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 diff --git a/uvloop/loop.pyx b/uvloop/loop.pyx index 802cdb69..ff3efe35 100644 --- a/uvloop/loop.pyx +++ b/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):