RuntimeError when cancelling asyncio.Task running within a BlockingPortal.

[jonathanslenders]

Hi, I'm facing some issues with cancellation and blocking portals and wondering whether that's a bug in anyio, or in my code.

To reproduce:

• Create a blocking portal.
• Schedule a coroutine in the blocking portal.
• This coroutine creates an asyncio.Task and waits for it.
• This asyncio.Task gets cancelled from elsewhere, within the event loop. So, we don't cancel the concurrent.futures.Future that gets returned by the BlockingPortal.start_task_soon.
• A RuntimeError gets logged.

I don't have a code snippet to reproduce, but if required, I can probably create it. To me, the RuntimeError doesn't make sense.

Traceback (most recent call last):
  File "/home/jonathan/.pyenv/versions/3.9.5/lib/python3.9/concurrent/futures/_base.py", line 329, in _invoke_callbacks
    callback(self)
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 166, in callback
    self.call(scope.cancel)
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 230, in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 291, in start_task_soon
    self._check_running()
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 140, in _check_running
    raise RuntimeError('This method cannot be called from the event loop thread')
RuntimeError: This method cannot be called from the event loop thread

[agronholm]

Yes, please create a minimal reproducing case. This traceback doesn't show where the call really originated from.

[jonathanslenders]

This should do it:

import asyncio
from anyio.from_thread import BlockingPortal


async def main():
    async with BlockingPortal() as portal:
        await asyncio.get_running_loop().run_in_executor(
            None, lambda: in_thread(portal)
        )


def in_thread(portal):
    async def coro():
        pass

    async def in_loop():
        t = asyncio.create_task(coro())
        t.cancel()
        await t

    portal.start_task_soon(in_loop)


asyncio.run(main())

Output:

 $ python /tmp/test.py
exception calling callback for <Future at 0x7f9ebd9b4280 state=cancelled>
Traceback (most recent call last):
  File "/tmp/test.py", line 13, in coro
    async def coro():
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 177, in _call_func
    retval = await retval
  File "/tmp/test.py", line 19, in in_loop
    await t
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/jonathan/.pyenv/versions/3.9.5/lib/python3.9/concurrent/futures/_base.py", line 329, in _invoke_callbacks
    callback(self)
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 166, in callback
    self.call(scope.cancel)
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 230, in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 291, in start_task_soon
    self._check_running()
  File "/home/jonathan/git/anyio/src/anyio/from_thread.py", line 140, in _check_running
    raise RuntimeError('This method cannot be called from the event loop thread')
RuntimeError: This method cannot be called from the event loop thread

[agronholm]

Thanks, I'll start investigating this in a while.

[jonathanslenders]

Thanks @agronholm, I appreciate it!

[agronholm]

In the AnyIO test suite, all the calls to https://github.com/agronholm/anyio/blob/master/src/anyio/from_thread.py#L164 happen in the worker thread. But in your script the call happens in the event loop thread. I'm now trying to figure out why.

[agronholm]

I have a fix for this, but I need to create a regression test too before I commit it.

[agronholm]

I was able to reduce the test case into a smaller snippet:

import anyio
from anyio.from_thread import BlockingPortal
from anyio import to_thread


async def main():
    async with BlockingPortal() as portal:
        await to_thread.run_sync(in_thread, portal)


def in_thread(portal):
    async def in_loop():
        raise anyio.get_cancelled_exc_class()

    portal.start_task_soon(in_loop)


anyio.run(main)

But when I copy the same code into a test function, the test passes. Quite infuriating.

[agronholm]

The RuntimeError is in fact being raised, but gets lost somewhere along the way. Further investigation is needed.

[jonathanslenders]

Thanks for looking into it!

[agronholm]

I managed to produce a failing test case. Seems like this is only applicable to asyncio. There seems to be no way for a cancellation exception to come through unless the portal was already stopped.

[jonathanslenders]

Thanks a lot for fixing it that quick!
I'm looking forward to the next release! :-)