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):