open_loop() doesn't close properly when cancelled from trio

[vcalvert]

Using trio-asyncio 0.11.0 and trio 0.16.0 (same behavior on trio 0.15.1 also).

Sample code:

import asyncio
import trio
import trio_asyncio
import logging

logger = logging.getLogger(__name__)

@trio_asyncio.aio_as_trio
async def sleeper(timeout):
    try:
        logger.info("Beginning sleep for %d", timeout)
        await asyncio.sleep(timeout)
        logger.info("Finished sleep (%d)", timeout)
    except asyncio.CancelledError:
        logger.info("Cancelled sleep (originally for %d)", timeout)
        raise
    return timeout


async def trio_func(timeout1, timeout2, *args):
    try:
        logger.info("Starting fail_after(%d)", timeout1)
        with trio.fail_after(timeout1):
            logger.info("Opening trio_asyncio loop")
            async with trio_asyncio.open_loop() as loop:
                logger.info("Beginning asyncio_sleep(%d)", timeout2)
                foo = await sleeper(timeout2)
                logger.info("Finished asyncio sleep")
                logger.info("Received value %d", foo)
            logger.info("Closed trio_asyncio loop")
        logger.info("Finished happy path")
    except trio.TooSlowError:
        logger.info("First timeout triggered (%d)", timeout1)
    except Exception as exc:
        logger.error("Failure in trio_asyncio call (%d, %d)", timeout1, timeout2, exc_info=exc)
    logger.info("Finished processing")


async def multiple_runs():
    timeout_sequences: List[List[int]] = []
    timeout_sequences.append([2,1])
    timeout_sequences.append([1,2])
    for seq in timeout_sequences:
        logger.info("Starting sequence for %s", str(seq))
        await trio_func(*seq)
        logger.info("Finished sequence for %s", str(seq))


if __name__=="__main__":
    logging.basicConfig(level="INFO", format="{asctime} - {funcName}: {lineno}: {message}", style='{')
    logger.info("Starting trio mainloop")
    trio.run(multiple_runs)
    logger.info("Finished trio mainloop")

The output I get:

<snip>/python3.7/site-packages/trio_asyncio/_base.py:20: TrioDeprecationWarning: trio.hazmat is deprecated since Trio 0.15.0; use trio.lowlevel instead (https://github.com/python-trio/trio/issues/476)
  from trio.hazmat import wait_for_child
2020-08-21 01:16:46,401 - <module>: 51: Starting trio mainloop
2020-08-21 01:16:46,401 - multiple_runs: 44: Starting sequence for [2, 1]
2020-08-21 01:16:46,402 - trio_func: 22: Starting fail_after(2)
2020-08-21 01:16:46,402 - trio_func: 24: Opening trio_asyncio loop
2020-08-21 01:16:46,402 - trio_func: 26: Beginning asyncio_sleep(1)
2020-08-21 01:16:46,402 - sleeper: 11: Beginning sleep for 1
2020-08-21 01:16:47,404 - sleeper: 13: Finished sleep (1)
2020-08-21 01:16:47,405 - trio_func: 28: Finished asyncio sleep
2020-08-21 01:16:47,405 - trio_func: 29: Received value 1
2020-08-21 01:16:47,407 - trio_func: 30: Closed trio_asyncio loop
2020-08-21 01:16:47,407 - trio_func: 31: Finished happy path
2020-08-21 01:16:47,407 - trio_func: 36: Finished processing
2020-08-21 01:16:47,407 - multiple_runs: 46: Finished sequence for [2, 1]
2020-08-21 01:16:47,407 - multiple_runs: 44: Starting sequence for [1, 2]
2020-08-21 01:16:47,407 - trio_func: 22: Starting fail_after(1)
2020-08-21 01:16:47,407 - trio_func: 24: Opening trio_asyncio loop
2020-08-21 01:16:47,408 - trio_func: 26: Beginning asyncio_sleep(2)
2020-08-21 01:16:47,409 - sleeper: 11: Beginning sleep for 2
2020-08-21 01:16:48,410 - sleeper: 15: Cancelled sleep (originally for 2)

At this point I have to kill the process; it doesn't clean up the asyncio loop as expected on cancellation of the coroutine for the second run. If it completes within the timeout, everything is ok, and exceptions propagate as expected.

If a timeout occurs, however, while the trio.Cancelled exception is propagated (and translated to asyncio.CancelledError as expected), then it doesn't matter what exception is raised (including a re-raise of the existing one), everything hangs prior to returning to the invoking trio function.

The expectation: if the last task run in a trio_asyncio.open_loop() loop is cancelled, it should return to the main process and continue from there with the appropriate exception.
The output for the [1, 2] sequence should look something like:

2020-08-21 01:16:47,407 - multiple_runs: 44: Starting sequence for [1, 2]
2020-08-21 01:16:47,407 - trio_func: 22: Starting fail_after(1)
2020-08-21 01:16:47,407 - trio_func: 24: Opening trio_asyncio loop
2020-08-21 01:16:47,408 - trio_func: 26: Beginning asyncio_sleep(2)
2020-08-21 01:16:47,409 - sleeper: 11: Beginning sleep for 2
2020-08-21 01:16:48,410 - sleeper: 15: Cancelled sleep (originally for 2)
2020-08-21 01:16:48,412 - trio_func: 33: First timeout triggered (1).
2020-08-21 01:16:48,413 - trio_func: 36: Finished processing
2020-08-21 01:16:48,415 - multiple_runs: 46: Finished sequence for [1, 2]
2020-08-21 01:16:48,415 - <module>: 53: Finished trio mainloop

Using await foo = loop.run_aio_coroutine(sleeper(timeout2)) instead of the @aio_as_trio decorator has identical behavior to the example above.

It looks like #88 may also be running into a similar issue.

Perhaps this is caused by the unusual exception base class causing a problem cleaning up the loop?

I have not tested whether async_timeout would work within the asyncio function, but it's less useful for my use case (and pulls in another dependency).