How to use asyncio event loop in library function

Question

I'm trying to create a function performing some asynchronous operations using asyncio, users of this function should not need to know that asyncio is involved under the hood. I'm having a very hard time understanding how this shall be done with the asyncio API as most functions seem to operate under some global loop-variable accessed with get_event_loop and calls to this are affected by the global state inside this loop.

I have four examples here where two (foo1 and foo3) seem to be reasonable use cases but they all show very strange behaviors:

async def bar(loop):
    # Disregard how simple this is, it's just for example
    s = await asyncio.create_subprocess_exec("ls", loop=loop)


def foo1():
    # Example1: Just use get_event_loop
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000))
    # On exit this is written to stderr:
    #    Exception ignored in: <bound method BaseEventLoop.__del__ of <_UnixSelectorEventLoop running=False closed=True debug=False>>
    #    Traceback (most recent call last):
    #      File "/usr/lib/python3.5/asyncio/base_events.py", line 510, in __del__
    #      File "/usr/lib/python3.5/asyncio/unix_events.py", line 65, in close
    #      File "/usr/lib/python3.5/asyncio/unix_events.py", line 146, in remove_signal_handler
    #      File "/usr/lib/python3.5/signal.py", line 47, in signal
    #    TypeError: signal handler must be signal.SIG_IGN, signal.SIG_DFL, or a callable object


def foo2():
    # Example2: Use get_event_loop and close it when done
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000))  # RuntimeError: Event loop is closed  --- if foo2() is called twice
    loop.close()


def foo3():
    # Example3: Always use new_event_loop
    loop = asyncio.new_event_loop()
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000)) #RuntimeError: Cannot add child handler, the child watcher does not have a loop attached
    loop.close()


def foo4():
    # Example4: Same as foo3 but also set_event_loop to the newly created one
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)        # Polutes global event loop, callers of foo4 does not expect this.
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000))  # OK
    loop.close()

这些功能都不起作用，我没有看到任何其他明显的方法来做到这一点，应该如何使用 asyncio？似乎它只是设计用于假设应用程序的入口点是唯一可以创建和关闭全局循环的地方。我是否必须摆弄事件循环策略？

foo3 似乎是正确的解决方案，但即使我明确地传递循环，我也会收到错误，因为在 create_subprocess_exec 内部深处，它正在使用当前策略来获取一个新的无循环，这是 asyncio 子进程中的错误吗？

我在 Ubuntu 上使用 Python 3.5.3。

Answer 1

发生foo1错误是因为您没有关闭事件循环，请参阅此问题。

foo2因为你不能重用封闭的事件循环。

foo3因为您没有将新事件循环设置为全局。

foo4几乎就是你想要的，你剩下要做的就是存储旧的事件循环并在 bar 执行后将其设置回全局：

import asyncio


async def bar():
    # After you set new event loop global,
    # there's no need to pass loop as param to bar or anywhere else.
    process = await asyncio.create_subprocess_exec("ls")
    await process.communicate()


def sync_exec(coro):  # foo5
    old_loop = asyncio.get_event_loop()
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(coro)
    finally:
        loop.close()
        asyncio.set_event_loop(old_loop)


sync_exec(asyncio.wait_for(bar(), 1000))

更重要的一件事：尚不清楚为什么要在某些同步函数后面隐藏 asyncio 的使用，但通常这是个坏主意。一个全局事件循环的全部内容是允许用户在这个单个事件循环中运行不同的并发作业。你正试图消除这种可能性。我认为你应该重新考虑这个决定。