Handle stopped main event loops in async_to_sync

[davitacols]

Summary

• only reuse the threadlocal main_event_loop when it is still running
• fall back to a new loop if that threadlocal loop has stopped but not closed
• add a regression test covering the stopped-but-open loop case

A stopped loop can still accept call_soon_threadsafe() without ever running the scheduled task, which leaves async_to_sync() blocked waiting for a result.

Fixes #545.

Testing

• python -m pytest -q
• python -m mypy asgiref tests setup.py docs/conf.py

[carltongibson]

@andersk Can I get you to look at this example? — I think it's the same issue as #528 but with a better example.

I had similar experimenting:

def test_async_to_sync_with_stopped_main_loop():
    """
    A wrapper built while a loop is running captures that loop. If the loop is
    later stopped but not closed (e.g. a shared pytest-asyncio loop between
    tests), calling the wrapper from a sync thread must fall back to a new loop
    rather than deadlock.
    """
    holder = {}

    async def inner():
        return 42

    async def build():
        holder["fn"] = async_to_sync(inner)

    loop = asyncio.new_event_loop()
    try:
        loop.run_until_complete(build())
        assert not loop.is_running() and not loop.is_closed()

        result = {}

        def worker():
            result["value"] = holder["fn"]()

        thread = threading.Thread(target=worker, daemon=True)
        thread.start()
        thread.join(timeout=5)
        assert not thread.is_alive(), "async_to_sync deadlocked on a stopped loop"
        assert result["value"] == 42
    finally:
        loop.close()

It looks like the claim in #507 that the is_running() check was redundant was not right.

In particular, pytest-asyncio reuses the same loop between tests with multiple loop.run_until_complete() invocations. (That seems legitimate.)

WDYT? Thanks!

[andersk]

I wonder if the real problem there is that that we capture the loop in AsyncToSync.__init__ at all? It’s unintuitive that merely defining a function in a different location should affect which loop it uses when called.

If we remove that capture, your example passes, and the only failure in the test suite is the final assert in test_async_to_sync_to_thread_decorator:

asgiref/tests/test_sync.py

Lines 278 to 279 in 4d233dd

	# Make sure that it didn't needlessly make a new async loop
	assert result["thread"] == threading.current_thread()

Indeed, this is the only place in the test suite where a captured loop makes it through the top of AsyncToSync.__call__ without being replaced.

Perhaps this test’s expectation is wrong? asyncio.to_thread intentionally doesn’t copy the loop to the new thread, because a loop object is not thread-safe. Granted, we only use the thread-safe subset of the loop interface. But does the user really expect that? If the user wanted to go back to the old thread, why did they write asyncio.to_thread instead of sync_to_async?

[andersk]

I drafted #562 for that proposal. I’d be curious if it also prevents the pytest-asyncio deadlocks that people say they’re seeing (I mean I don’t think they’re making it up, but I haven’t seen a real reproduction posted yet).