`test_raise_baseexception_from_task` occasionally deadlocks

[gschaffner]

Things to check first

• I have searched the existing issues and didn't find my bug already reported there

• I have checked that my bug is still present in the latest release

AnyIO version

master (14356be)

Python version

CPython 3.13.7

What happened?

test_raise_baseexception_from_task occasionally deadlocks. a71de90

How can we reproduce the bug?

#1012 (comment)

[gschaffner]

Analysis: It appears to be asyncio-only. The test thread hangs in

tests/test_from_thread.py:710: in test_raise_baseexception_from_task
    with start_blocking_portal(anyio_backend_name, anyio_backend_options) as portal:
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../../.pyenv/versions/3.13/lib/python3.13/contextlib.py:148: in __exit__
    next(self.gen)
src/anyio/from_thread.py:544: in start_blocking_portal
    portal.call(portal.stop, cancel_remaining_tasks)
src/anyio/from_thread.py:326: in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
src/anyio/from_thread.py:370: in start_task_soon
    self._spawn_task_from_thread(func, args, {}, name, f)
src/anyio/_backends/_asyncio.py:1024: in _spawn_task_from_thread
    AsyncIOBackend.run_sync_from_thread(
src/anyio/_backends/_asyncio.py:2587: in run_sync_from_thread
    return f.result()

I believe that the race condition is due to the difference in semantics between TrioToken.run_sync_soon and asyncio.AbstractEventLoop.call_soon_threadsafe. If run_sync_soon returns without raising, it guarantees that the callback will be executed:

Raises: trio.RunFinishedError – if the associated call to trio.run() has already exited. (Any call that doesn’t raise this error is guaranteed to be fully processed before trio.run() exits.)

call_soon_threadsafe does not have this guarantee.

The deadlock occurs when: The portal's event loop is going to shut down very soon. (In the case of the test, this is because of a BaseException that is bubbling up in the portal's event loop.) Then start_blocking_portal.__exit__ calls portal.call(portal.stop) calls AsyncIOBackend.run_sync_from_thread(task_group.start_soon, ...). In run_sync_from_thread:

anyio/src/anyio/_backends/_asyncio.py

Lines 2579 to 2587 in 14356be

	loop = cast(
	"AbstractEventLoop", token or threadlocals.current_token.native_token
	)
	if loop.is_closed():
	raise RunFinishedError
	f: concurrent.futures.Future[T_Retval] = Future()
	loop.call_soon_threadsafe(wrapper)
	return f.result()

call_soon_threadsafe(wrapper) returns without raising RuntimeError because the event loop is not yet closed. Then the event loop closes without ever running wrapper. This causes portal.call(portal.stop) to wait forever for wrapper to run.

start_blocking_portal.__exit__ does make an effort to handle the case where the event loop is already closed by catching the RuntimeError that portal._check_running or call_soon_threadsafe raises:

anyio/src/anyio/from_thread.py

Lines 547 to 550 in 14356be

	try:
	portal.call(portal.stop, cancel_remaining_tasks)
	except RuntimeError:
	pass

(Aside: Shouldn't it be catching RunFinishedError too?)

However, unlike TrioToken.run_sync_soon, call_soon_threadsafe doesn't guarantee that the callback will actually execute, so there's still a race.

TL;DR: #472 did not fully fix #471. It usually fixed it, but not always.

[gschaffner]

I'm not sure whether or not it's possible to implement run_sync_soon semantics on asyncio. I tried to implement RunToken.run_sync_soon on asyncio while working on #256 in Nov. 2023. Here's what I had tried to do in order to get the "if run_sync_soon returns, the callback is guaranteed to run" guarantee:

    def run_sync_soon(...):
        # `asyncio.call_soon_threadsafe` raises `RuntimeError` if the loop is already
        # closed, but it doesn't raise if the loop is about to close (i.e. the loop is
        # not going to run any new callbacks). In either case, it's too late for a
        # `run_sync_soon` call.
        if find_root_task() is None:
            raise RuntimeError("run() has exited/is about to exit")

        call_soon_threadsafe(...)

But this code still has a race, because there is a TOCTOU bug: the state of the root task can change between if and call_soon_threadsafe, so the event loop can still shut down after call_soon_threadsafe returns success but before the event loop runs the callback.

[gschaffner]

Okay, I think that having run_sync_soon semantics on asyncio requires doing actual locking. Hypothetically, what if we ran a background task that we use to synchronize a flag? Roughly:

async def _notify_not_accepting_further_callbacks() -> Never:
    try:
        await sleep_forever()
    finally:  # CancelledError
        with not_accepting_further_callbacks_lock:
            not_accepting_further_callbacks = True
        # Execute the callbacks that were scheduled before we set the flag:
        await sleep(0)


create_task(_notify_not_accepting_further_callbacks)


def similar_to_run_sync_soon(...):
    with not_accepting_further_callbacks_lock:  # this lock could be a RW lock for improved perf
        if not_accepting_further_callbacks:
            raise RunFinishedError
        call_soon_threadsafe(func)  # func is now _guaranteed_ to execute before loop shutdown.

However, like _shutdown_process_pool_on_exit, this background task approach wouldn't work for asyncio loops that are being driven manually (i.e. loops using the old pre-run APIs).

[gschaffner]

Another potential approach to fix this is: have AsyncIOBackend.run_sync_from_thread do

def root_task_cb():
    f.cancel()
    f.set_running_or_notify_cancel()

find_root_task().add_done_callback(root_task_cb)

This also doesn't work for loops using the old pre-run APIs.

Edit: Nevermind, asyncio.Task.add_done_callback isn't thread-safe.

[agronholm]

Would it help if I did this:

 f: concurrent.futures.Future[T_Retval] = Future() 
 loop.call_soon_threadsafe(wrapper) 
 if loop.is_closed(): 
     raise RunFinishedError 
  
 return f.result() 

[agronholm]

To answer my own question, it doesn't stop the hangs, but it does cause additional test failures.

[agronholm]

Yet now that I try, I cannot reproduce any of the problems. 🤔