Why does asyncio subprocess behave differently with created event loop unless you set_event_loop?

Question

I have this simple async code that spawns sleep 3 and then waits for it to complete:

from asyncio import SelectorEventLoop, create_subprocess_exec, \
    wait_for, get_event_loop, set_event_loop


def run_timeout(loop, awaitable, timeout):
    timed_awaitable = wait_for(awaitable, timeout=timeout, loop=loop)
    return loop.run_until_complete(timed_awaitable)

async def foo(loop):
    process = await create_subprocess_exec('sleep', '3', loop=loop)
    await process.wait()
    print(process.returncode)

Notice how it takes a custom loop. If I run it with the following:

loop = get_event_loop()
run_timeout(loop, foo(loop), 5)
loop.close()

It works as expected (after 3 seconds sleep 3 completes successfully and 0 is printed). However, if I run it with my own event loop:

loop = SelectorEventLoop()
run_timeout(loop, foo(loop), 5)
loop.close()

I get a TimeoutError (from the wait_for in run_timeout):

Traceback (most recent call last):
  File "test.py", line 15, in <module>
    _run_async(loop, foo(loop), 5)
  File "test.py", line 7, in _run_async
    return loop.run_until_complete(timed_coroutine)
  File "/usr/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
    return future.result()
  File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/lib/python3.5/asyncio/tasks.py", line 239, in _step
    result = coro.send(None)
  File "/usr/lib/python3.5/asyncio/tasks.py", line 396, in wait_for
    raise futures.TimeoutError()
concurrent.futures._base.TimeoutError

The only way I can get my custom event loop to work is if I set_event_loop() after creating my own SelectorEventLoop:

loop = SelectorEventLoop()
set_event_loop(loop)
run_timeout(loop, foo(loop), 3)
loop.close()

What gives here? Am I misunderstanding the docs? Must all event loops (that you use) be made the default one? If so, it seems useless to allow custom loops to be passed into many of the async methods (eg. create_subprocess_exec and wait_for), because the only value you could pass in is get_event_loop(), which is the default.

Answer 1

It's really strange. I debugged the program and found that is hard to say if it is a bug.

Let's make a long story short, when executing create_subprocess_exec, you need not only an event loop but also a child watcher(which is used to monitor child processes). But create_subprocess_exec doesn't provide a way to let you set custom child watcher, it just use the default watcher which attaches to the default event loop but not current running event loop.

If you use the following code, it will work:

from asyncio import SelectorEventLoop, create_subprocess_exec, \
    wait_for, get_event_loop, set_event_loop, get_child_watcher


def run_timeout(loop, awaitable, timeout):
    timed_awaitable = wait_for(awaitable, timeout=timeout)
    return loop.run_until_complete(timed_awaitable)

async def foo():
    process = await create_subprocess_exec('sleep', '3')
    await process.wait()
    print(process.returncode)

loop = SelectorEventLoop()
# core line, get default child watcher and attach it to your custom loop.
get_child_watcher().attach_loop(loop)
run_timeout(loop, foo(), 5)
loop.close()

And if you use set_event_loop to set default loop, it will also reattach the default child watcher to new default loop. That's why it works.

---

It's really hard to say if it is a bug or a problem about API design. Should create_subprocess_exec let you pass a custom watcher? If it should, it will make confusion as you will only touch child watcher when you are working with child processes.