python asyncio 从 3.4 迁移到 3.5+

Question

大家晚上好，我正在尝试创建互联网机器人，但在将脚本从 python 3.4 迁移到 3.5 或 3.6+ 时遇到了问题。它使用 asyncio 并且在 3.4 python 上运行良好，但是当我使用 python3.5+ 启动它时出现错误：RuntimeError: Cannot run the event loop while another loop is running

这里是代码方案：

import multiprocessing as mp
import asyncio
import concurrent.futures
import aiohttp

def create_proccesses(separate_loop_creator, coro):
    proccesses = []
    for n in range(2):
        proc = mp.Process(target=separate_loop_creator, args=(coro,))
        proc.start()
        proccesses.append(proc)
    for p in proccesses:
        p.join()

def separate_loop_creator(coro):
    sep_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(sep_loop)
    tasks = [asyncio.async(coro(sep_loop)) for _ in range(100)]
    try:
        sep_loop.run_until_complete(asyncio.wait(tasks))
        sep_loop.close()
    except Exception as err:
        print(err)
        for task in tasks:
            task.cancel()
        sep_loop.close()


@asyncio.coroutine
def manager(exe, loop):
    # some calculations and start coros in several processes
    loop.run_in_executor(
        exe,
        create_proccesses,
        separate_loop_creator,
        some_coro
    )

@asyncio.coroutine
def some_work_in_mainloop():
    while True:
        print('Some server dealing with connections here...')
        yield from asyncio.sleep(1)

@asyncio.coroutine
def some_coro(loop):
    with aiohttp.ClientSession(loop=loop) as session:
        response = yield from session.get('http://google.com')
        yield from asyncio.sleep(2)
        print(response.status)

if __name__ == '__main__':
    mainloop = asyncio.get_event_loop()
    executor = concurrent.futures.ProcessPoolExecutor(5)
    asyncio.async(some_work_in_mainloop())
    asyncio.async(manager(executor, mainloop))
    try:
        mainloop.run_forever()
    finally:
        mainloop.close()

separate_loop_creator() 协程中引发的异常是RuntimeError: Cannot run the event loop while another loop is running。我认为这是因为更改了get_event_loop() mechnics，但我不明白我的代码有什么问题。

这是我想做的：

                       +--------------+
               +-------+other service |
    +----------+       +--------------+
    | mainloop |
    +----------+
               |     +------------+
               +-----+   executor |
                     +------+-----+
                            |
                     +------+--------+
                     |start proccess |
                     +---+-------+---+
+-----------------+      |       |      +---------------+
|start new loop   +------+       +------+ start new loop|
+--------+--------+                     +-------+-------+
         |                                      |
 +-------+-------+                       +------v-------+
 |   run coro    |                       | run coro     |
 +---------------+                       +--------------+

这是我在 python3.5.3 上得到的跟踪：

Traceback (most recent call last):
  File "tst.py", line 21, in separate_loop_creator
    sep_loop.run_until_complete(asyncio.wait(tasks))
  File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 454, in run_until_complete
    self.run_forever()
  File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 411, in run_forever
    'Cannot run the event loop while another loop is running')
RuntimeError: Cannot run the event loop while another loop is running
Cannot run the event loop while another loop is running
Traceback (most recent call last):
  File "tst.py", line 21, in separate_loop_creator
    sep_loop.run_until_complete(asyncio.wait(tasks))
  File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 454, in run_until_complete
    self.run_forever()
  File "/root/.pyenv/versions/3.5.3/lib/python3.5/asyncio/base_events.py", line 411, in run_forever
    'Cannot run the event loop while another loop is running')
RuntimeError: Cannot run the event loop while another loop is running

Python 3.4.3 结果：

...
200
Some server dealing with connections here...
200
200
Some server dealing with connections here...
200
200
Some server dealing with connections here...
200
...

Answer 1

这实际上是 CPython 3.6.0 中 asyncio 中的一个错误。有一个 PR 来解决这个问题，因此 3.6.1 可以按预期工作。

作为一种解决方法，您可以在项目中添加以下代码：

import sys

if sys.version_info[:3] == (3, 6, 0):
    import asyncio.events as _ae
    import os as _os

    _ae._RunningLoop._pid = None

    def _get_running_loop():
        if _ae._running_loop._pid == _os.getpid():
            return _ae._running_loop._loop

    def _set_running_loop(loop):
        _ae._running_loop._pid = _os.getpid()
        _ae._running_loop._loop = loop

    _ae._get_running_loop = _get_running_loop
    _ae._set_running_loop = _set_running_loop

Answer 2

如果可能的话，最好的解决方案是尝试从您的程序中完全删除 multiprocessing，并且只使用一个事件循环（对于独立的 CPU 密集型任务，可以选择使用 ProcessPoolExecutor）。

截至 2017 年 3 月 2 日，存在一个针对此问题的开放 python 错误，影响非 Windows 平台：https://bugs.python.org/issue22087。

这是一个触发相同问题的较短程序：

import asyncio
import multiprocessing as mp


def create_another_loop():
    loop = asyncio.new_event_loop()
    loop.run_forever()


async def create_process():
    proc = mp.Process(target=create_another_loop)
    proc.start()
    proc.join()


if __name__ == '__main__':
    main_loop = asyncio.get_event_loop()
    main_loop.run_until_complete(create_process())
    main_loop.close()

受https://github.com/python/asyncio/pull/497 此处建议的修复启发的一种骇人听闻的解决方法（注意！使用风险自负！）是将此代码添加到新创建的Process：

if asyncio.events._running_loop:
    asyncio.events._running_loop._loop = None

例子：

import asyncio
import multiprocessing as mp
import time
from concurrent.futures.process import ProcessPoolExecutor


async def clock(label, n=5, sleep=1):
    print(label, "start")
    for i in range(n):
        await asyncio.sleep(sleep)
        print(label, i + 1)
    print(label, "end")
    return label


def create_another_loop():
    # HACK START
    if asyncio.events._running_loop:
        asyncio.events._running_loop._loop = None
    # HACK END

    loop = asyncio.new_event_loop()
    loop.run_until_complete(clock("sub"))
    loop.close()


def create_process():
    time.sleep(2)
    proc = mp.Process(target=create_another_loop)
    proc.start()
    proc.join()
    return "ok"


async def create_process_in_pool():
    return await main_loop.run_in_executor(ProcessPoolExecutor(), create_process)


if __name__ == '__main__':
    main_loop = asyncio.get_event_loop()
    tasks = (
        clock("main"),
        create_process_in_pool(),
    )
    print(main_loop.run_until_complete(asyncio.gather(*tasks)))
    main_loop.close()

其他可能的解决方法：在开始循环之前创建进程或使用asyncio.create_subprocess_exec，它甚至允许communicate with the subprocess via a stream。