Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.txt
Original file line number Diff line number Diff line change
@@ -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)
-------------------

Expand Down
22 changes: 19 additions & 3 deletions asgiref/sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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]):
Expand Down
128 changes: 128 additions & 0 deletions tests/test_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -690,6 +690,134 @@ async def test_thread_sensitive_context_without_sync_work():
pass


def cancel_inside_thread_sensitive_context():

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice, the new unit test makes the error obvious 👍

"""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
Expand Down
Loading