support IOLoop.close() when called after asyncio.run

Open
#3,292 1 comment 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Assessment

Difficulty
4/5
Estimated time
3-5 days
Newbie friendliness
35/100
Issue type
Bug
Clarity
Mostly clear
Activity status
Stale
Tech stack
python
Domain
networking

Research direction

Start with the test_inproc_specific_different_threads failure in distributed/comm/tests/test_comms.py and the IOLoop.close() traceback. Inspect the cleanup of IOLoop._ioloop_for_asyncio when AbstractEventLoop.close() has already run; done means the reported close sequence completes without a KeyError.

Written by the indexing model from the issue text.

Description

ioloop

I've been using the following function to run the distributed test suite:

def _run_and_close_tornado(async_fn, /, *args, **kwargs):
    tornado_loop = None

    async def inner_fn():
        nonlocal tornado_loop
        tornado_loop = IOLoop.current()
        return await async_fn(*args, **kwargs)

    try:
        return asyncio.run(inner_fn())
    finally:
        tornado_loop.close(all_fds=True)

Which has worked great with non-overlapping calls to IOLoop.current(), however I recently introduced an overlapping call, which results in a KeyError:

____________________ test_inproc_specific_different_threads ____________________

    @gen_test()
    async def test_inproc_specific_different_threads():
>       await check_inproc_specific(run_coro_in_thread)

distributed/comm/tests/test_comms.py:457: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
distributed/comm/tests/test_comms.py:431: in check_inproc_specific
    await asyncio.gather(*futures)
distributed/comm/tests/test_comms.py:447: in run_coro_in_thread
    return await asyncio.to_thread(run_and_close_tornado, run_with_timeout)
/usr/share/miniconda3/envs/dask-distributed/lib/python3.10/asyncio/threads.py:25: in to_thread
    return await loop.run_in_executor(None, func_call)
/usr/share/miniconda3/envs/dask-distributed/lib/python3.10/concurrent/futures/thread.py:58: in run
    result = self.fn(*self.args, **self.kwargs)
distributed/utils.py:506: in run_and_close_tornado
    tornado_loop.close(all_fds=True)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tornado.platform.asyncio.AsyncIOMainLoop object at 0x7f7a6addb280>
all_fds = True

    def close(self, all_fds: bool = False) -> None:
        self.closing = True
        for fd in list(self.handlers):
            fileobj, handler_func = self.handlers[fd]
            self.remove_handler(fd)
            if all_fds:
                self.close_fd(fileobj)
        # Remove the mapping before closing the asyncio loop. If this
        # happened in the other order, we could race against another
        # initialize() call which would see the closed asyncio loop,
        # assume it was closed from the asyncio side, and do this
        # cleanup for us, leading to a KeyError.
>       del IOLoop._ioloop_for_asyncio[self.asyncio_loop]
E       KeyError: <_UnixSelectorEventLoop running=False closed=True debug=False>

https://github.com/dask/distributed/actions/runs/5475797836/jobs/9972404903#step:19:2420

@bdarnell could you catch the KeyError so that I can run AbstractEventLoop.close() then IOLoop.close()?

Dominant language
Python
Stars
22.2k
Forks
5.6k
Avg merge
3h 42m
Merged PRs (30d)
16

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from tornadoweb/tornado

All issues in tornadoweb/tornado

Similar issues

More Python issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.