diff --git a/CHANGELOG.txt b/CHANGELOG.txt index ee4c69c0..0ee4ea92 100644 --- a/CHANGELOG.txt +++ b/CHANGELOG.txt @@ -1,3 +1,10 @@ +UNRELEASED +---------- + +* Fixed a deadlock of the entire event loop when exiting + ``ThreadSensitiveContext`` while its executor thread was still blocked + waiting on the event loop. (#535) + 3.11.1 (2026-02-03) ------------------- diff --git a/asgiref/sync.py b/asgiref/sync.py index e733c8e5..4c9f532e 100644 --- a/asgiref/sync.py +++ b/asgiref/sync.py @@ -8,7 +8,7 @@ import threading import warnings import weakref -from concurrent.futures import Future, ThreadPoolExecutor +from concurrent.futures import Future, InvalidStateError, ThreadPoolExecutor from typing import ( TYPE_CHECKING, Any, @@ -144,9 +144,25 @@ async def __aexit__(self, exc, value, tb): return executor = SyncToAsync.context_to_thread_executor.pop(self, None) - if executor: - executor.shutdown() SyncToAsync.thread_sensitive_context.reset(self.token) + if executor: + # The executor's worker thread may itself be waiting for this + # event loop, so a blocking shutdown() here would deadlock it. + # Join in a dedicated thread, not the loop's default executor: + # work queued there may itself be needed to unpark the worker, + # and joins occupying its slots would starve it. + future: "Future[None]" = Future() + + def join() -> None: + executor.shutdown() + try: + future.set_result(None) + except InvalidStateError: + # The await below was cancelled while we were joining. + pass + + threading.Thread(target=join, daemon=True).start() + await asyncio.wrap_future(future) class AsyncToSync(Generic[_P, _R]): diff --git a/tests/test_sync.py b/tests/test_sync.py index 4199ce65..8021868d 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -690,6 +690,134 @@ async def test_thread_sensitive_context_without_sync_work(): pass +def cancel_inside_thread_sensitive_context(): + """Cancels a thread-sensitive task parked in async_to_sync, then exits the context""" + worker_started = threading.Event() + + async def inner(): + await asyncio.sleep(0.2) + + def sync_code(): + worker_started.set() + async_to_sync(inner)() + + async def main(): + async with ThreadSensitiveContext(): + task = asyncio.create_task( + sync_to_async(sync_code, thread_sensitive=True)() + ) + # Let the executor pick up sync_code, then hold the event loop + # with sync sleeps so the call_soon_threadsafe callback enqueued + # by async_to_sync is still queued when the cancellation lands. + await asyncio.sleep(0) + assert worker_started.wait(5) + time.sleep(0.1) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + asyncio.run(main()) + + +def test_thread_sensitive_context_exit_does_not_block_event_loop(): + """ + Tests that exiting ThreadSensitiveContext does not deadlock the event + loop if its executor thread is still parked in AsyncToSync, as happens + when the task running the sync code is cancelled before the event loop + runs the create_task callback enqueued by async_to_sync. (#535) + + Runs in a separate process as the parked executor thread would otherwise + hang the test suite at interpreter exit. + """ + process = multiprocessing.Process(target=cancel_inside_thread_sensitive_context) + process.start() + process.join(30) + # Force cleanup in failed test case + if process.is_alive(): + process.terminate() + process.join(5) + pytest.fail("event loop deadlocked exiting ThreadSensitiveContext") + assert process.exitcode == 0 + + +def starve_default_executor_exiting_thread_sensitive_context(): + """ + Wedges two shutdown joins and the work they depend on into a two-slot + default executor. + + Each request's sync code parks in async_to_sync, and the async side needs + a default-executor slot only after a sleep - by which time both requests + have been cancelled and their context exits occupy both slots. + """ + + async def inner(): + # Suspend first, so the context exits (and the shutdown joins claim + # their executor slots) before this needs a slot of its own. + await asyncio.sleep(0.3) + await sync_to_async(time.sleep, thread_sensitive=False)(0.05) + + def make_sync_code(worker_started): + def sync_code(): + worker_started.set() + async_to_sync(inner)() + + return sync_code + + async def request(worker_started): + async with ThreadSensitiveContext(): + task = asyncio.create_task( + sync_to_async(make_sync_code(worker_started), thread_sensitive=True)() + ) + # Let the executor pick up sync_code, then hold the event loop + # with sync sleeps so the call_soon_threadsafe callback enqueued + # by async_to_sync is still queued when the cancellation lands. + await asyncio.sleep(0) + assert worker_started.wait(5) + time.sleep(0.1) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + async def main(): + # Two slots stand in for the real default executor's min(32, cpus + 4) + # so two requests are enough to fill it. + asyncio.get_running_loop().set_default_executor( + ThreadPoolExecutor(max_workers=2) + ) + await asyncio.gather(request(threading.Event()), request(threading.Event())) + + asyncio.run(main()) + + +def test_thread_sensitive_context_exit_does_not_starve_default_executor(): + """ + Tests that exiting ThreadSensitiveContext cannot starve the event loop's + default executor. If the shutdown join runs on the default executor, + concurrent context exits can fill it while their parked worker threads + are waiting on work that is queued behind those joins in the same + executor - a circular wait that wedges the executor permanently even + though the event loop stays responsive. (#535) + + Runs in a separate process as the wedged threads would otherwise hang + the test suite at interpreter exit. + """ + process = multiprocessing.Process( + target=starve_default_executor_exiting_thread_sensitive_context + ) + process.start() + process.join(30) + # Force cleanup in failed test case + if process.is_alive(): + process.terminate() + process.join(5) + pytest.fail("default executor starved exiting ThreadSensitiveContext") + assert process.exitcode == 0 + + def test_thread_sensitive_double_nested_sync(): """ Tests that thread_sensitive SyncToAsync nests inside itself where the